Port to latest bleak and test each tutorial

demos/python/tutorial/tutorial_modules/__init__.py

@@ -3,8 +3,11 @@
 
 # pylint: disable=wrong-import-position
 
+from typing import Callable
 import logging
 
+from bleak.backends.characteristic import BleakGATTCharacteristic
+
 from rich.logging import RichHandler
 from rich import traceback
 
@@ -25,6 +28,8 @@
 GOPRO_BASE_UUID = "b5f9{}-aa8d-11e3-9046-0002a5d5c51b"
 GOPRO_BASE_URL = "http://10.5.5.9:8080"
 
+noti_handler_T = Callable[[BleakGATTCharacteristic, bytearray], None]
+
 from tutorial_modules.tutorial_1_connect_ble.ble_connect import connect_ble
 from tutorial_modules.tutorial_3_parse_ble_tlv_responses.ble_command_get_state import Response
 from tutorial_modules.tutorial_5_connect_wifi.wifi_enable import enable_wifi

demos/python/tutorial/tutorial_modules/tutorial_1_connect_ble/ble_connect.py

@@ -5,12 +5,12 @@
 import sys
 import asyncio
 import argparse
-from typing import Dict, Any, List, Callable, Optional
+from typing import Dict, Any, List, Optional
 
 from bleak import BleakScanner, BleakClient
 from bleak.backends.device import BLEDevice as BleakDevice
 
-from tutorial_modules import logger
+from tutorial_modules import logger, noti_handler_T
 
 
 def exception_handler(loop: asyncio.AbstractEventLoop, context: Dict[str, Any]) -> None:
@@ -25,18 +25,15 @@ def exception_handler(loop: asyncio.AbstractEventLoop, context: Dict[str, Any])
     logger.critical("This is unexpected and unrecoverable.")
 
 
