From 059d04440d6717308151fc697040a67dad32e085 Mon Sep 17 00:00:00 2001 From: Frederik Leonhardt Date: Sat, 21 Feb 2026 22:39:33 +1300 Subject: [PATCH 1/9] Update dev dependencies --- pyproject.toml | 10 +++++----- src/pb_ble/cli/__init__.py | 8 ++++---- 2 files changed, 9 insertions(+), 9 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 9d8bf97..8e417e0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,12 +28,12 @@ dependencies = [ debug = ["bluetooth-data-tools ~= 1.15"] dev = [ "async-timer ~= 1.1.6", - "mypy ~= 1.16.0", + "mypy ~= 1.19.1", "pytest ~= 8.3", - "pytest-asyncio ~= 1.0.0", - "pytest-mock ~= 3.14.1", - "python-dbusmock ~= 0.34.3", - "ruff ~= 0.11.8", + "pytest-asyncio ~= 1.3.0", + "pytest-mock ~= 3.15.1", + "python-dbusmock ~= 0.38.1", + "ruff ~= 0.15.2", "types-cachetools ~= 5.3", "types-setuptools", ] diff --git a/src/pb_ble/cli/__init__.py b/src/pb_ble/cli/__init__.py index d37ba96..5451d3f 100644 --- a/src/pb_ble/cli/__init__.py +++ b/src/pb_ble/cli/__init__.py @@ -93,11 +93,11 @@ def setup_cli_logging(): # ISO8601 dateformat for logging including milliseconds logging.Formatter.formatTime = ( # type: ignore [method-assign] - lambda self, record, datefmt=None: datetime.datetime.fromtimestamp( - record.created, datetime.timezone.utc + lambda self, record, datefmt=None: ( + datetime.datetime.fromtimestamp(record.created, datetime.timezone.utc) + .astimezone() + .isoformat(sep="T", timespec="milliseconds") ) - .astimezone() - .isoformat(sep="T", timespec="milliseconds") ) From e75c27cdfa42db425eb34944904930d37f379541 Mon Sep 17 00:00:00 2001 From: Frederik Leonhardt Date: Sat, 21 Feb 2026 22:52:32 +1300 Subject: [PATCH 2/9] Update bleak and bluetooth-adapters --- pyproject.toml | 4 ++-- src/pb_ble/bluezdbus/observer.py | 3 +-- 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 8e417e0..42fc1bc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,8 +17,8 @@ classifiers = [ ] dependencies = [ - "bleak >= 0.22.0", - "bluetooth-adapters ~= 0.16", + "bleak ~= 2.0", + "bluetooth-adapters ~= 2.1", "cachetools ~= 5.3", "dbus-fast ~= 2.15", "pybricks @ git+https://github.com/fkleon/pybricks-api@ble-type-fixes", diff --git a/src/pb_ble/bluezdbus/observer.py b/src/pb_ble/bluezdbus/observer.py index e300e17..703abe2 100644 --- a/src/pb_ble/bluezdbus/observer.py +++ b/src/pb_ble/bluezdbus/observer.py @@ -8,9 +8,8 @@ from typing import NamedTuple, Sequence from bleak import AdvertisementData, BleakScanner, BLEDevice +from bleak.args.bluez import BlueZDiscoveryFilters, BlueZScannerArgs, OrPattern from bleak.assigned_numbers import AdvertisementDataType -from bleak.backends.bluezdbus.advertisement_monitor import OrPattern -from bleak.backends.bluezdbus.scanner import BlueZDiscoveryFilters, BlueZScannerArgs from cachetools import TTLCache from ..constants import ( From ec001d5d307bd22eafa82645f489d63cfccc9db2 Mon Sep 17 00:00:00 2001 From: Frederik Leonhardt Date: Sat, 21 Feb 2026 23:06:16 +1300 Subject: [PATCH 3/9] Update dbus-fast --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 42fc1bc..320b1a5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,7 +20,7 @@ dependencies = [ "bleak ~= 2.0", "bluetooth-adapters ~= 2.1", "cachetools ~= 5.3", - "dbus-fast ~= 2.15", + "dbus-fast ~= 4.0", "pybricks @ git+https://github.com/fkleon/pybricks-api@ble-type-fixes", ] From 9805c3348896d75eb4e2b66cb61328ccdbd31402 Mon Sep 17 00:00:00 2001 From: Frederik Leonhardt Date: Sat, 21 Feb 2026 23:06:35 +1300 Subject: [PATCH 4/9] Replace `method()` with `dbus_method()` --- src/pb_ble/bluezdbus/advertisement.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/pb_ble/bluezdbus/advertisement.py b/src/pb_ble/bluezdbus/advertisement.py index 14d9349..4ae34c6 100644 --- a/src/pb_ble/bluezdbus/advertisement.py +++ b/src/pb_ble/bluezdbus/advertisement.py @@ -19,7 +19,7 @@ from dbus_fast.aio import ProxyInterface, ProxyObject from dbus_fast.constants import PropertyAccess -from dbus_fast.service import ServiceInterface, _Property, dbus_property, method +from dbus_fast.service import ServiceInterface, _Property, dbus_method, dbus_property from dbus_fast.signature import Variant from ..constants import ( @@ -165,8 +165,8 @@ def _disable_props(self, *prop_names: str): else: prop.disabled = True - @method() - def Release(self): + @dbus_method() + def Release(self) -> None: logger.debug("Released advertisement: %s", self) @dbus_property(access=PropertyAccess.READ) @@ -445,8 +445,8 @@ def __init__( # for prop in ServiceInterface._get_properties(self): # logger.debug("Property %s (%s)", prop.name, "DISABLED" if prop.disabled else "ENABLED") - @method() - def Release(self): + @dbus_method() + def Release(self) -> None: super().Release() self.on_release(self.path) From a33e6690a7cbb384d239990030cb61eee83c20e9 Mon Sep 17 00:00:00 2001 From: Frederik Leonhardt Date: Sat, 21 Feb 2026 23:48:08 +1300 Subject: [PATCH 5/9] Migrate to dbus-fast type annotations --- src/pb_ble/bluezdbus/advertisement.py | 164 +++++++++++--------------- 1 file changed, 72 insertions(+), 92 deletions(-) diff --git a/src/pb_ble/bluezdbus/advertisement.py b/src/pb_ble/bluezdbus/advertisement.py index 4ae34c6..76e56d6 100644 --- a/src/pb_ble/bluezdbus/advertisement.py +++ b/src/pb_ble/bluezdbus/advertisement.py @@ -11,13 +11,22 @@ import logging from enum import Enum from typing import ( + Annotated, Any, Callable, - no_type_check, overload, ) from dbus_fast.aio import ProxyInterface, ProxyObject +from dbus_fast.annotations import ( + DBusBool, + DBusDict, + DBusInt16, + DBusSignature, + DBusStr, + DBusUInt16, + DBusUInt32, +) from dbus_fast.constants import PropertyAccess from dbus_fast.service import ServiceInterface, _Property, dbus_method, dbus_property from dbus_fast.signature import Variant @@ -79,6 +88,11 @@ class Feature(Enum): """Indicates whether multiple advertising will be offloaded to the controller.""" +DBusArrayString = Annotated[list[str], DBusSignature("as")] +DBusUInt16Dict = Annotated[dict[int, Variant], DBusSignature("a{qv}")] +DBusByteDict = Annotated[dict[int, Variant], DBusSignature("a{yv}")] + + class LEAdvertisement(ServiceInterface): """ Implementation of the `org.bluez.LEAdvertisement1` D-Bus interface. @@ -101,10 +115,10 @@ def __init__( self._type: str = advertising_type.value self._service_uuids: list[str] = [] - self._manufacturer_data: dict[int, bytes] = {} # uint16 -> bytes + self._manufacturer_data: dict[int, Variant] = {} # uint16 -> bytes self._solicit_uuids: list[str] = [] - self._service_data: dict[str | int, bytes] = {} # uint16 | str -> bytes - self._data: dict[int, bytes] = {} # EXPERIMENTAL # uint8 -> bytes + self._service_data: dict[str, Variant] = {} # uint16 | str -> bytes + self._data: dict[int, Variant] = {} # EXPERIMENTAL # uint8 -> bytes self._discoverable: bool = False # EXPERIMENTAL self._discoverable_timeout: int = 0 # EXPERIMENTAL # uint16 self._includes: list[str] = [i.value for i in includes] @@ -170,198 +184,169 @@ def Release(self) -> None: logger.debug("Released advertisement: %s", self) @dbus_property(access=PropertyAccess.READ) - @no_type_check - def Type(self) -> "s": # type: ignore # noqa: F821 + def Type(self) -> DBusStr: """ Determines the type of advertising packet requested. """ return self._type - @Type.setter # type: ignore - @no_type_check - def Type(self, type: "s"): # type: ignore # noqa: F821 + @Type.setter # type: ignore[no-redef] + def Type(self, type: DBusStr) -> None: self._type = type @dbus_property() - @no_type_check - def ServiceUUIDs(self) -> "as": # type: ignore # noqa: F821 F722 + def ServiceUUIDs(self) -> DBusArrayString: """ List of UUIDs to include in the "Service UUID" field of the Advertising Data. """ return self._service_uuids - @ServiceUUIDs.setter # type: ignore - @no_type_check - def ServiceUUIDs(self, service_uuids: "as"): # type: ignore # noqa: F821 F722 + @ServiceUUIDs.setter # type: ignore[no-redef] + def ServiceUUIDs(self, service_uuids: DBusArrayString) -> None: self._service_uuids = service_uuids @dbus_property() - @no_type_check - def ManufacturerData(self) -> "a{qv}": # type: ignore # noqa: F821 F722 + def ManufacturerData(self) -> DBusUInt16Dict: """ Manufacturer Data fields to include in the Advertising Data. Keys are the Manufacturer ID to associate with the data. """ return self._manufacturer_data - @ManufacturerData.setter # type: ignore - @no_type_check - def ManufacturerData(self, data: "a{qv}"): # type: ignore # noqa: F821 F722 + @ManufacturerData.setter # type: ignore[no-redef] + def ManufacturerData(self, data: DBusUInt16Dict) -> None: self._manufacturer_data = data @dbus_property() - @no_type_check - def SolicitUUIDs(self) -> "as": # type: ignore # noqa: F821 F722 + def SolicitUUIDs(self) -> DBusArrayString: """ Array of UUIDs to include in "Service Solicitation" Advertisement Data. """ return self._solicit_uuids - @SolicitUUIDs.setter # type: ignore - @no_type_check - def SolicitUUIDs(self, uuids: "as"): # type: ignore # noqa: F821 F722 + @SolicitUUIDs.setter # type: ignore[no-redef] + def SolicitUUIDs(self, uuids: DBusArrayString) -> None: self._solicit_uuids = uuids @dbus_property() - @no_type_check - def ServiceData(self) -> "a{sv}": # type: ignore # noqa: F821 F722 + def ServiceData(self) -> DBusDict: """ Service Data elements to include. The keys are the UUID to associate with the data. """ return self._service_data - @ServiceData.setter # type: ignore - @no_type_check - def ServiceData(self, data: "a{sv}"): # type: ignore # noqa: F821 F722 + @ServiceData.setter # type: ignore[no-redef] + def ServiceData(self, data: DBusDict) -> None: self._service_data = data @dbus_property(disabled=True) - @no_type_check - def Data(self) -> "a{yv}": # type: ignore # noqa: F821 F722 + def Data(self) -> DBusByteDict: """ Advertising Data to include. Key is the advertising type and value is the data as byte array. """ return self._data - @Data.setter # type: ignore - @no_type_check - def Data(self, data: "a{yv}"): # type: ignore # noqa: F821 F722 + @Data.setter # type: ignore[no-redef] + def Data(self, data: DBusByteDict) -> None: self._data = data @dbus_property(disabled=True) - @no_type_check - def Discoverable(self) -> "b": # type: ignore # noqa: F821 F722 + def Discoverable(self) -> DBusBool: """ Advertise as general discoverable. When present this will override adapter Discoverable property. """ return self._discoverable - @Discoverable.setter # type: ignore - @no_type_check - def Discoverable(self, discoverable: "b"): # type: ignore # noqa: F821 F722 + @Discoverable.setter # type: ignore[no-redef] + def Discoverable(self, discoverable: DBusBool) -> None: self._discoverable = discoverable @dbus_property(disabled=True) - @no_type_check - def DiscoverableTimeout(self) -> "q": # type: ignore # noqa: F821 F722 + def DiscoverableTimeout(self) -> DBusUInt16: """ The discoverable timeout in seconds. A value of zero means that the timeout is disabled and it will stay in discoverable/limited mode forever. """ return self._discoverable_timeout - @DiscoverableTimeout.setter # type: ignore - @no_type_check - def DiscoverableTimeout(self, timeout: "q"): # type: ignore # noqa: F821 F722 + @DiscoverableTimeout.setter # type: ignore[no-redef] + def DiscoverableTimeout(self, timeout: DBusUInt16) -> None: self._discoverable_timeout = timeout @dbus_property() - @no_type_check - def Includes(self) -> "as": # type: ignore # noqa: F821 F722 + def Includes(self) -> DBusArrayString: """ List of features to be included in the advertising packet. """ return self._includes - @Includes.setter # type: ignore - @no_type_check - def Includes(self, includes: "as"): # type: ignore # noqa: F821 F722 + @Includes.setter # type: ignore[no-redef] + def Includes(self, includes: DBusArrayString) -> None: self._includes = includes @dbus_property() - @no_type_check - def LocalName(self) -> "s": # type: ignore # noqa: F821 N802 + def LocalName(self) -> DBusStr: """ Local name to be used in the advertising report. If the string is too big to fit into the packet it will be truncated. """ return self._local_name - @LocalName.setter # type: ignore - @no_type_check - def LocalName(self, name: "s"): # type: ignore # noqa: F821 N802 + @LocalName.setter # type: ignore[no-redef] + def LocalName(self, name: DBusStr) -> None: self._local_name = name @dbus_property() - @no_type_check - def Appearance(self) -> "q": # type: ignore # noqa: F821 N802 + def Appearance(self) -> DBusUInt16: """ Appearance to be used in the advertising report. """ return self._appearance - @Appearance.setter # type: ignore - @no_type_check - def Appearance(self, appearance: "q"): # type: ignore # noqa: F821 N802 + @Appearance.setter # type: ignore[no-redef] + def Appearance(self, appearance: DBusUInt16) -> None: self._appearance = appearance @dbus_property() - @no_type_check - def Duration(self) -> "q": # type: ignore # noqa: F821 N802 + def Duration(self) -> DBusUInt16: """ Rotation duration of the advertisement in seconds. If there are other applications advertising no duration is set the default is 2 seconds. """ return self._duration - @Duration.setter # type: ignore - @no_type_check - def Duration(self, seconds: "q"): # type: ignore # noqa: F821 N802 + @Duration.setter # type: ignore[no-redef] + def Duration(self, seconds: DBusUInt16) -> None: self._duration = seconds @dbus_property() - @no_type_check - def Timeout(self) -> "q": # type: ignore # noqa: F821 N802 + def Timeout(self) -> DBusUInt16: """ Timeout of the advertisement in seconds. This defines the lifetime of the advertisement. """ return self._timeout - @Timeout.setter # type: ignore - @no_type_check - def Timeout(self, seconds: "q"): # type: ignore # noqa: F821 N802 + @Timeout.setter # type: ignore[no-redef] + def Timeout(self, seconds: DBusUInt16) -> None: self._timeout = seconds @dbus_property(disabled=True) - @no_type_check - def SecondaryChannel(self) -> "s": # type: ignore # noqa: F821 N802 + def SecondaryChannel(self) -> DBusStr: """ Secondary channel to be used. Primary channel is always set to "1M" except when "Coded" is set. """ return self._secondary_channel - @SecondaryChannel.setter # type: ignore - @no_type_check - def SecondaryChannel(self, channel: "q"): # type: ignore # noqa: F821 N802 + @SecondaryChannel.setter # type: ignore[no-redef] + def SecondaryChannel(self, channel: DBusStr) -> None: self._secondary_channel = channel @dbus_property(disabled=True) - @no_type_check - def MinInterval(self) -> "u": # type: ignore # noqa: F821 N802 + def MinInterval(self) -> DBusUInt32: """ Minimum advertising interval to be used by the advertising set, in milliseconds. Acceptable values are in the range [20ms, 10,485s]. @@ -370,14 +355,12 @@ def MinInterval(self) -> "u": # type: ignore # noqa: F821 N802 """ return self._min_interval - @MinInterval.setter # type: ignore - @no_type_check - def MinInterval(self, milliseconds: "u"): # type: ignore # noqa: F821 N802 + @MinInterval.setter # type: ignore[no-redef] + def MinInterval(self, milliseconds: DBusUInt32) -> None: self._min_interval = milliseconds @dbus_property(disabled=True) - @no_type_check - def MaxInterval(self) -> "u": # type: ignore # noqa: F821 N802 + def MaxInterval(self) -> DBusUInt32: """ Maximum advertising interval to be used by the advertising set, in milliseconds. Acceptable values are in the range [20ms, 10,485s]. @@ -386,14 +369,12 @@ def MaxInterval(self) -> "u": # type: ignore # noqa: F821 N802 """ return self._max_interval - @MaxInterval.setter # type: ignore - @no_type_check - def MaxInterval(self, milliseconds: "u"): # type: ignore # noqa: F821 N802 + @MaxInterval.setter # type: ignore[no-redef] + def MaxInterval(self, milliseconds: DBusUInt32) -> None: self._max_interval = milliseconds @dbus_property(disabled=True) - @no_type_check - def TxPower(self) -> "n": # type: ignore # noqa: F821 N802 + def TxPower(self) -> DBusInt16: """ Requested transmission power of this advertising set. The provided value is used only if the "CanSetTxPower" feature is enabled on the org.bluez.LEAdvertisingManager(5). @@ -401,9 +382,8 @@ def TxPower(self) -> "n": # type: ignore # noqa: F821 N802 """ return self._tx_power - @TxPower.setter # type: ignore - @no_type_check - def TxPower(self, dbm: "n"): # type: ignore # noqa: F821 N802 + @TxPower.setter # type: ignore[no-redef] + def TxPower(self, dbm: DBusInt16) -> None: self._tx_power = dbm @@ -482,17 +462,17 @@ def message(self) -> PybricksBroadcastData | None: """The data contained in this broadcast message.""" if self.LEGO_CID in self._manufacturer_data: channel, value = decode_message( - self._manufacturer_data[self.LEGO_CID].value # type: ignore + self._manufacturer_data[self.LEGO_CID].value ) return value else: return None @message.setter - def message(self, value: PybricksBroadcastData): + def message(self, value: PybricksBroadcastData) -> None: value = value if isinstance(value, tuple) else (value,) message = encode_message(self.channel, *value) - self._manufacturer_data[self.LEGO_CID] = Variant("ay", message) # type: ignore + self._manufacturer_data[self.LEGO_CID] = Variant("ay", message) # Notify BlueZ of the changed manufacturer data so the advertisement is updated self.emit_properties_changed( changed_properties={"ManufacturerData": self._manufacturer_data} From b6fe40c278dc7819638fea8ca98c4893985256f3 Mon Sep 17 00:00:00 2001 From: Frederik Leonhardt Date: Sun, 22 Feb 2026 00:18:45 +1300 Subject: [PATCH 6/9] Add more type hints --- tests/fixtures/bluetooth.py | 8 +++++--- tests/test_adapters.py | 8 ++++---- tests/test_advertisement.py | 38 ++++++++++++++++++++++++++----------- tests/test_broadcaster.py | 27 ++++++++++++++++++-------- tests/test_messages.py | 38 ++++++++++++++++++------------------- tests/test_observer.py | 26 ++++++++++++++++++------- tests/test_vhub.py | 22 +++++++++++---------- 7 files changed, 105 insertions(+), 62 deletions(-) diff --git a/tests/fixtures/bluetooth.py b/tests/fixtures/bluetooth.py index a17b36b..f370d36 100644 --- a/tests/fixtures/bluetooth.py +++ b/tests/fixtures/bluetooth.py @@ -1,11 +1,13 @@ import pytest +from _pytest.config import Config, Parser from dbus_fast.aio import MessageBus, ProxyObject from dbus_fast.constants import BusType from pb_ble.bluezdbus import get_adapter, get_adapter_details +from pb_ble.bluezdbus.adapters import AdapterDetailsExt -def pytest_addoption(parser): +def pytest_addoption(parser: Parser): parser.addoption( "--adapter", action="store", @@ -15,7 +17,7 @@ def pytest_addoption(parser): @pytest.fixture -def adapter_name(pytestconfig) -> str: +def adapter_name(pytestconfig: Config) -> str: return pytestconfig.getoption("adapter") @@ -32,6 +34,6 @@ async def adapter(message_bus: MessageBus, adapter_name: str) -> ProxyObject: @pytest.fixture -async def adapter_details(adapter_name: str): +async def adapter_details(adapter_name: str) -> AdapterDetailsExt: _, details = await get_adapter_details(adapter_name) return details diff --git a/tests/test_adapters.py b/tests/test_adapters.py index 9c31899..4fcb7ef 100644 --- a/tests/test_adapters.py +++ b/tests/test_adapters.py @@ -1,10 +1,10 @@ import pytest -from dbus_fast.aio import ProxyObject +from dbus_fast.aio import MessageBus, ProxyObject from pb_ble.bluezdbus import get_adapter -async def test_get_default_adapter(message_bus, adapter_name): +async def test_get_default_adapter(message_bus: MessageBus, adapter_name: str) -> None: if adapter_name != "hci0": pytest.skip(reason=f"Bluetooth adapter name '{adapter_name}' is not default") @@ -13,12 +13,12 @@ async def test_get_default_adapter(message_bus, adapter_name): assert adapter.path == "/org/bluez/hci0" -async def test_get_adapter_by_name(message_bus, adapter_name): +async def test_get_adapter_by_name(message_bus: MessageBus, adapter_name: str) -> None: adapter: ProxyObject = await get_adapter(message_bus, adapter_name) assert adapter.bus_name == "org.bluez" assert adapter.path == f"/org/bluez/{adapter_name}" -async def test_get_adapter_unavailable(message_bus): +async def test_get_adapter_unavailable(message_bus: MessageBus) -> None: with pytest.raises(ValueError): await get_adapter(message_bus, "non-existent") diff --git a/tests/test_advertisement.py b/tests/test_advertisement.py index 74571be..cd51ecf 100644 --- a/tests/test_advertisement.py +++ b/tests/test_advertisement.py @@ -1,8 +1,12 @@ +from typing import AsyncGenerator + import pytest import pytest_asyncio +from dbus_fast.aio import ProxyObject from dbus_fast.errors import DBusError from pb_ble.bluezdbus import LEAdvertisement, LEAdvertisingManager +from pb_ble.bluezdbus.adapters import AdapterDetailsExt from pb_ble.bluezdbus.advertisement import Include, Type @@ -15,19 +19,19 @@ class TestLEAdvertising: ("test", 1000, "/org/bluez/test/advertisement1000"), ], ) - def test_advertisement_path(self, name, index, path): + def test_advertisement_path(self, name: str, index: int, path: str) -> None: adv = LEAdvertisement( advertising_type=Type.BROADCAST, local_name=name, index=index ) assert adv.path == path - def test_invalid_index(self): + def test_invalid_index(self) -> None: with pytest.raises(ValueError): LEAdvertisement( advertising_type=Type.BROADCAST, local_name="anything", index=-1 ) - def test_includes(self): + def test_includes(self) -> None: adv = LEAdvertisement( advertising_type=Type.BROADCAST, local_name="test", @@ -39,18 +43,22 @@ def test_includes(self): class TestLEAdvertisingManager: @pytest_asyncio.fixture(autouse=True) - async def require_advertise(self, adapter_details, adapter_name): + async def require_advertise( + self, adapter_details: AdapterDetailsExt, adapter_name: str + ) -> None: if not adapter_details["advertise"]: pytest.skip( reason=f"Bluetooth adapter '{adapter_name}' does not support BLE advertising" ) @pytest.fixture - def adv_manager(self, adapter): + def adv_manager(self, adapter: ProxyObject) -> LEAdvertisingManager: return LEAdvertisingManager(adapter) @pytest_asyncio.fixture - async def adv(self, adv_manager): + async def adv( + self, adv_manager: LEAdvertisingManager + ) -> AsyncGenerator[LEAdvertisement, None]: adv = LEAdvertisement(advertising_type=Type.BROADCAST, local_name="myadv") yield adv try: @@ -58,7 +66,7 @@ async def adv(self, adv_manager): except DBusError: pass - async def test_create(self, adapter): + async def test_create(self, adapter: ProxyObject) -> None: adv_manager = LEAdvertisingManager(adapter) assert isinstance(await adv_manager.active_instances(), int) @@ -68,18 +76,26 @@ async def test_create(self, adapter): assert isinstance(await adv_manager.supported_capabilities(), dict) assert isinstance(await adv_manager.supported_features(), list) - async def test_register_advertisement(self, adv_manager, adv): + async def test_register_advertisement( + self, adv_manager: LEAdvertisingManager, adv: LEAdvertisement + ) -> None: await adv_manager.register_advertisement(adv) - async def test_unregister_advertisement(self, adv_manager, adv): + async def test_unregister_advertisement( + self, adv_manager: LEAdvertisingManager, adv: LEAdvertisement + ) -> None: await adv_manager.register_advertisement(adv) await adv_manager.unregister_advertisement(adv) - async def test_unregister_advertisement_by_path(self, adv_manager, adv): + async def test_unregister_advertisement_by_path( + self, adv_manager: LEAdvertisingManager, adv: LEAdvertisement + ) -> None: await adv_manager.register_advertisement(adv) await adv_manager.unregister_advertisement(adv.path) - async def test_unregister_advertisement_does_not_exist(self, adv_manager): + async def test_unregister_advertisement_does_not_exist( + self, adv_manager: LEAdvertisingManager + ) -> None: # TODO check exception type and message with pytest.raises(DBusError): await adv_manager.unregister_advertisement("/some/path") diff --git a/tests/test_broadcaster.py b/tests/test_broadcaster.py index a7ec221..057399b 100644 --- a/tests/test_broadcaster.py +++ b/tests/test_broadcaster.py @@ -1,18 +1,23 @@ from asyncio import Lock, Semaphore +from typing import AsyncGenerator import pytest import pytest_asyncio +from dbus_fast.aio import MessageBus, ProxyObject from pb_ble.bluezdbus import ( BlueZBroadcaster, BroadcastAdvertisement, ) +from pb_ble.bluezdbus.adapters import AdapterDetailsExt lock = Lock() @pytest_asyncio.fixture(autouse=True) -async def require_advertise(adapter_details, adapter_name): +async def require_advertise( + adapter_details: AdapterDetailsExt, adapter_name: str +) -> None: if not adapter_details["advertise"]: pytest.skip( reason=f"Bluetooth adapter '{adapter_name}' does not support BLE advertising" @@ -20,7 +25,9 @@ async def require_advertise(adapter_details, adapter_name): @pytest_asyncio.fixture -async def broadcaster(message_bus, adapter): +async def broadcaster( + message_bus: MessageBus, adapter: ProxyObject +) -> AsyncGenerator[BlueZBroadcaster, None]: broadcaster = BlueZBroadcaster(bus=message_bus, adapter=adapter, name="vhub") await lock.acquire() yield broadcaster @@ -29,19 +36,23 @@ async def broadcaster(message_bus, adapter): @pytest_asyncio.fixture(autouse=False) -async def broadcast_lock(): +async def broadcast_lock() -> AsyncGenerator[None, None]: await lock.acquire() yield lock.release() class TestBlueZBroadcaster: - def test_create_broadcaster(self, message_bus, adapter): + def test_create_broadcaster( + self, message_bus: MessageBus, adapter: ProxyObject + ) -> None: broadcaster = BlueZBroadcaster(bus=message_bus, adapter=adapter, name="vhub") assert broadcaster is not None assert len(broadcaster.advertisements) == 0 - async def test_stop_broadcaster(self, message_bus, adapter): + async def test_stop_broadcaster( + self, message_bus: MessageBus, adapter: ProxyObject + ) -> None: # GIVEN an advertisement on the bus is known to the broadcaster broadcaster = BlueZBroadcaster(bus=message_bus, adapter=adapter, name="vhub") adv = BroadcastAdvertisement("vhub") @@ -55,7 +66,7 @@ async def test_stop_broadcaster(self, message_bus, adapter): # THEN the advertisements are removed assert len(broadcaster.advertisements) == 0 - async def test_broadcast(self, broadcaster: BlueZBroadcaster): + async def test_broadcast(self, broadcaster: BlueZBroadcaster) -> None: # GIVEN a broadcast adv = BroadcastAdvertisement( broadcaster.name, @@ -69,7 +80,7 @@ async def test_broadcast(self, broadcaster: BlueZBroadcaster): assert adv.path in broadcaster.advertisements @pytest.mark.skip_on_bluez_mock("Does not implement release timeout") - async def test_broadcast_release(self, broadcaster: BlueZBroadcaster): + async def test_broadcast_release(self, broadcaster: BlueZBroadcaster) -> None: # GIVEN a broadcast semaphore = Semaphore(1) adv = BroadcastAdvertisement( @@ -97,7 +108,7 @@ async def test_broadcast_release(self, broadcaster: BlueZBroadcaster): # TODO: test that it's unexported from the bus - async def test_broadcast_twice(self, broadcaster: BlueZBroadcaster): + async def test_broadcast_twice(self, broadcaster: BlueZBroadcaster) -> None: # GIVEN a broadcast adv = BroadcastAdvertisement(broadcaster.name) diff --git a/tests/test_messages.py b/tests/test_messages.py index 2e29f26..d95ac25 100644 --- a/tests/test_messages.py +++ b/tests/test_messages.py @@ -6,7 +6,7 @@ class TestPybricksBleDecodeMessage: # h.ble.broadcast(5) - def test_decode_message_single_object(self): + def test_decode_message_single_object(self) -> None: # channel 200 # single object marker # int8: 5 @@ -19,7 +19,7 @@ def test_decode_message_single_object(self): assert data == 5 # h.ble.broadcast((5,)) - def test_decode_message_single_object_tuple(self): + def test_decode_message_single_object_tuple(self) -> None: # channel 200 # int8: 5 message = b"\xc8\x61\x05" @@ -32,7 +32,7 @@ def test_decode_message_single_object_tuple(self): assert data == (5,) # h.ble.broadcast((5,)) - def test_decode_message_single_object_tuple_0(self): + def test_decode_message_single_object_tuple_0(self) -> None: # channel 0 # int8: 5 message = b"\x00a\x05" @@ -44,7 +44,7 @@ def test_decode_message_single_object_tuple_0(self): assert data == (5,) - def test_decode_message_int8_int16_int32(self): + def test_decode_message_int8_int16_int32(self) -> None: # channel: 200 # str: '8_16_32' # int8: 127 @@ -64,7 +64,7 @@ def test_decode_message_int8_int16_int32(self): assert data[3] == 32_767 assert data[4] == 32_876 - def test_decode_message_int32_max(self): + def test_decode_message_int32_max(self) -> None: # channel: 200 # str: 'int32' # int32: 536_870_912 @@ -82,7 +82,7 @@ def test_decode_message_int32_max(self): data[2] == 1_073_741_823 ) # max int32 in micropython seems to be actually int31 - def test_decode_message_float(self): + def test_decode_message_float(self) -> None: # channel: 200 # str: 'float' # float32: PI @@ -96,7 +96,7 @@ def test_decode_message_float(self): assert data[0] == "float" assert data[1] == 3.1415927410125732 # float32 pi - def test_decode_message_str_bool(self): + def test_decode_message_str_bool(self) -> None: # channel: 200 # str: 'NTF' # bool: True @@ -112,7 +112,7 @@ def test_decode_message_str_bool(self): assert data[1] is True assert data[2] is False - def test_decode_message_bytes(self): + def test_decode_message_bytes(self) -> None: # channel: 200 # str: 'bytes' # bytes: b'\x00\xc4\x81' @@ -126,7 +126,7 @@ def test_decode_message_bytes(self): assert data[0] == "bytes" assert data[1] == b"\x00\xc4\x81" - def test_decode_message_empty(self): + def test_decode_message_empty(self) -> None: # channel: 200 message = b"\xc8" channel, data = decode_message(message) @@ -137,46 +137,46 @@ def test_decode_message_empty(self): class TestPybricksBleEncodeMessage: - def test_encode_message_single_object(self): + def test_encode_message_single_object(self) -> None: data = encode_message(200, 5) assert data == b"\xc8\x00\x61\x05" @pytest.mark.skip("Encoding single-object tuples is not supported") - def test_encode_message_single_object_tuple(self): + def test_encode_message_single_object_tuple(self) -> None: data = encode_message(200, (1,)) # type: ignore[arg-type] assert data == b"\xc8\x61\x01" - def test_encode_message_int8_int16_int32(self): + def test_encode_message_int8_int16_int32(self) -> None: data = encode_message(200, "8_16_32", 127, 128, 32_767, 32_876) assert data == b"\xc8\xa78_16_32a\x7fb\x80\x00b\xff\x7fdl\x80\x00\x00" - def test_encode_message_int32_max(self): + def test_encode_message_int32_max(self) -> None: data = encode_message(200, "int32", 536_870_912, 1_073_741_823) assert data == b"\xc8\xa5int32d\x00\x00\x00 d\xff\xff\xff?" - def test_encode_message_float(self): + def test_encode_message_float(self) -> None: data = encode_message(0, "float", 3.1415927410125732) # float32 pi assert data == b"\x00\xa5float\x84\xdb\x0fI@" - def test_encode_message_str_bool(self): + def test_encode_message_str_bool(self) -> None: data = encode_message(200, "NTF", True, False) assert data == b"\xc8\xa3NTF @" - def test_encode_message_bytes(self): + def test_encode_message_bytes(self) -> None: data = encode_message(200, "bytes", b"\x00\xc4\x81") assert data == b"\xc8\xa5bytes\xc3\x00\xc4\x81" - def test_encode_message_empty(self): + def test_encode_message_empty(self) -> None: data = encode_message(200) assert data == b"\xc8" class TestPybricksBlePnpId: - def test_pack_pnp_id(self): + def test_pack_pnp_id(self) -> None: pnp_id = pack_pnp_id(0x00, 0x00, vendor_id_type="BT", vendor_id=LEGO_CID) assert pnp_id == b"\x01\x97\x03\x00\x00\x00\x00" - def test_unpack_pnp_id(self): + def test_unpack_pnp_id(self) -> None: pnp_id = pack_pnp_id(0x00, 0x00, vendor_id_type="BT", vendor_id=LEGO_CID) (vid_type, vid, pid, rev) = unpack_pnp_id(pnp_id) assert vid_type == "BT" diff --git a/tests/test_observer.py b/tests/test_observer.py index b8a1b33..a0a5bf8 100644 --- a/tests/test_observer.py +++ b/tests/test_observer.py @@ -1,5 +1,9 @@ +from typing import AsyncGenerator + import pytest import pytest_asyncio +from bluetooth_adapters import AdapterDetails +from dbus_fast.aio import MessageBus, ProxyObject from pb_ble.bluezdbus import ( BlueZPybricksObserver, @@ -12,18 +16,20 @@ def get_adapter1(adapter): class TestPassiveBlueZObserver: @pytest_asyncio.fixture(autouse=True) - async def require_passive_scan(sef, adapter_details, adapter_name): + async def require_passive_scan( + sef, adapter_details: AdapterDetails, adapter_name: str + ) -> None: if not adapter_details["passive_scan"]: pytest.skip( reason=f"Bluetooth adapter '{adapter_name}' does not support BLE passive scanning" ) @pytest_asyncio.fixture() - async def observer(self): + async def observer(self) -> AsyncGenerator[BlueZPybricksObserver, None]: async with BlueZPybricksObserver(scanning_mode="passive") as observer: yield observer - def test_create_observer(self, message_bus): + def test_create_observer(self, message_bus: MessageBus) -> None: observer = BlueZPybricksObserver( scanning_mode="passive", channels=[0], @@ -35,7 +41,9 @@ def test_create_observer(self, message_bus): assert observer.rssi_threshold == -100 assert observer.device_pattern == "Pybricks" - async def test_observe(self, adapter, observer: BlueZPybricksObserver): + async def test_observe( + self, adapter: ProxyObject, observer: BlueZPybricksObserver + ) -> None: # WHEN a channel is observed data = observer.observe(0) @@ -49,11 +57,13 @@ async def test_observe(self, adapter, observer: BlueZPybricksObserver): class TestActiveBlueZObserver: @pytest_asyncio.fixture() - async def observer(self): + async def observer(self) -> AsyncGenerator[BlueZPybricksObserver, None]: async with BlueZPybricksObserver(scanning_mode="active") as observer: yield observer - async def test_create_observer(self, message_bus, adapter): + async def test_create_observer( + self, message_bus: MessageBus, adapter: ProxyObject + ) -> None: observer = BlueZPybricksObserver( scanning_mode="active", channels=[0], @@ -64,7 +74,9 @@ async def test_create_observer(self, message_bus, adapter): assert observer.rssi_threshold is None assert observer.device_pattern == "Name" - async def test_observe(self, adapter, observer: BlueZPybricksObserver): + async def test_observe( + self, adapter: ProxyObject, observer: BlueZPybricksObserver + ) -> None: # WHEN a channel is observed data = observer.observe(0) diff --git a/tests/test_vhub.py b/tests/test_vhub.py index b011087..7baca9b 100644 --- a/tests/test_vhub.py +++ b/tests/test_vhub.py @@ -1,13 +1,15 @@ import pytest import pytest_asyncio +from dbus_fast.aio import ProxyObject from pytest_mock import MockerFixture from pb_ble import get_virtual_ble +from pb_ble.bluezdbus.adapters import AdapterDetailsExt from pb_ble.bluezdbus.observer import ObservedAdvertisement @pytest_asyncio.fixture(autouse=True) -async def require_ble(adapter_details, adapter_name): +async def require_ble(adapter_details: AdapterDetailsExt, adapter_name: str) -> None: if not adapter_details["advertise"] or not adapter_details["passive_scan"]: pytest.skip( reason=f"Bluetooth adapter '{adapter_name}' does not support BLE capabilities" @@ -15,25 +17,25 @@ async def require_ble(adapter_details, adapter_name): class TestVirtualBLE: - async def test_create_vble(self, adapter): + async def test_create_vble(self, adapter: ProxyObject) -> None: ble = await get_virtual_ble(broadcast_channel=1, observe_channels=[2]) assert ble is not None assert not ble._broadcaster.is_broadcasting() - async def test_observe_none(self): + async def test_observe_none(self) -> None: ble = await get_virtual_ble(broadcast_channel=1, observe_channels=[2]) data = ble.observe(2) assert data is None assert not ble._broadcaster.is_broadcasting() - async def test_observe_single(self, mocker: MockerFixture): + async def test_observe_single(self, mocker: MockerFixture) -> None: ble = await get_virtual_ble(broadcast_channel=1, observe_channels=[2]) observe_mock = mocker.patch.object(ble._observer, "observe") observe_mock.return_value = ObservedAdvertisement(data=b"val", rssi=0) data = ble.observe(2) assert data == b"val" - async def test_observe_multiple(self, mocker: MockerFixture): + async def test_observe_multiple(self, mocker: MockerFixture) -> None: ble = await get_virtual_ble(broadcast_channel=1, observe_channels=[2]) observe_mock = mocker.patch.object(ble._observer, "observe") observe_mock.return_value = ObservedAdvertisement( @@ -42,31 +44,31 @@ async def test_observe_multiple(self, mocker: MockerFixture): data = ble.observe(2) assert data == (b"val", 0, True, 1.0, "str") - async def test_broadcast_single(self): + async def test_broadcast_single(self) -> None: ble = await get_virtual_ble(broadcast_channel=1, observe_channels=[2]) await ble.broadcast(42) assert ble._broadcaster.is_broadcasting() assert ble._adv.message == 42 - async def test_broadcast_multiple(self): + async def test_broadcast_multiple(self) -> None: ble = await get_virtual_ble(broadcast_channel=1, observe_channels=[2]) await ble.broadcast(42, 24) assert ble._broadcaster.is_broadcasting() assert ble._adv.message == (42, 24) - async def test_broadcast_none(self): + async def test_broadcast_none(self) -> None: ble = await get_virtual_ble(broadcast_channel=1, observe_channels=[2]) await ble.broadcast(None) assert not ble._broadcaster.is_broadcasting() - async def test_broadcast_start_stop(self): + async def test_broadcast_start_stop(self) -> None: ble = await get_virtual_ble(broadcast_channel=1, observe_channels=[2]) await ble.broadcast(42) assert ble._broadcaster.is_broadcasting() await ble.broadcast(None) assert not ble._broadcaster.is_broadcasting() - async def test_context(self): + async def test_context(self) -> None: async with await get_virtual_ble( broadcast_channel=1, observe_channels=[2] ) as ble: From 8c77820b0fc3f9058f35bc6f56818192c5b497b6 Mon Sep 17 00:00:00 2001 From: Frederik Leonhardt Date: Sun, 22 Feb 2026 20:38:39 +1300 Subject: [PATCH 7/9] Export additional type alias symbols --- src/pb_ble/__init__.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/pb_ble/__init__.py b/src/pb_ble/__init__.py index ccd83a5..e3bcdd7 100644 --- a/src/pb_ble/__init__.py +++ b/src/pb_ble/__init__.py @@ -62,6 +62,8 @@ "get_virtual_ble", "VirtualBLE", "PybricksBroadcast", + "PybricksBroadcastData", + "PybricksBroadcastValue", # Submodules "vhub", "bluezdbus", From f5adfa59ef5297129d825ba1e264ae9d5c9bd55a Mon Sep 17 00:00:00 2001 From: Frederik Leonhardt Date: Sun, 22 Feb 2026 20:38:54 +1300 Subject: [PATCH 8/9] Update pdoc to 16.x --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 320b1a5..c23de09 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -37,7 +37,7 @@ dev = [ "types-cachetools ~= 5.3", "types-setuptools", ] -docs = ["pdoc ~= 15.0"] +docs = ["pdoc ~= 16.0"] [project.urls] Repository = "https://github.com/fkleon/pybricks-ble" From d19d0b948b8a24bf05a1e82c10dc7ab6d04f469c Mon Sep 17 00:00:00 2001 From: Frederik Leonhardt Date: Sun, 22 Feb 2026 21:02:07 +1300 Subject: [PATCH 9/9] Add Protocol for `org.bluez.LEAdvertisingManager1` interface --- src/pb_ble/bluezdbus/advertisement.py | 92 ++++++++++++++++++++++++--- 1 file changed, 82 insertions(+), 10 deletions(-) diff --git a/src/pb_ble/bluezdbus/advertisement.py b/src/pb_ble/bluezdbus/advertisement.py index 76e56d6..3d16790 100644 --- a/src/pb_ble/bluezdbus/advertisement.py +++ b/src/pb_ble/bluezdbus/advertisement.py @@ -14,6 +14,8 @@ Annotated, Any, Callable, + Protocol, + cast, overload, ) @@ -482,6 +484,69 @@ def __str__(self): return f"PybricksBroadcastAdvertisement(channel={self.channel}, data={self.message!r}, timeout={self._timeout})" +class LEAdvertisingManager1(Protocol): + """Protocol for the 'org.bluez.LEAdvertisingManager1' proxy interface.""" + + async def call_register_advertisement(self, path: str, options: dict) -> None: + """Registers an advertisement object to be sent over the LE Advertising channel. + + The service must implement org.bluez.LEAdvertisement(5) interface. + + Possible errors: + + org.bluez.Error.InvalidArguments + Indicates that the object has invalid or conflicting properties. + org.bluez.Error.AlreadyExists + Indicates the object is already registered. + org.bluez.Error.InvalidLength + Indicates that the data provided generates a data packet which is too long. + org.bluez.Error.NotPermitted + Indicates the maximum number of advertisement instances has been reached. + """ + ... + + async def call_unregister_advertisement(self, path: str) -> None: + """Unregisters an advertisement that has been previously registered using RegisterAdvertisement(). + + The object path parameter must match the same value that has been used on registration. + + Possible errors: + + org.bluez.Error.InvalidArguments + org.bluez.Error.DoesNotExist + """ + ... + + async def get_active_instances(self) -> int: + """Number of active advertising instances.""" + ... + + async def get_supported_instances(self) -> int: + """Number of available advertising instances.""" + ... + + async def get_supported_includes(self) -> list[Include]: + """List of supported system includes.""" + ... + + async def get_supported_secondary_channels(self) -> list[SecondaryChannel]: + """List of supported Secondary channels. + + Secondary channels can be used to advertise with the corresponding PHY. + """ + ... + + async def get_supported_capabilities(self) -> dict[Capability, Any]: + """Enumerates Advertising-related controller capabilities useful to the client.""" + ... + + async def get_supported_features(self) -> list[Feature]: + """List of supported platform features. + + If no features are available on the platform, the SupportedFeatures array will be empty.""" + ... + + class LEAdvertisingManager: """ Client implementation of the `org.bluez.LEAdvertisementManager1` D-Bus interface. @@ -489,6 +554,8 @@ class LEAdvertisingManager: INTERFACE_NAME: str = "org.bluez.LEAdvertisingManager1" + _adv_manager: LEAdvertisingManager1 + def __init__( self, adapter: ProxyObject | None = None, @@ -497,7 +564,12 @@ def __init__( if adapter is None and adv_manager is None: raise ValueError("adapter or adv_manager required") - self._adv_manager = adv_manager or adapter.get_interface(self.INTERFACE_NAME) # type: ignore + if adv_manager: + self._adv_manager = cast(LEAdvertisingManager1, adv_manager) + elif adapter: + self._adv_manager = cast( + LEAdvertisingManager1, adapter.get_interface(self.INTERFACE_NAME) + ) async def register_advertisement( self, adv: LEAdvertisement, options: dict | None = None @@ -511,7 +583,7 @@ async def register_advertisement( :return: `None` """ options = options or {} - return await self._adv_manager.call_register_advertisement(adv.path, options) # type: ignore + return await self._adv_manager.call_register_advertisement(adv.path, options) @overload async def unregister_advertisement(self, adv: LEAdvertisement): ... @@ -526,34 +598,34 @@ async def unregister_advertisement(self, adv): :return: `None` """ if isinstance(adv, str): - return await self._adv_manager.call_unregister_advertisement(adv) # type: ignore + return await self._adv_manager.call_unregister_advertisement(adv) else: - return await self._adv_manager.call_unregister_advertisement(adv.path) # type: ignore + return await self._adv_manager.call_unregister_advertisement(adv.path) async def active_instances(self) -> int: """Number of active advertising instances.""" - return await self._adv_manager.get_active_instances() # type: ignore + return await self._adv_manager.get_active_instances() async def supported_instances(self) -> int: """Number of available advertising instances.""" - return await self._adv_manager.get_supported_instances() # type: ignore + return await self._adv_manager.get_supported_instances() async def supported_includes(self) -> list[Include]: """List of supported system includes.""" - return await self._adv_manager.get_supported_includes() # type: ignore + return await self._adv_manager.get_supported_includes() async def supported_secondary_channels(self) -> list[SecondaryChannel]: """List of supported Secondary channels. Secondary channels can be used to advertise with the corresponding PHY. """ - return await self._adv_manager.get_supported_secondary_channels() # type: ignore + return await self._adv_manager.get_supported_secondary_channels() async def supported_capabilities(self) -> dict[Capability, Any]: """Enumerates Advertising-related controller capabilities useful to the client.""" - return await self._adv_manager.get_supported_capabilities() # type: ignore + return await self._adv_manager.get_supported_capabilities() async def supported_features(self) -> list[Feature]: """List of supported platform features. If no features are available on the platform, the SupportedFeatures array will be empty. """ - return await self._adv_manager.get_supported_features() # type: ignore + return await self._adv_manager.get_supported_features()