Typing for bluetooth module

Author: Fantomas42

term_timer/bluetooth/constants.py

@@ -1,36 +1,37 @@
-# ruff: noqa: E222
+# ruff: noqa: E222 E501
+from typing import Final
 
-DEBOUNCE = 0.5
+DEBOUNCE: Final[float] = 0.5
 
-PREFIX = [
+PREFIX: Final[list[str]] = [
     'GAN',
     'MG',
     'AiCube',
     'WCU_MY32',
 ]
 
 # GAN Gen2 protocol BLE service
-GAN_GEN2_SERVICE =                '6e400001-b5a3-f393-e0a9-e50e24dc4179'
-GAN_GEN2_STATE_CHARACTERISTIC =   '28be4cb6-cd67-11e9-a32f-2a2ae2dbcce4'
-GAN_GEN2_COMMAND_CHARACTERISTIC = '28be4a4a-cd67-11e9-a32f-2a2ae2dbcce4'
+GAN_GEN2_SERVICE: Final[str] =                '6e400001-b5a3-f393-e0a9-e50e24dc4179'
+GAN_GEN2_STATE_CHARACTERISTIC: Final[str] =   '28be4cb6-cd67-11e9-a32f-2a2ae2dbcce4'
+GAN_GEN2_COMMAND_CHARACTERISTIC: Final[str] = '28be4a4a-cd67-11e9-a32f-2a2ae2dbcce4'
 
 # GAN Gen3 protocol BLE service
-GAN_GEN3_SERVICE =                '8653000a-43e6-47b7-9cb0-5fc21d4ae340'
-GAN_GEN3_STATE_CHARACTERISTIC =   '8653000b-43e6-47b7-9cb0-5fc21d4ae340'
-GAN_GEN3_COMMAND_CHARACTERISTIC = '8653000c-43e6-47b7-9cb0-5fc21d4ae340'
+GAN_GEN3_SERVICE: Final[str] =                '8653000a-43e6-47b7-9cb0-5fc21d4ae340'
+GAN_GEN3_STATE_CHARACTERISTIC: Final[str] =   '8653000b-43e6-47b7-9cb0-5fc21d4ae340'
+GAN_GEN3_COMMAND_CHARACTERISTIC: Final[str] = '8653000c-43e6-47b7-9cb0-5fc21d4ae340'
 
 # GAN Gen4 protocol BLE service
-GAN_GEN4_SERVICE =                '00000010-0000-fff7-fff6-fff5fff4fff0'
-GAN_GEN4_STATE_CHARACTERISTIC =   '0000fff6-0000-1000-8000-00805f9b34fb'
-GAN_GEN4_COMMAND_CHARACTERISTIC = '0000fff5-0000-1000-8000-00805f9b34fb'
+GAN_GEN4_SERVICE: Final[str] =                '00000010-0000-fff7-fff6-fff5fff4fff0'
+GAN_GEN4_STATE_CHARACTERISTIC: Final[str] =   '0000fff6-0000-1000-8000-00805f9b34fb'
+GAN_GEN4_COMMAND_CHARACTERISTIC: Final[str] = '0000fff5-0000-1000-8000-00805f9b34fb'
 
 # Moyu Weilong v10 protocol BLE service
-MOYU_WEILONG_SERVICE =                '0783b03e-7735-b5a0-1760-a305d2795cb0'
-MOYU_WEILONG_STATE_CHARACTERISTIC =   '0783b03e-7735-b5a0-1760-a305d2795cb1'
-MOYU_WEILONG_COMMAND_CHARACTERISTIC = '0783b03e-7735-b5a0-1760-a305d2795cb2'
+MOYU_WEILONG_SERVICE: Final[str] =                '0783b03e-7735-b5a0-1760-a305d2795cb0'
+MOYU_WEILONG_STATE_CHARACTERISTIC: Final[str] =   '0783b03e-7735-b5a0-1760-a305d2795cb1'
+MOYU_WEILONG_COMMAND_CHARACTERISTIC: Final[str] = '0783b03e-7735-b5a0-1760-a305d2795cb2'
 
 # Key used by GAN Gen2, Gen3 and Gen4 cubes
-GAN_ENCRYPTION_KEY = {
+GAN_ENCRYPTION_KEY: Final[dict[str, list[int]]] = {
     'key': [
         0x01, 0x02, 0x42, 0x28,
         0x31, 0x91, 0x16, 0x07,
@@ -46,7 +47,7 @@
 }
 
 # Key used by MoYu AI 2023
-MOYU_AI_ENCRYPTION_KEY = {
+MOYU_AI_ENCRYPTION_KEY: Final[dict[str, list[int]]] = {
     'key': [
         0x05, 0x12, 0x02, 0x45,
         0x02, 0x01, 0x29, 0x56,
@@ -63,7 +64,7 @@
 
 
 # Key used by MoYu Weilong v10
-MOYU_WEILONG_ENCRYPTION_KEY = {
+MOYU_WEILONG_ENCRYPTION_KEY: Final[dict[str, list[int]]] = {
     'key': [
         0x15, 0x77, 0x3A, 0x5C,
         0x67, 0x0E, 0x2D, 0x1F,

term_timer/bluetooth/drivers/base.py

@@ -1,34 +1,41 @@
+from collections.abc import Sequence
+from typing import Any
+from typing import ClassVar
+
 from bleak import BleakClient
 
+from term_timer.bluetooth.encrypter import GanGen2CubeEncrypter
+from term_timer.bluetooth.types import EventDict
+
 
 class Driver:
-    service_uid = ''
-    state_characteristic_uid = ''
-    command_characteristic_uid = ''
+    service_uid: ClassVar[str] = ''
+    state_characteristic_uid: ClassVar[str] = ''
+    command_characteristic_uid: ClassVar[str] = ''
 
-    disable_gyro = True
+    disable_gyro: bool = True
 
     def __init__(self, client: BleakClient) -> None:
-        self.client = client
+        self.client: BleakClient = client
 
-        self.events: list[dict[str, object]] = []
-        self.cypher = self.init_cypher()
+        self.events: list[EventDict] = []
+        self.cypher: GanGen2CubeEncrypter = self.init_cypher()
 
-    def init_cypher(self) -> None:
-        pass
+    def init_cypher(self) -> GanGen2CubeEncrypter:
+        raise NotImplementedError
 
-    def send_command_handler(self, command: str) -> bool:
+    def send_command_handler(self, command: str) -> bytes | bool:
         raise NotImplementedError
 
-    def event_handler(self, sender, data) -> list[dict[str, object]]:
+    async def event_handler(self, sender: int, data: bytes) -> list[EventDict]:
         raise NotImplementedError
 
-    def add_event(self,
-                  store: list[dict[str, object]],
-                  event: dict[str, object] | list[dict[str, object]]) -> None:
-        if isinstance(event, list):
-            store.extend(event)
-            self.events.extend(event)
+    def add_event(self, store: list[EventDict],
+                  event: dict[str, Any] | Sequence[dict[str, Any]]) -> None:
+        if isinstance(event, (list, tuple)):
+            for e in event:
+                store.append(e)
+                self.events.append(e)
         else:
-            store.append(event)
-            self.events.append(event)
+            store.append(event)  # type: ignore[arg-type]
+            self.events.append(event)  # type: ignore[arg-type]

term_timer/bluetooth/drivers/gan_gen2.py

@@ -6,7 +6,9 @@
 import time
 from datetime import datetime
 from datetime import timezone
+from typing import ClassVar
 
+from bleak import BleakClient
 from cubing_algs.facelets import cubies_to_facelets
 
 from term_timer.bluetooth.constants import GAN_ENCRYPTION_KEY
@@ -18,6 +20,7 @@
 from term_timer.bluetooth.encrypter import GanGen2CubeEncrypter
 from term_timer.bluetooth.message import GanProtocolMessage
 from term_timer.bluetooth.salt import get_salt
+from term_timer.bluetooth.types import EventDict
 
 logger = logging.getLogger(__name__)
 
@@ -33,20 +36,20 @@ class GanGen2Driver(Driver):
     Monster Go 3Ai
     MoYu AI 2023
     """
-    service_uid = GAN_GEN2_SERVICE
-    state_characteristic_uid = GAN_GEN2_STATE_CHARACTERISTIC
-    command_characteristic_uid = GAN_GEN2_COMMAND_CHARACTERISTIC
-    encrypter = GanGen2CubeEncrypter
+    service_uid: ClassVar[str] = GAN_GEN2_SERVICE
+    state_characteristic_uid: ClassVar[str] = GAN_GEN2_STATE_CHARACTERISTIC
+    command_characteristic_uid: ClassVar[str] = GAN_GEN2_COMMAND_CHARACTERISTIC
+    encrypter: ClassVar[type[GanGen2CubeEncrypter]] = GanGen2CubeEncrypter
 
-    def __init__(self, client):
+    def __init__(self, client: BleakClient) -> None:
         super().__init__(client)
 
-        self.last_serial = -1
-        self.cube_timestamp = 0
-        self.last_move_timestamp = 0
+        self.last_serial: int = -1
+        self.cube_timestamp: float = 0.0
+        self.last_move_timestamp: datetime | None = None
 
-    def init_cypher(self):
-        if self.client.name.startswith('AiCube'):
+    def init_cypher(self) -> GanGen2CubeEncrypter:
+        if self.client.name and self.client.name.startswith('AiCube'):
             return self.encrypter(
                 MOYU_AI_ENCRYPTION_KEY['key'],
                 MOYU_AI_ENCRYPTION_KEY['iv'],
@@ -58,7 +61,7 @@ def init_cypher(self):
             get_salt(self.client.address),
         )
 
-    def send_command_handler(self, command: str):
+    def send_command_handler(self, command: str) -> bytes | bool:
         msg = bytearray(20)
 
         if command == 'REQUEST_FACELETS':
@@ -78,12 +81,12 @@ def send_command_handler(self, command: str):
 
         return self.cypher.encrypt(bytes(msg))
 
-    async def event_handler(self, sender, data):  # noqa: ARG002
+    async def event_handler(self, sender: int, data: bytes) -> list[EventDict]:  # noqa: ARG002
         """Process notifications from the cube"""
         clock = time.perf_counter_ns()
         timestamp = datetime.now(tz=timezone.utc)  # noqa: UP017
 
-        events = []
+        events: list[EventDict] = []
 
         msg = GanProtocolMessage(
             self.cypher.decrypt(data),
@@ -140,13 +143,16 @@ async def event_handler(self, sender, data):  # noqa: ARG002
                 face = msg.get_bit_word(12 + 5 * i, 4)
                 direction = msg.get_bit_word(16 + 5 * i, 1)
                 move = 'URFDLB'[face] + " '"[direction]
-                elapsed = msg.get_bit_word(47 + 16 * i, 16)
+                elapsed_raw = msg.get_bit_word(47 + 16 * i, 16)
 
                 # In case of 16-bit cube timestamp register overflow
-                if elapsed == 0 and self.last_move_timestamp:
+                elapsed: float
+                if elapsed_raw == 0 and self.last_move_timestamp is not None:
                     elapsed = (
                         timestamp - self.last_move_timestamp
                     ).total_seconds()
+                else:
+                    elapsed = float(elapsed_raw)
 
                 self.cube_timestamp += elapsed
                 payload = {

term_timer/bluetooth/drivers/gan_gen3.py

@@ -4,9 +4,14 @@
 """
 import logging
 import time
+from collections.abc import Sequence
 from datetime import datetime
 from datetime import timezone
+from typing import Any
+from typing import ClassVar
+from typing import cast
 
+from bleak import BleakClient
 from cubing_algs.facelets import cubies_to_facelets
 
 from term_timer.bluetooth.constants import DEBOUNCE
@@ -15,6 +20,8 @@
 from term_timer.bluetooth.constants import GAN_GEN3_STATE_CHARACTERISTIC
 from term_timer.bluetooth.drivers.gan_gen2 import GanGen2Driver
 from term_timer.bluetooth.message import GanProtocolMessage
+from term_timer.bluetooth.types import EventDict
+from term_timer.bluetooth.types import MoveEventDict
 
 logger = logging.getLogger(__name__)
 
@@ -23,19 +30,19 @@ class GanGen3Driver(GanGen2Driver):
     """
     GAN356 i Carry 2
     """
-    service_uid = GAN_GEN3_SERVICE
-    state_characteristic_uid = GAN_GEN3_STATE_CHARACTERISTIC
-    command_characteristic_uid = GAN_GEN3_COMMAND_CHARACTERISTIC
+    service_uid: ClassVar[str] = GAN_GEN3_SERVICE
+    state_characteristic_uid: ClassVar[str] = GAN_GEN3_STATE_CHARACTERISTIC
+    command_characteristic_uid: ClassVar[str] = GAN_GEN3_COMMAND_CHARACTERISTIC
 
-    def __init__(self, client):
+    def __init__(self, client: BleakClient) -> None:
         super().__init__(client)
 
-        self.serial = -1
-        self.last_serial = -1
-        self.last_local_timestamp = None
-        self.move_buffer = []
+        self.serial: int = -1
+        self.last_serial: int = -1
+        self.last_local_timestamp: datetime | None = None
+        self.move_buffer: list[MoveEventDict] = []
 
-    def send_command_handler(self, command: str):
+    def send_command_handler(self, command: str) -> bytes | bool:
         msg = bytearray(16)
 
         if command == 'REQUEST_FACELETS':
@@ -58,7 +65,7 @@ def send_command_handler(self, command: str):
 
         return self.cypher.encrypt(bytes(msg))
 
-    async def request_move_history(self, serial, count):
+    async def request_move_history(self, serial: int, count: int) -> None:
         msg = bytearray(16)
 
         # Move history response data is byte-aligned,
@@ -89,8 +96,8 @@ async def request_move_history(self, serial, count):
             self.cypher.encrypt(bytes(msg)),
         )
 
-    async def evict_move_buffer(self):
-        evicted_events = []
+    async def evict_move_buffer(self) -> list[MoveEventDict]:
+        evicted_events: list[MoveEventDict] = []
 
         while len(self.move_buffer) > 0:
             buffer_head = self.move_buffer[0]
@@ -104,19 +111,20 @@ async def evict_move_buffer(self):
             self.last_serial = buffer_head['serial']
 
         if len(self.move_buffer) > 16:
-            self.client.disconnect()
+            await self.client.disconnect()
 
         return evicted_events
 
-    def is_serial_in_range(self, start, end, serial,
-                           *, closed_start=False, closed_end=False):
+    def is_serial_in_range(self, start: int, end: int, serial: int, *,
+                           closed_start: bool = False,
+                           closed_end: bool = False) -> bool:
         return (
             ((end - start) & 0xFF) >= ((serial - start) & 0xFF)
             and (closed_start or ((start - serial) & 0xFF) > 0)
             and (closed_end or ((end - serial) & 0xFF) > 0)
         )
 
-    def inject_missed_move_to_buffer(self, move):
+    def inject_missed_move_to_buffer(self, move: MoveEventDict) -> None:
         if len(self.move_buffer) > 0:
             buffer_head = self.move_buffer[0]
 
@@ -142,7 +150,7 @@ def inject_missed_move_to_buffer(self, move):
         ):
             self.move_buffer.insert(0, move)
 
-    async def check_if_move_missed(self):
+    async def check_if_move_missed(self) -> None:
         diff = (self.serial - self.last_serial) & 0xFF
 
         if diff > 0 and self.serial != 0:
@@ -152,12 +160,12 @@ async def check_if_move_missed(self):
             ) & 0xFF
             await self.request_move_history(start_serial, diff + 1)
 
-    async def event_handler(self, sender, data):  # noqa: ARG002
+    async def event_handler(self, sender: int, data: bytes) -> list[EventDict]:  # noqa: ARG002
         """Process notifications from the cube"""
         clock = time.perf_counter_ns()
         timestamp = datetime.now(tz=timezone.utc)  # noqa: UP017
 
-        events = []
+        events: list[EventDict] = []
 
         msg = GanProtocolMessage(
             self.cypher.decrypt(data),
@@ -196,7 +204,9 @@ async def event_handler(self, sender, data):  # noqa: ARG002
                         'move': move.strip(),
                     },
                 )
-            self.add_event(events, await self.evict_move_buffer())
+            evicted = await self.evict_move_buffer()
+            if evicted:
+                self.add_event(events, cast(Sequence[dict[str, Any]], evicted))
 
         elif event == 0x02:  # Facelets
             serial = msg.get_bit_word(24, 16, little_endian=True)
@@ -280,7 +290,9 @@ async def event_handler(self, sender, data):  # noqa: ARG002
                         },
                     )
 
-            self.add_event(events, await self.evict_move_buffer())
+            evicted = await self.evict_move_buffer()
+            if evicted:
+                self.add_event(events, cast(Sequence[dict[str, Any]], evicted))
 
         elif event == 0x07:  # Hardware
             sw_major = msg.get_bit_word(72, 4)