From 4c3746a5b26f0a67b736b435354a181a511bde24 Mon Sep 17 00:00:00 2001 From: Josh Wu Date: Mon, 5 Jan 2026 23:21:12 +0800 Subject: [PATCH] Fix some typos and annotations --- apps/pair.py | 81 ++++++++++++++++++++++--------------------- bumble/device.py | 13 +++++-- bumble/gatt_server.py | 2 +- tests/device_test.py | 4 +-- tests/self_test.py | 6 ++-- tests/test_utils.py | 14 ++++---- 6 files changed, 65 insertions(+), 55 deletions(-) diff --git a/apps/pair.py b/apps/pair.py index b9f3ae9..135c9be 100644 --- a/apps/pair.py +++ b/apps/pair.py @@ -15,6 +15,8 @@ # ----------------------------------------------------------------------------- # Imports # ----------------------------------------------------------------------------- +from __future__ import annotations + import asyncio import logging import os @@ -63,7 +65,7 @@ POST_PAIRING_DELAY = 1 # ----------------------------------------------------------------------------- class Waiter: - instance = None + instance: Waiter | None = None def __init__(self, linger=False): self.done = asyncio.get_running_loop().create_future() @@ -327,25 +329,25 @@ async def on_pairing_failure(connection, reason): # ----------------------------------------------------------------------------- async def pair( - mode, - sc, - mitm, - bond, - ctkd, - advertising_address, - identity_address, - linger, - io, - oob, - prompt, - request, - print_keys, - keystore_file, - advertise_service_uuids, - advertise_appearance, - device_config, - hci_transport, - address_or_name, + mode: str, + sc: bool, + mitm: bool, + bond: bool, + ctkd: bool, + advertising_address: str, + identity_address: str, + linger: bool, + io: str, + oob: str, + prompt: bool, + request: bool, + print_keys: bool, + keystore_file: str, + advertise_service_uuids: str, + advertise_appearance: str, + device_config: str, + hci_transport: str, + address_or_name: str, ): Waiter.instance = Waiter(linger=linger) @@ -403,6 +405,7 @@ async def pair( # Create an OOB context if needed if oob: our_oob_context = OobContext() + legacy_context: OobLegacyContext | None if oob == '-': shared_data = None legacy_context = OobLegacyContext() @@ -661,25 +664,25 @@ class LogHandler(logging.Handler): @click.argument('hci_transport') @click.argument('address-or-name', required=False) def main( - mode, - sc, - mitm, - bond, - ctkd, - advertising_address, - identity_address, - linger, - io, - oob, - prompt, - request, - print_keys, - keystore_file, - advertise_service_uuid, - advertise_appearance, - device_config, - hci_transport, - address_or_name, + mode: str, + sc: bool, + mitm: bool, + bond: bool, + ctkd: bool, + advertising_address: str, + identity_address: str, + linger: bool, + io: str, + oob: str, + prompt: bool, + request: bool, + print_keys: bool, + keystore_file: str, + advertise_service_uuid: str, + advertise_appearance: str, + device_config: str, + hci_transport: str, + address_or_name: str, ): # Setup logging log_handler = LogHandler() diff --git a/bumble/device.py b/bumble/device.py index 813f394..7789198 100644 --- a/bumble/device.py +++ b/bumble/device.py @@ -171,6 +171,7 @@ class Advertisement: ) sid: int = 0 data_bytes: bytes = b'' + data: AdvertisingData = field(init=False) # Constants TX_POWER_NOT_AVAILABLE: ClassVar[int] = ( @@ -480,6 +481,7 @@ class PeriodicAdvertisement: rssi: int = hci.HCI_LE_Periodic_Advertising_Report_Event.RSSI_NOT_AVAILABLE is_truncated: bool = False data_bytes: bytes = b'' + data: AdvertisingData | None = field(init=False) # Constants TX_POWER_NOT_AVAILABLE: ClassVar[int] = ( @@ -1099,6 +1101,7 @@ class Big(utils.EventEmitter): max_pdu: int = 0 iso_interval: float = 0.0 # ISO interval, in milliseconds bis_links: Sequence[BisLink] = () + device: Device = field(init=False) def __post_init__(self) -> None: super().__init__() @@ -1160,6 +1163,7 @@ class BigSync(utils.EventEmitter): max_pdu: int = 0 iso_interval: float = 0.0 bis_links: Sequence[BisLink] = () + device: Device = field(init=False) def __post_init__(self) -> None: super().__init__() @@ -1655,6 +1659,7 @@ class BisLink(_IsoLink): handle: int big: Big | BigSync sink: Callable[[hci.HCI_IsoDataPacket], Any] | None = None + device: Device = field(init=False) def __post_init__(self) -> None: super().__init__() @@ -2088,9 +2093,10 @@ class DeviceConfiguration: l2cap.L2CAP_Information_Request.ExtendedFeatures.ENHANCED_RETRANSMISSION_MODE, ) eatt_enabled: bool = False + gatt_services: list[dict[str, Any]] = field(init=False) def __post_init__(self) -> None: - self.gatt_services: list[dict[str, Any]] = [] + self.gatt_services = [] def load_from_dict(self, config: dict[str, Any]) -> None: config = copy.deepcopy(config) @@ -2270,6 +2276,7 @@ class Device(utils.CompositeEventEmitter): big_syncs: dict[int, BigSync] _pending_cis: dict[int, tuple[int, int]] gatt_service: gatt_service.GenericAttributeProfileService | None = None + keystore: KeyStore | None = None EVENT_ADVERTISEMENT = "advertisement" EVENT_PERIODIC_ADVERTISING_SYNC_TRANSFER = "periodic_advertising_sync_transfer" @@ -4527,8 +4534,8 @@ class Device(utils.CompositeEventEmitter): ediv = 0 elif keys.ltk_central is not None: ltk = keys.ltk_central.value - rand = keys.ltk_central.rand - ediv = keys.ltk_central.ediv + rand = keys.ltk_central.rand or b'' + ediv = keys.ltk_central.ediv or 0 else: raise InvalidOperationError('no LTK found for peer') diff --git a/bumble/gatt_server.py b/bumble/gatt_server.py index 3d29a76..29e4c28 100644 --- a/bumble/gatt_server.py +++ b/bumble/gatt_server.py @@ -111,7 +111,7 @@ class Server(utils.EventEmitter): ) -> l2cap.LeCreditBasedChannelServer: def on_channel(channel: l2cap.LeCreditBasedChannel): logger.debug( - "New EATT Bearer Conenction=0x%04X CID=0x%04X", + "New EATT Bearer Connection=0x%04X CID=0x%04X", channel.connection.handle, channel.source_cid, ) diff --git a/tests/device_test.py b/tests/device_test.py index 8f2df8a..511dbba 100644 --- a/tests/device_test.py +++ b/tests/device_test.py @@ -292,9 +292,9 @@ async def test_legacy_advertising_disconnection(auto_restart): await devices[0].start_advertising( auto_restart=auto_restart, advertising_interval_min=1.0 ) - connecion = await devices[1].connect(devices[0].random_address) + connection = await devices[1].connect(devices[0].random_address) - await connecion.disconnect() + await connection.disconnect() await async_barrier() await async_barrier() diff --git a/tests/self_test.py b/tests/self_test.py index aa2e7a5..17acfdb 100644 --- a/tests/self_test.py +++ b/tests/self_test.py @@ -57,15 +57,13 @@ async def test_self_disconnection(): await two_devices.setup_connection() await two_devices.connections[0].disconnect() await async_barrier() - assert two_devices.connections[0] is None - assert two_devices.connections[1] is None + assert not two_devices.connections two_devices = TwoDevices() await two_devices.setup_connection() await two_devices.connections[1].disconnect() await async_barrier() - assert two_devices.connections[0] is None - assert two_devices.connections[1] is None + assert not two_devices.connections # ----------------------------------------------------------------------------- diff --git a/tests/test_utils.py b/tests/test_utils.py index 154ade6..861dc26 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -31,10 +31,10 @@ from bumble.transport.common import AsyncPipeSink # ----------------------------------------------------------------------------- class Devices: - connections: list[Connection | None] + connections: dict[int, Connection] def __init__(self, num_devices: int) -> None: - self.connections = [None for _ in range(num_devices)] + self.connections = {} self.link = LocalLink() addresses = [":".join([f"F{i}"] * 6) for i in range(num_devices)] @@ -60,12 +60,14 @@ class Devices: asyncio.get_event_loop().create_future() for _ in range(num_devices) ] - def on_connection(self, which, connection): + def on_connection(self, which: int, connection: Connection) -> None: self.connections[which] = connection - connection.on('disconnection', lambda code: self.on_disconnection(which)) + connection.on( + connection.EVENT_DISCONNECTION, lambda *_: self.on_disconnection(which) + ) - def on_disconnection(self, which): - self.connections[which] = None + def on_disconnection(self, which: int) -> None: + self.connections.pop(which, None) def on_paired(self, which: int, keys: PairingKeys) -> None: self.paired[which].set_result(keys)