-async def connect_ble(
-    notification_handler: Callable[[int, bytes], None],
-    identifier: Optional[str] = None,
-) -> BleakClient:
+async def connect_ble(notification_handler: noti_handler_T, identifier: Optional[str] = None) -> BleakClient:
     """Connect to a GoPro, then pair, and enable notifications
 
     If identifier is None, the first discovered GoPro will be connected to.
 
     Retry 10 times
 
     Args:
-        notification_handler (Callable[[int, bytes], None]): callback when notification is received
+        notification_handler (noti_handler_T): callback when notification is received
         identifier (str, optional): Last 4 digits of GoPro serial number. Defaults to None.
 
     Raises:
@@ -102,7 +99,7 @@ def _scan_callback(device: BleakDevice, _: Any) -> None:
                 for char in service.characteristics:
                     if "notify" in char.properties:
                         logger.info(f"Enabling notification on char {char.uuid}")
-                        await client.start_notify(char, notification_handler)  # type: ignore
+                        await client.start_notify(char, notification_handler)
             logger.info("Done enabling notifications")
 
             return client

demos/python/tutorial/tutorial_modules/tutorial_2_send_ble_commands/ble_command_load_group.py

@@ -5,11 +5,11 @@
 import asyncio
 import argparse
 from typing import Optional
-from binascii import hexlify
 
 from bleak import BleakClient
+from bleak.backends.characteristic import BleakGATTCharacteristic
 
-from tutorial_modules import GOPRO_BASE_UUID, connect_ble, logger
+from tutorial_modules import GOPRO_BASE_UUID, connect_ble, logger, noti_handler_T
 
 
 async def main(identifier: Optional[str]) -> None:
@@ -23,11 +23,11 @@ async def main(identifier: Optional[str]) -> None:
 
     client: BleakClient
 
-    def notification_handler(handle: int, data: bytes) -> None:
-        logger.info(f'Received response at {handle=}: {hexlify(data, ":")!r}')
+    def notification_handler(characteristic: BleakGATTCharacteristic, data: bytes) -> None:
+        logger.info(f'Received response at handle {characteristic.handle}: {data.hex(":")}')
 
         # If this is the correct handle and the status is success, the command was a success
-        if client.services.characteristics[handle].uuid == response_uuid and data[2] == 0x00:
+        if client.services.characteristics[characteristic.handle].uuid == response_uuid and data[2] == 0x00:
             logger.info("Command sent successfully")
         # Anything else is unexpected. This shouldn't happen
         else:

demos/python/tutorial/tutorial_modules/tutorial_2_send_ble_commands/ble_command_set_fps.py

@@ -8,6 +8,7 @@
 from binascii import hexlify
 
 from bleak import BleakClient
+from bleak.backends.characteristic import BleakGATTCharacteristic
 
 from tutorial_modules import GOPRO_BASE_UUID, connect_ble, logger
 
@@ -23,11 +24,11 @@ async def main(identifier: Optional[str]) -> None:
 
     client: BleakClient
 
-    def notification_handler(handle: int, data: bytes) -> None:
-        logger.info(f'Received response at {handle=}: {hexlify(data, ":")!r}')
+    def notification_handler(characteristic: BleakGATTCharacteristic, data: bytes) -> None:
+        logger.info(f'Received response at handle {characteristic.handle}: {data.hex(":")}')
 
         # If this is the correct handle and the status is success, the command was a success
-        if client.services.characteristics[handle].uuid == response_uuid and data[2] == 0x00:
+        if client.services.characteristics[characteristic.handle].uuid == response_uuid and data[2] == 0x00:
             logger.info("Command sent successfully")
         # Anything else is unexpected. This shouldn't happen
         else:

demos/python/tutorial/tutorial_modules/tutorial_2_send_ble_commands/ble_command_set_resolution.py

@@ -8,6 +8,7 @@
 from binascii import hexlify
 
 from bleak import BleakClient
+from bleak.backends.characteristic import BleakGATTCharacteristic
 
 from tutorial_modules import GOPRO_BASE_UUID, connect_ble, logger
 
@@ -23,11 +24,11 @@ async def main(identifier: Optional[str]) -> None:
 
     client: BleakClient
 
-    def notification_handler(handle: int, data: bytes) -> None:
-        logger.info(f'Received response at {handle=}: {hexlify(data, ":")!r}')
+    def notification_handler(characteristic: BleakGATTCharacteristic, data: bytes) -> None:
+        logger.info(f'Received response at handle {characteristic.handle}: {data.hex(":")}')
 
         # If this is the correct handle and the status is success, the command was a success
-        if client.services.characteristics[handle].uuid == response_uuid and data[2] == 0x00:
+        if client.services.characteristics[characteristic.handle].uuid == response_uuid and data[2] == 0x00:
             logger.info("Command sent successfully")
         # Anything else is unexpected. This shouldn't happen
         else:

demos/python/tutorial/tutorial_modules/tutorial_2_send_ble_commands/ble_command_set_shutter.py

@@ -6,9 +6,9 @@
 import asyncio
 import argparse
 from typing import Optional
-from binascii import hexlify
 
 from bleak import BleakClient
+from bleak.backends.characteristic import BleakGATTCharacteristic
 
 from tutorial_modules import GOPRO_BASE_UUID, connect_ble, logger
 
@@ -24,11 +24,11 @@ async def main(identifier: Optional[str]) -> None:
 
     client: BleakClient
 
-    def notification_handler(handle: int, data: bytes) -> None:
-        logger.info(f'Received response at {handle}: {data.hex(":")}')
+    def notification_handler(characteristic: BleakGATTCharacteristic, data: bytes) -> None:
+        logger.info(f'Received response at handle {characteristic.handle}: {data.hex(":")}')
 
         # If this is the correct handle and the status is success, the command was a success
-        if client.services.characteristics[handle].uuid == response_uuid and data[2] == 0x00:
+        if client.services.characteristics[characteristic.handle].uuid == response_uuid and data[2] == 0x00:
             logger.info("Command sent successfully")
         # Anything else is unexpected. This shouldn't happen
         else:

demos/python/tutorial/tutorial_modules/tutorial_3_parse_ble_tlv_responses/ble_command_get_state.py

@@ -10,6 +10,7 @@
 from typing import Dict, Optional
 
 from bleak import BleakClient
+from bleak.backends.characteristic import BleakGATTCharacteristic
 
 from tutorial_modules import GOPRO_BASE_UUID, connect_ble, logger
 
@@ -94,16 +95,19 @@ async def main(identifier: Optional[str]) -> None:
     client: BleakClient
     response = Response()
 
-    def notification_handler(handle: int, data: bytes) -> None:
-        logger.info(f'Received response at {handle=}: {hexlify(data, ":")!r}')
+    def notification_handler(characteristic: BleakGATTCharacteristic, data: bytes) -> None:
+        logger.info(f'Received response at handle {characteristic.handle}: {data.hex(":")}')
 
         response.accumulate(data)
 
         if response.is_received:
             response.parse()
 
             # If this is the correct handle and the status is success, the command was a success
-            if client.services.characteristics[handle].uuid == response_uuid and response.status == 0:
+            if (
+                client.services.characteristics[characteristic.handle].uuid == response_uuid
+                and response.status == 0
+            ):
                 logger.info("Successfully received the response")
             # Anything else is unexpected. This shouldn't happen
             else:

demos/python/tutorial/tutorial_modules/tutorial_3_parse_ble_tlv_responses/ble_command_get_version.py

@@ -4,10 +4,10 @@
 import sys
 import asyncio
 import argparse
-from binascii import hexlify
 from typing import Dict, Optional
 
 from bleak import BleakClient
+from bleak.backends.characteristic import BleakGATTCharacteristic
 
 from tutorial_modules import GOPRO_BASE_UUID, connect_ble, logger
 
@@ -23,11 +23,11 @@ async def main(identifier: Optional[str]) -> None:
 
     client: BleakClient
 
-    def notification_handler(handle: int, data: bytes) -> None:
-        logger.info(f'Received response at {handle=}: {hexlify(data, ":")!r}')
+    def notification_handler(characteristic: BleakGATTCharacteristic, data: bytes) -> None:
+        logger.info(f'Received response at handle {characteristic.handle}: {data.hex(":")}')
 
         # If this is the correct handle and the status is success, the command was a success
-        if client.services.characteristics[handle].uuid == response_uuid:
+        if client.services.characteristics[characteristic.handle].uuid == response_uuid:
             # First byte is the length for this command.
             length = data[0]
             # Second byte is the ID

demos/python/tutorial/tutorial_modules/tutorial_4_ble_queries/ble_query_poll_multiple_setting_values.py

@@ -6,9 +6,9 @@
 import asyncio
 import argparse
 from typing import Optional
-from binascii import hexlify
 
 from bleak import BleakClient
+from bleak.backends.characteristic import BleakGATTCharacteristic
 
 from tutorial_modules import GOPRO_BASE_UUID, connect_ble, Response
 
@@ -70,8 +70,8 @@ async def main(identifier: Optional[str]) -> None:
     client: BleakClient
     response = Response()
 
-    def notification_handler(handle: int, data: bytes) -> None:
-        logger.info(f'Received response at {handle=}: {hexlify(data, ":")!r}')
+    def notification_handler(characteristic: BleakGATTCharacteristic, data: bytes) -> None:
+        logger.info(f'Received response at handle {characteristic.handle}: {data.hex(":")}')
 
         response.accumulate(data)
 
@@ -80,15 +80,15 @@ def notification_handler(handle: int, data: bytes) -> None:
             response.parse()
 
             # If this is query response, it must contain a resolution value
-            if client.services.characteristics[handle].uuid == QUERY_RSP_UUID:
+            if client.services.characteristics[characteristic.handle].uuid == QUERY_RSP_UUID:
                 global resolution
                 global fps
                 global video_fov
                 resolution = Resolution(response.data[RESOLUTION_ID][0])
                 fps = FPS(response.data[FPS_ID][0])
                 video_fov = VideoFOV(response.data[FOV_ID][0])
             # If this is a setting response, it will just show the status
-            elif client.services.characteristics[handle].uuid == SETTINGS_RSP_UUID:
+            elif client.services.characteristics[characteristic.handle].uuid == SETTINGS_RSP_UUID:
                 logger.info("Command sent successfully")
             # Anything else is unexpected. This shouldn't happen
             else:

demos/python/tutorial/tutorial_modules/tutorial_4_ble_queries/ble_query_poll_resolution_value.py

@@ -6,9 +6,9 @@
 import asyncio
 import argparse
 from typing import Optional
-from binascii import hexlify
 
 from bleak import BleakClient
+from bleak.backends.characteristic import BleakGATTCharacteristic
 
 from tutorial_modules import GOPRO_BASE_UUID, connect_ble, Response
 
@@ -42,8 +42,8 @@ async def main(identifier: Optional[str]) -> None:
     client: BleakClient
     response = Response()
 
-    def notification_handler(handle: int, data: bytes) -> None:
-        logger.info(f'Received response at {handle=}: {hexlify(data, ":")!r}')
+    def notification_handler(characteristic: BleakGATTCharacteristic, data: bytes) -> None:
+        logger.info(f'Received response at handle {characteristic.handle}: {data.hex(":")}')
 
         response.accumulate(data)
 
@@ -52,11 +52,11 @@ def notification_handler(handle: int, data: bytes) -> None:
             response.parse()
 
             # If this is query response, it must contain a resolution value
-            if client.services.characteristics[handle].uuid == QUERY_RSP_UUID:
+            if client.services.characteristics[characteristic.handle].uuid == QUERY_RSP_UUID:
                 global resolution
                 resolution = Resolution(response.data[RESOLUTION_ID][0])
             # If this is a setting response, it will just show the status
-            elif client.services.characteristics[handle].uuid == SETTINGS_RSP_UUID:
+            elif client.services.characteristics[characteristic.handle].uuid == SETTINGS_RSP_UUID:
                 logger.info("Command sent successfully")
             # Anything else is unexpected. This shouldn't happen
             else:

demos/python/tutorial/tutorial_modules/tutorial_4_ble_queries/ble_query_register_resolution_value_updates.py

@@ -2,14 +2,13 @@
 # This copyright was auto-generated on Wed, Sep  1, 2021  5:06:00 PM
 
 import sys
-import time
 import enum
 import asyncio
 import argparse
 from typing import Optional
-from binascii import hexlify
 
 from bleak import BleakClient
+from bleak.backends.characteristic import BleakGATTCharacteristic
 
 from tutorial_modules import GOPRO_BASE_UUID, connect_ble, Response
 
@@ -44,8 +43,8 @@ async def main(identifier: Optional[str]) -> None:
     client: BleakClient
     response = Response()
 
-    def notification_handler(handle: int, data: bytes) -> None:
-        logger.info(f'Received response at {handle=}: {hexlify(data, ":")!r}')
+    def notification_handler(characteristic: BleakGATTCharacteristic, data: bytes) -> None:
+        logger.info(f'Received response at handle {characteristic.handle}: {data.hex(":")}')
 
         response.accumulate(data)
 
@@ -54,11 +53,11 @@ def notification_handler(handle: int, data: bytes) -> None:
             response.parse()
 
             # If this is query response, it must contain a resolution value
-            if client.services.characteristics[handle].uuid == QUERY_RSP_UUID:
+            if client.services.characteristics[characteristic.handle].uuid == QUERY_RSP_UUID:
                 global resolution
                 resolution = Resolution(response.data[RESOLUTION_ID][0])
             # If this is a setting response, it will just show the status
-            elif client.services.characteristics[handle].uuid == SETTINGS_RSP_UUID:
+            elif client.services.characteristics[characteristic.handle].uuid == SETTINGS_RSP_UUID:
                 logger.info("Command sent successfully")
             # Anything else is unexpected. This shouldn't happen
             else:

demos/python/tutorial/tutorial_modules/tutorial_5_connect_wifi/wifi_enable.py

@@ -5,10 +5,10 @@
 import time
 import asyncio
 import argparse
-from binascii import hexlify
 from typing import Tuple, Optional
 
 from bleak import BleakClient
+from bleak.backends.characteristic import BleakGATTCharacteristic
 
 from tutorial_modules import GOPRO_BASE_UUID, connect_ble, logger
 
@@ -35,11 +35,11 @@ async def enable_wifi(identifier: Optional[str] = None) -> Tuple[str, str, Bleak
 
     client: BleakClient
 
-    def notification_handler(handle: int, data: bytes) -> None:
-        logger.info(f'Received response at {handle=}: {hexlify(data, ":")!r}')
+    def notification_handler(characteristic: BleakGATTCharacteristic, data: bytes) -> None:
+        logger.info(f'Received response at handle {characteristic.handle}: {data.hex(":")}')
 
         # If this is the correct handle and the status is success, the command was a success
-        if client.services.characteristics[handle].uuid == COMMAND_RSP_UUID and data[2] == 0x00:
+        if client.services.characteristics[characteristic.handle].uuid == COMMAND_RSP_UUID and data[2] == 0x00:
             logger.info("Command sent successfully")
         # Anything else is unexpected. This shouldn't happen
         else: