improve enum typing and tests

Author: OkGoDoIt

README.md

@@ -24,7 +24,8 @@ Here's a simple example of how to use the Frame SDK to display text, take a phot
 ```python
 import asyncio
 from frame_sdk import Frame
-from frame_sdk.display import Alignment
+from frame_sdk.display import Alignment, PaletteColors
+from frame_sdk.camera import Quality, AutofocusType
 import datetime
 
 async def main():
@@ -59,9 +60,9 @@ async def main():
         # take a photo and save to disk
         await f.display.show_text("Taking photo...", align=Alignment.MIDDLE_CENTER)
         await f.camera.save_photo("frame-test-photo.jpg")
-        await f.display.show_text("Photo saved!", align=Alignment.MIDDLE_CENTER)
+        await f.display.show_text("Photo saved!", align=Alignment.MIDDLE_CENTER, color=PaletteColors.GREEN)
         # or with more control
-        await f.camera.save_photo("frame-test-photo-2.jpg", autofocus_seconds=3, quality=f.camera.HIGH_QUALITY, autofocus_type=f.camera.AUTOFOCUS_TYPE_CENTER_WEIGHTED)
+        await f.camera.save_photo("frame-test-photo-2.jpg", autofocus_seconds=3, quality=Quality.HIGH, autofocus_type=AutofocusType.CENTER_WEIGHTED)
         # or get the raw bytes
         photo_bytes = await f.camera.take_photo(autofocus_seconds=1)
 
@@ -78,7 +79,7 @@ async def main():
         audio_data = await f.microphone.record_audio(max_length_in_seconds=10)
         await f.display.show_text(f"Playing back {len(audio_data) / f.microphone.sample_rate:01.1f} seconds of audio", align=Alignment.MIDDLE_CENTER)
         # you can play back the audio on your computer
-        f.microphone.play_audio(audio_data)
+        f.microphone.play_audio_background(audio_data)
         # or process it using other audio handling libraries, upload to a speech-to-text service, etc.
 
         print("Move around to track intensity of your motion")
@@ -101,7 +102,7 @@ async def main():
         for color in range(0, 16):
             tile_x = (color % 4)
             tile_y = (color // 4)
-            await f.display.draw_rect(tile_x*width+1, tile_y*height+1, width, height, color)
+            await f.display.draw_rect(tile_x*width+1, tile_y*height+1, width, height, PaletteColors(color))
             await f.display.write_text(f"{color}", tile_x*width+width//2+1, tile_y*height+height//2+1)
         await f.display.show()
 
@@ -114,14 +115,14 @@ async def main():
         # display battery indicator and time as a home screen
         batteryPercent = await f.get_battery_level()
         # select a battery fill color from the default palette based on level
-        color = 2 if batteryPercent < 20 else 6 if batteryPercent < 50 else 9
+        color = PaletteColors.RED if batteryPercent < 20 else PaletteColors.YELLOW if batteryPercent < 50 else PaletteColors.GREEN
         # specify the size of the battery indicator in the top-right
         batteryWidth = 150
         batteryHeight = 75
         # draw the endcap of the battery
-        await f.display.draw_rect(640-32,40 + batteryHeight//2-8, 32, 16, 1)
+        await f.display.draw_rect(640-32,40 + batteryHeight//2-8, 32, 16, PaletteColors.WHITE)
         # draw the battery outline
-        await f.display.draw_rect_filled(640-16-batteryWidth, 40-8, batteryWidth+16, batteryHeight+16, 8, 1, 15)
+        await f.display.draw_rect_filled(640-16-batteryWidth, 40-8, batteryWidth+16, batteryHeight+16, PaletteColors.WHITE, 1, 15)
         # fill the battery based on level
         await f.display.draw_rect(640-8-batteryWidth, 40, int(batteryWidth * 0.01 * batteryPercent), batteryHeight, color)
         # write the battery level
@@ -167,4 +168,6 @@ With a Frame device in range, run:
 
 ```sh
 python3 -m pytest tests/*
-```
+```
+
+Note that one of the audio playback tests fails on Windows.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "frame-sdk"
-version = "1.1.0"
+version = "1.2.0"
 authors = [{ name = "Roger Pincombe", email = "pip@betechie.com" },{ name = "Brilliant Labs", email = "info@brilliant.xyz" }]
 description = "Python Developer SDK for Brilliant Frame glasses"
 readme = "readme.md"

src/frame_sdk/bluetooth.py

@@ -1,16 +1,24 @@
 import asyncio
 from typing import Optional, Callable, List, Tuple, Dict, Any
-
+from enum import Enum
 from bleak import BleakClient, BleakScanner, BleakError
 
 _FRAME_DATA_PREFIX = 1
-_FRAME_LONG_TEXT_PREFIX = 10
-_FRAME_LONG_TEXT_END_PREFIX = 11
-_FRAME_LONG_DATA_PREFIX = 1
-_FRAME_LONG_DATA_END_PREFIX = 2
 
-_FRAME_TAP_PREFIX = b'\x04'
-_FRAME_MIC_DATA_PREFIX = b'\x05'
+class FrameDataTypePrefixes(Enum):
+    LONG_DATA = 0x01
+    LONG_DATA_END = 0x02
+    WAKE = 0x03
+    TAP = 0x04
+    MIC_DATA = 0x05
+    DEBUG_PRINT = 0x06
+    LONG_TEXT = 0x0A
+    LONG_TEXT_END = 0x0B
+
+    @property
+    def value_as_hex(self):
+        return f'{self.value:02x}'
+
 
 class Bluetooth:
     """
@@ -41,7 +49,7 @@ def __init__(self):
         self._ongoing_data_response: Optional[bytearray] = None
         self._ongoing_data_response_chunk_count: Optional[int] = None
         self._data_response_event: asyncio.Event = asyncio.Event()
-        self._user_data_response_handlers: Dict[bytes, Callable[[bytes], None]] = {}
+        self._user_data_response_handlers: Dict[FrameDataTypePrefixes, Callable[[bytes], None]] = {}
 
 
     def _disconnect_handler(self, _: Any) -> None:
@@ -57,7 +65,7 @@ async def _notification_handler(self, _: Any, data: bytearray) -> None:
         Args:
             data (bytearray): The data received from the device as raw bytes
         """
-        if data[0] == _FRAME_LONG_TEXT_PREFIX:
+        if data[0] == FrameDataTypePrefixes.LONG_TEXT.value:
             # start of long printed data from prntLng() function
             if self._ongoing_print_response is None or self._ongoing_print_response_chunk_count is None:
                 self._ongoing_print_response = bytearray()
@@ -71,7 +79,7 @@ async def _notification_handler(self, _: Any, data: bytearray) -> None:
             if len(self._ongoing_print_response) > self._max_receive_buffer:
                 raise Exception(f"Buffered received long printed string is more than {self._max_receive_buffer} bytes")
 
-        elif data[0] == _FRAME_LONG_TEXT_END_PREFIX:
+        elif data[0] == FrameDataTypePrefixes.LONG_TEXT_END.value:
             # end of long printed data from prntLng() function
             total_expected_chunk_count_as_string: str = data[1:].decode()
             if len(total_expected_chunk_count_as_string) > 0:
@@ -88,7 +96,7 @@ async def _notification_handler(self, _: Any, data: bytearray) -> None:
                 print("Finished receiving long printed string: "+self._last_print_response)
             self._user_print_response_handler(self._last_print_response)
 
-        elif data[0] == _FRAME_DATA_PREFIX and data[1] == _FRAME_LONG_DATA_PREFIX:
+        elif data[0] == _FRAME_DATA_PREFIX and data[1] == FrameDataTypePrefixes.LONG_DATA.value:
             # start of long raw data from frame.bluetooth.send("\001"..data)
             if self._ongoing_data_response is None or self._ongoing_data_response_chunk_count is None:
                 self._ongoing_data_response = bytearray()
@@ -103,7 +111,7 @@ async def _notification_handler(self, _: Any, data: bytearray) -> None:
             if len(self._ongoing_data_response) > self._max_receive_buffer:
                 raise Exception(f"Buffered received long raw data is more than {self._max_receive_buffer} bytes")
 
-        elif data[0] == _FRAME_DATA_PREFIX and data[1] == _FRAME_LONG_DATA_END_PREFIX:
+        elif data[0] == _FRAME_DATA_PREFIX and data[1] == FrameDataTypePrefixes.LONG_DATA_END.value:
             # end of long raw data from frame.bluetooth.send("\002"..chunkCount)
             total_expected_chunk_count_as_string: str = data[2:].decode()
             if len(total_expected_chunk_count_as_string) > 0:
@@ -139,7 +147,7 @@ async def _notification_handler(self, _: Any, data: bytearray) -> None:
             self._print_response_event.set()
             self._user_print_response_handler(data.decode())
 
-    def register_data_response_handler(self, prefix: bytes = None, handler: Callable[[bytes], None] = None) -> None:
+    def register_data_response_handler(self, prefix: FrameDataTypePrefixes = None, handler: Callable[[bytes], None] = None) -> None:
         """Registers a data response handler which will be called when data is received from the device that starts with the specified prefix."""
         if handler is None:
             self._user_data_response_handlers.pop(prefix, None)
@@ -152,9 +160,9 @@ def register_data_response_handler(self, prefix: bytes = None, handler: Callable
     def call_data_response_handlers(self, data: bytes) -> None:
         """Calls all data response handlers which match the received data."""
         for prefix, handler in self._user_data_response_handlers.items():
-            if prefix is None or data.startswith(prefix):
+            if prefix is None or (len(data) > 0 and data[0] == prefix.value):
                 if handler is not None:
-                    handler(data[len(prefix):])
+                    handler(data[1:])
 
     @property
     def print_response_handler(self) -> Callable[[str], None]:

src/frame_sdk/camera.py

@@ -6,19 +6,24 @@
 if TYPE_CHECKING:
     from .frame import Frame
 
+from enum import Enum
+
+class Quality(Enum):
+    LOW = 10
+    MEDIUM = 25
+    HIGH = 50
+    FULL = 100
+
+class AutofocusType(Enum):
+    SPOT = "SPOT"
+    AVERAGE = "AVERAGE"
+    CENTER_WEIGHTED = "CENTER_WEIGHTED"
+
+
 class Camera:
     """Helpers for working with the Frame camera."""
 
     frame: "Frame" = None
-    
-    LOW_QUALITY = 10
-    MEDIUM_QUALITY = 25
-    HIGH_QUALITY = 50
-    FULL_QUALITY = 100
-    
-    AUTOFOCUS_TYPE_SPOT = "SPOT"
-    AUTOFOCUS_TYPE_AVERAGE = "AVERAGE"
-    AUTOFOCUS_TYPE_CENTER_WEIGHTED = "CENTER_WEIGHTED"
 
     _auto_process_photo = True
 
@@ -38,13 +43,13 @@ def auto_process_photo(self, value: bool):
         self._auto_process_photo = value
 
 
-    async def take_photo(self, autofocus_seconds: Optional[int] = 3, quality: int = MEDIUM_QUALITY, autofocus_type: str = AUTOFOCUS_TYPE_AVERAGE) -> bytes:
+    async def take_photo(self, autofocus_seconds: Optional[int] = 3, quality: Quality = Quality.MEDIUM, autofocus_type: AutofocusType = AutofocusType.AVERAGE) -> bytes:
         """Take a photo with the camera.
 
         Args:
             autofocus_seconds (Optional[int]): If provided, the camera will attempt to focus for the specified number of seconds.  Defaults to 3.  If `None`, the camera will not attempt to focus at all.
-            quality (int): The quality of the photo. Defaults to MEDIUM_QUALITY.  May be one of LOW_QUALITY (10), MEDIUM_QUALITY (25), HIGH_QUALITY (50), or FULL_QUALITY (100).
-            autofocus_type (str): The type of autofocus. Defaults to AUTOFOCUS_TYPE_AVERAGE.  May be one of AUTOFOCUS_TYPE_SPOT, AUTOFOCUS_TYPE_AVERAGE, or AUTOFOCUS_TYPE_CENTER_WEIGHTED.
+            quality (Quality): The quality of the photo. Defaults to Quality.MEDIUM.
+            autofocus_type (AutofocusType): The type of autofocus. Defaults to AutofocusType.AVERAGE.
         
         Returns:
             bytes: The photo as a byte array.
@@ -57,7 +62,12 @@ async def take_photo(self, autofocus_seconds: Optional[int] = 3, quality: int =
             await self.frame.run_lua("frame.camera.wake()", checked=True)
             self.is_awake = True
 
-        await self.frame.bluetooth.send_lua(f"cameraCaptureAndSend({quality},{autofocus_seconds or 'nil'},{autofocus_type})")
+        if type(quality) == int:
+            quality = Quality(quality)
+        if type(autofocus_type) == int:
+            autofocus_type = AutofocusType(autofocus_type)
+
+        await self.frame.bluetooth.send_lua(f"cameraCaptureAndSend({quality.value},{autofocus_seconds or 'nil'},'{autofocus_type.value}')")
         image_buffer = await self.frame.bluetooth.wait_for_data()
 
         if image_buffer is None or len(image_buffer) == 0:
@@ -67,26 +77,26 @@ async def take_photo(self, autofocus_seconds: Optional[int] = 3, quality: int =
             image_buffer = self.process_photo(image_buffer, autofocus_type)
         return image_buffer
 
-    async def save_photo(self, filename: str, autofocus_seconds: Optional[int] = 3, quality: int = MEDIUM_QUALITY, autofocus_type: str = AUTOFOCUS_TYPE_AVERAGE):
+    async def save_photo(self, filename: str, autofocus_seconds: Optional[int] = 3, quality: Quality = Quality.MEDIUM, autofocus_type: AutofocusType = AutofocusType.AVERAGE):
         """Save a photo to a file.
         
         Args:
             filename (str): The name of the file to save the photo.  The file will always be saved as a jpeg image regardless of the file extension.
             autofocus_seconds (Optional[int]): If provided, the camera will attempt to focus for the specified number of seconds.  Defaults to 3.  If `None`, the camera will not attempt to focus at all.
-            quality (int): The quality of the photo. Defaults to MEDIUM_QUALITY.  May be one of LOW_QUALITY (10), MEDIUM_QUALITY (25), HIGH_QUALITY (50), or FULL_QUALITY (100).
-            autofocus_type (str): The type of autofocus. Defaults to AUTOFOCUS_TYPE_AVERAGE.  May be one of AUTOFOCUS_TYPE_SPOT, AUTOFOCUS_TYPE_AVERAGE, or AUTOFOCUS_TYPE_CENTER_WEIGHTED.
+            quality (Quality): The quality of the photo. Defaults to Quality.MEDIUM.
+            autofocus_type (AutofocusType): The type of autofocus. Defaults to AutofocusType.AVERAGE.
         """
         image_buffer = await self.take_photo(autofocus_seconds, quality, autofocus_type)
 
         with open(filename, "wb") as f:
             f.write(image_buffer)
 
-    def process_photo(self, image_buffer: bytes, autofocus_type: str) -> bytes:
+    def process_photo(self, image_buffer: bytes, autofocus_type: AutofocusType) -> bytes:
         """Process a photo to correct rotation and add metadata.
         
         Args:
             image_buffer (bytes): The photo as a byte array.
-            autofocus_type (str): The type of autofocus that was used to capture the photo.  Should be one of AUTOFOCUS_TYPE_SPOT, AUTOFOCUS_TYPE_AVERAGE, or AUTOFOCUS_TYPE_CENTER_WEIGHTED.
+            autofocus_type (AutofocusType): The type of autofocus that was used to capture the photo.
         
         Returns:
             bytes: The processed photo as a byte array.
@@ -96,11 +106,11 @@ def process_photo(self, image_buffer: bytes, autofocus_type: str) -> bytes:
         image.make = "Brilliant Labs"
         image.model = "Frame"
         image.software = "Frame Python SDK"
-        if autofocus_type == self.AUTOFOCUS_TYPE_AVERAGE:
+        if autofocus_type == AutofocusType.AVERAGE:
             image.metering_mode = 1
-        elif autofocus_type == self.AUTOFOCUS_TYPE_CENTER_WEIGHTED:
+        elif autofocus_type == AutofocusType.CENTER_WEIGHTED:
             image.metering_mode = 2
-        elif autofocus_type == self.AUTOFOCUS_TYPE_SPOT:
+        elif autofocus_type == AutofocusType.SPOT:
             image.metering_mode = 3
         image.datetime_original = datetime.now().strftime("%Y:%m:%d %H:%M:%S")
         return image.get_file()