From 318a9e324c68027c1d4fc340a1aa18594ca0a828 Mon Sep 17 00:00:00 2001 From: Frederik Leonhardt Date: Tue, 24 Jun 2025 23:17:04 +1200 Subject: [PATCH 1/3] Add some type hints to test fixture parameters --- tests/test_broadcaster.py | 6 +++--- tests/test_observer.py | 4 ++-- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/tests/test_broadcaster.py b/tests/test_broadcaster.py index 6060f8d..a7ec221 100644 --- a/tests/test_broadcaster.py +++ b/tests/test_broadcaster.py @@ -55,7 +55,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): + async def test_broadcast(self, broadcaster: BlueZBroadcaster): # GIVEN a broadcast adv = BroadcastAdvertisement( broadcaster.name, @@ -69,7 +69,7 @@ async def test_broadcast(self, broadcaster): assert adv.path in broadcaster.advertisements @pytest.mark.skip_on_bluez_mock("Does not implement release timeout") - async def test_broadcast_release(self, broadcaster): + async def test_broadcast_release(self, broadcaster: BlueZBroadcaster): # GIVEN a broadcast semaphore = Semaphore(1) adv = BroadcastAdvertisement( @@ -97,7 +97,7 @@ async def test_broadcast_release(self, broadcaster): # TODO: test that it's unexported from the bus - async def test_broadcast_twice(self, broadcaster): + async def test_broadcast_twice(self, broadcaster: BlueZBroadcaster): # GIVEN a broadcast adv = BroadcastAdvertisement(broadcaster.name) diff --git a/tests/test_observer.py b/tests/test_observer.py index 10b29e7..b8a1b33 100644 --- a/tests/test_observer.py +++ b/tests/test_observer.py @@ -35,7 +35,7 @@ def test_create_observer(self, message_bus): assert observer.rssi_threshold == -100 assert observer.device_pattern == "Pybricks" - async def test_observe(self, adapter, observer): + async def test_observe(self, adapter, observer: BlueZPybricksObserver): # WHEN a channel is observed data = observer.observe(0) @@ -64,7 +64,7 @@ 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): + async def test_observe(self, adapter, observer: BlueZPybricksObserver): # WHEN a channel is observed data = observer.observe(0) From 3507dd94369d906694854643f01ddcfb1f71233e Mon Sep 17 00:00:00 2001 From: Frederik Leonhardt Date: Tue, 24 Jun 2025 23:18:42 +1200 Subject: [PATCH 2/3] Add type hints for VirtualBLE private fields --- src/pb_ble/vhub.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/src/pb_ble/vhub.py b/src/pb_ble/vhub.py index a82ae90..3c01311 100644 --- a/src/pb_ble/vhub.py +++ b/src/pb_ble/vhub.py @@ -30,6 +30,13 @@ class VirtualBLE(_common.BLE, AsyncExitStack): _adv: PybricksBroadcastAdvertisement """The current data broadcast.""" + _device_version: str + """The version string configured for this VirtualBLE device.""" + + _broadcaster: BlueZBroadcaster + """The broadcaster to use.""" + _observer: BlueZPybricksObserver + """The observer to use.""" def __init__( self, From 264331332179e533db02904f73da12d06f31bf48 Mon Sep 17 00:00:00 2001 From: Frederik Leonhardt Date: Tue, 24 Jun 2025 23:19:55 +1200 Subject: [PATCH 3/3] vhub: support returning single values from observe() This matches the Pybricks behaviour, but requires updated type hints in the pybricks-api. Temporarily switch to a forked version of this package. --- pyproject.toml | 3 ++- src/pb_ble/vhub.py | 15 +++------------ tests/test_vhub.py | 18 ++++++++++++++++++ tools/pybricks_virtual_ble.py | 2 +- 4 files changed, 24 insertions(+), 14 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 24c5959..9d8bf97 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,7 +21,7 @@ dependencies = [ "bluetooth-adapters ~= 0.16", "cachetools ~= 5.3", "dbus-fast ~= 2.15", - "pybricks ~= 3.6.0", + "pybricks @ git+https://github.com/fkleon/pybricks-api@ble-type-fixes", ] [dependency-groups] @@ -31,6 +31,7 @@ dev = [ "mypy ~= 1.16.0", "pytest ~= 8.3", "pytest-asyncio ~= 1.0.0", + "pytest-mock ~= 3.14.1", "python-dbusmock ~= 0.34.3", "ruff ~= 0.11.8", "types-cachetools ~= 5.3", diff --git a/src/pb_ble/vhub.py b/src/pb_ble/vhub.py index 3c01311..908f793 100644 --- a/src/pb_ble/vhub.py +++ b/src/pb_ble/vhub.py @@ -1,6 +1,6 @@ import sys from contextlib import AsyncExitStack -from typing import ClassVar, Optional, Sequence, Tuple, Union, cast +from typing import ClassVar, Sequence, cast from dbus_fast.aio import MessageBus, ProxyObject from dbus_fast.constants import BusType @@ -80,20 +80,11 @@ async def broadcast(self, *data: PybricksBroadcastValue | None) -> None: # type await self._broadcaster.broadcast(self._adv) self._adv.message = cast(PybricksBroadcastData, tuple(data)) - def observe( - self, channel: int - ) -> Optional[Tuple[Union[bool, int, float, str, bytes], ...]]: + def observe(self, channel: int) -> PybricksBroadcastData | None: advertisement = self._observer.observe(channel) if advertisement is not None: - if isinstance(advertisement.data, tuple): - return advertisement.data - else: - # TODO: Pybricks does expose single-value broadcasts - # in a single-object tuple. However, that doesn't match - # the type signature of the observe() method. To adhere - # to the type signature, we only return wrapped values. - return (advertisement.data,) + return advertisement.data else: return None diff --git a/tests/test_vhub.py b/tests/test_vhub.py index cb29364..b011087 100644 --- a/tests/test_vhub.py +++ b/tests/test_vhub.py @@ -1,7 +1,9 @@ import pytest import pytest_asyncio +from pytest_mock import MockerFixture from pb_ble import get_virtual_ble +from pb_ble.bluezdbus.observer import ObservedAdvertisement @pytest_asyncio.fixture(autouse=True) @@ -24,6 +26,22 @@ async def test_observe_none(self): assert data is None assert not ble._broadcaster.is_broadcasting() + async def test_observe_single(self, mocker: MockerFixture): + 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): + 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", 0, True, 1.0, "str"), rssi=0 + ) + data = ble.observe(2) + assert data == (b"val", 0, True, 1.0, "str") + async def test_broadcast_single(self): ble = await get_virtual_ble(broadcast_channel=1, observe_channels=[2]) await ble.broadcast(42) diff --git a/tools/pybricks_virtual_ble.py b/tools/pybricks_virtual_ble.py index 0b211b3..f8ec3a9 100644 --- a/tools/pybricks_virtual_ble.py +++ b/tools/pybricks_virtual_ble.py @@ -21,7 +21,7 @@ async def observe(vble: _common.BLE, observe_channel: int, interval: float = 1.0 data = vble.observe(observe_channel) rssi = vble.signal_strength(observe_channel) if data: - print(f"Observation: '{data}' [{rssi} dBm]") + print(f"Observation: '{data!r}' [{rssi} dBm]") async def broadcast(vble: _common.BLE, interval: float = 10.0):