pb_ble.vhub

  1import sys
  2from contextlib import AsyncExitStack
  3from typing import ClassVar, Sequence, cast
  4
  5from dbus_fast.aio import MessageBus, ProxyObject
  6from dbus_fast.constants import BusType
  7from pybricks.hubs import _common
  8
  9from .bluezdbus import (
 10    BlueZBroadcaster,
 11    BlueZPybricksObserver,
 12    PybricksBroadcastAdvertisement,
 13    get_adapter,
 14    get_adapter_details,
 15)
 16from .constants import (
 17    PYBRICKS_MAX_CHANNEL,
 18    PYBRICKS_MIN_CHANNEL,
 19    PybricksBroadcastData,
 20    PybricksBroadcastValue,
 21    ScanningMode,
 22)
 23
 24
 25class VirtualBLE(_common.BLE, AsyncExitStack):
 26    DEFAULT_DEVICE_NAME: ClassVar[str] = "pb_vhub"
 27    """The default device name to use in data broadcasts."""
 28    DEFAULT_DEVICE_VERSION: ClassVar[str] = "1.0"
 29    """The default device version to return from `version()`."""
 30
 31    _adv: PybricksBroadcastAdvertisement
 32    """The current data broadcast."""
 33    _device_version: str
 34    """The version string configured for this VirtualBLE device."""
 35
 36    _broadcaster: BlueZBroadcaster
 37    """The broadcaster to use."""
 38    _observer: BlueZPybricksObserver
 39    """The observer to use."""
 40
 41    def __init__(
 42        self,
 43        broadcaster: BlueZBroadcaster,
 44        observer: BlueZPybricksObserver,
 45        broadcast_channel: int,
 46        device_version: str = DEFAULT_DEVICE_VERSION,
 47    ):
 48        if (
 49            not isinstance(broadcast_channel, int)
 50            or PYBRICKS_MIN_CHANNEL < broadcast_channel > PYBRICKS_MAX_CHANNEL
 51        ):
 52            raise ValueError(
 53                f"Broadcast channel must be integer from {PYBRICKS_MIN_CHANNEL} to {PYBRICKS_MAX_CHANNEL}."
 54            )
 55
 56        super(AsyncExitStack, self).__init__()
 57
 58        self._device_version = device_version
 59        self._broadcaster = broadcaster
 60        self._observer = observer
 61
 62        self._adv = PybricksBroadcastAdvertisement(broadcaster.name, broadcast_channel)
 63
 64    async def __aenter__(self):
 65        try:
 66            await self.enter_async_context(self._broadcaster)
 67            await self.enter_async_context(self._observer)
 68        except Exception:
 69            if not await self.__aexit__(*sys.exc_info()):
 70                raise
 71        return self
 72
 73    async def broadcast(self, *data: PybricksBroadcastValue | None) -> None:  # type: ignore [override]
 74        if len(data) == 0:
 75            raise ValueError("Broadcast must be a value or tuple.")
 76        if None in data:
 77            await self._broadcaster.stop_broadcast(self._adv)
 78        else:
 79            if not self._broadcaster.is_broadcasting(self._adv):
 80                await self._broadcaster.broadcast(self._adv)
 81            self._adv.message = cast(PybricksBroadcastData, tuple(data))
 82
 83    def observe(self, channel: int) -> PybricksBroadcastData | None:
 84        advertisement = self._observer.observe(channel)
 85
 86        if advertisement is not None:
 87            return advertisement.data
 88        else:
 89            return None
 90
 91    def signal_strength(self, channel: int) -> int:
 92        advertisement = self._observer.observe(channel)
 93        return advertisement.rssi if advertisement is not None else -128
 94
 95    def version(self) -> str:
 96        return self._device_version
 97
 98
 99async def get_virtual_ble(
100    adapter_name: str | None = None,
101    device_name: str = VirtualBLE.DEFAULT_DEVICE_NAME,
102    broadcast_channel: int = 0,
103    observe_channels: Sequence[int] | None = None,
104    scanning_mode: ScanningMode = "passive",
105    device_filter: str | None = None,
106) -> VirtualBLE:
107    """
108    Creates a "virtual" Pybricks BLE radio that can be used to exchange
109    messages with other Pybricks Hubs via connectionless Bluetooth messaging.
110
111    The `VirtualBLE` object implements an interface very similar to the [common Pybricks BLE interface](https://github.com/pybricks/pybricks-api/blob/v3.5.0/src/pybricks/_common.py#L1331)
112    available on Pybricks Hubs. The crucial difference is that some methods
113    must be called asynchronously.
114
115    `get_virtual_ble()` should be used as an asynchronous context manager:
116
117    ```python
118    async with await get_virtual_ble(
119        broadcast_channel=2
120    ) as vble:
121        # Broadcast a random number on channel 2
122        val = random.randint(0, 3)
123        await vble.broadcast(val)
124        # Stop after 10 seconds
125        await asyncio.sleep(10)
126    ```
127
128    :param adapter_name: The Bluetooth adapter to use, defaults to `None`
129        (auto-discover default device).
130    :param device_name: The name of the hub. This may be used as local name
131        in the BLE advertisement data, defaults to `VirtualBLE.DEFAULT_DEVICE_NAME`.
132    :param broadcast_channel: A value from 0 to 255 indicating which channel
133        `VirtualBLE.broadcast()` will use, defaults to 0.
134    :param observe_channels: A list of channels to listen to for use
135        with `VirtualBLE.observe()`, defaults to `None` (all channels).
136    :param scanning_mode: The scanning mode to use for observing broadcasts,
137        defaults to `passive`.
138        - Passive scanning is the default and recommended mode.
139        However it is not supported by all devices.
140        - Active scanning is provided as a well-supported fallback.
141        It may negatively impact the power consumption of nearby BLE devices.
142    :param device_filter: Provides a mechanism to filter observed broadcasts
143        based on the custom name of the sending Pybricks Hub, defaults to `None` (no filter).
144        For example, set this to `Pybricks` to receive only broadcasts from Hubs
145        that have a name starting with "Pybricks".
146    :return: A `VirtualBLE` object which is loosely adhering to the Pybricks Hub
147        BLE interface.
148    """
149    bus: MessageBus = await MessageBus(bus_type=BusType.SYSTEM).connect()
150
151    # Find given adapter or default adapter
152    adapter_name, details = (
153        await get_adapter_details()
154        if adapter_name is None
155        else await get_adapter_details(adapter_name)
156    )
157    adapter: ProxyObject = await get_adapter(bus, adapter_name)
158
159    broadcaster = BlueZBroadcaster(bus=bus, adapter=adapter, name=device_name)
160    observer = BlueZPybricksObserver(
161        adapter_name=adapter_name,
162        scanning_mode=scanning_mode,
163        channels=observe_channels,
164        device_pattern=device_filter,
165    )
166
167    return VirtualBLE(
168        broadcaster=broadcaster,
169        observer=observer,
170        broadcast_channel=broadcast_channel,
171        device_version=str(details["hw_version"]),
172    )