pb_ble

100async def get_virtual_ble(
101    adapter_name: str | None = None,
102    device_name: str = VirtualBLE.DEFAULT_DEVICE_NAME,
103    broadcast_channel: int = 0,
104    observe_channels: Sequence[int] | None = None,
105    scanning_mode: ScanningMode = "passive",
106    device_filter: str | None = None,
107) -> VirtualBLE:
108    """
109    Creates a "virtual" Pybricks BLE radio that can be used to exchange
110    messages with other Pybricks Hubs via connectionless Bluetooth messaging.
111
112    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)
113    available on Pybricks Hubs. The crucial difference is that some methods
114    must be called asynchronously.
115
116    `get_virtual_ble()` should be used as an asynchronous context manager:
117
118    ```python
119    async with await get_virtual_ble(
120        broadcast_channel=2
121    ) as vble:
122        # Broadcast a random number on channel 2
123        val = random.randint(0, 3)
124        await vble.broadcast(val)
125        # Stop after 10 seconds
126        await asyncio.sleep(10)
127    ```
128
129    :param adapter_name: The Bluetooth adapter to use, defaults to `None`
130        (auto-discover default device).
131    :param device_name: The name of the hub. This may be used as local name
132        in the BLE advertisement data, defaults to `VirtualBLE.DEFAULT_DEVICE_NAME`.
133    :param broadcast_channel: A value from 0 to 255 indicating which channel
134        `VirtualBLE.broadcast()` will use, defaults to 0.
135    :param observe_channels: A list of channels to listen to for use
136        with `VirtualBLE.observe()`, defaults to `None` (all channels).
137    :param scanning_mode: The scanning mode to use for observing broadcasts,
138        defaults to `passive`.
139        - Passive scanning is the default and recommended mode.
140        However it is not supported by all devices.
141        - Active scanning is provided as a well-supported fallback.
142        It may negatively impact the power consumption of nearby BLE devices.
143    :param device_filter: Provides a mechanism to filter observed broadcasts
144        based on the custom name of the sending Pybricks Hub, defaults to `None` (no filter).
145        For example, set this to `Pybricks` to receive only broadcasts from Hubs
146        that have a name starting with "Pybricks".
147    :return: A `VirtualBLE` object which is loosely adhering to the Pybricks Hub
148        BLE interface.
149    """
150    bus: MessageBus = await MessageBus(bus_type=BusType.SYSTEM).connect()
151
152    # Find given adapter or default adapter
153    adapter_name, details = (
154        await get_adapter_details()
155        if adapter_name is None
156        else await get_adapter_details(adapter_name)
157    )
158    adapter: ProxyObject = await get_adapter(bus, adapter_name)
159
160    broadcaster = BlueZBroadcaster(bus=bus, adapter=adapter, name=device_name)
161    observer = BlueZPybricksObserver(
162        adapter_name=adapter_name,
163        scanning_mode=scanning_mode,
164        channels=observe_channels,
165        device_pattern=device_filter,
166    )
167
168    return VirtualBLE(
169        broadcaster=broadcaster,
170        observer=observer,
171        broadcast_channel=broadcast_channel,
172        device_version=str(details["hw_version"]),
173    )

26class VirtualBLE(_common.BLE, AsyncExitStack):
27    DEFAULT_DEVICE_NAME: ClassVar[str] = "pb_vhub"
28    """The default device name to use in data broadcasts."""
29    DEFAULT_DEVICE_VERSION: ClassVar[str] = "1.0"
30    """The default device version to return from `version()`."""
31
32    _adv: PybricksBroadcastAdvertisement
33    """The current data broadcast."""
34    _device_version: str
35    """The version string configured for this VirtualBLE device."""
36
37    _broadcaster: BlueZBroadcaster
38    """The broadcaster to use."""
39    _observer: BlueZPybricksObserver
40    """The observer to use."""
41
42    def __init__(
43        self,
44        broadcaster: BlueZBroadcaster,
45        observer: BlueZPybricksObserver,
46        broadcast_channel: int,
47        device_version: str = DEFAULT_DEVICE_VERSION,
48    ):
49        if (
50            not isinstance(broadcast_channel, int)
51            or PYBRICKS_MIN_CHANNEL < broadcast_channel > PYBRICKS_MAX_CHANNEL
52        ):
53            raise ValueError(
54                f"Broadcast channel must be integer from {PYBRICKS_MIN_CHANNEL} to {PYBRICKS_MAX_CHANNEL}."
55            )
56
57        super(AsyncExitStack, self).__init__()
58
59        self._device_version = device_version
60        self._broadcaster = broadcaster
61        self._observer = observer
62
63        self._adv = PybricksBroadcastAdvertisement(broadcaster.name, broadcast_channel)
64
65    async def __aenter__(self):
66        try:
67            await self.enter_async_context(self._broadcaster)
68            await self.enter_async_context(self._observer)
69        except Exception:
70            if not await self.__aexit__(*sys.exc_info()):
71                raise
72        return self
73
74    async def broadcast(self, *data: PybricksBroadcastValue | None) -> None:  # type: ignore [override]
75        if len(data) == 0:
76            raise ValueError("Broadcast must be a value or tuple.")
77        if None in data:
78            await self._broadcaster.stop_broadcast(self._adv)
79        else:
80            if not self._broadcaster.is_broadcasting(self._adv):
81                await self._broadcaster.broadcast(self._adv)
82            self._adv.message = cast(PybricksBroadcastData, tuple(data))
83
84    def observe(self, channel: int) -> PybricksBroadcastData | None:
85        advertisement = self._observer.observe(channel)
86
87        if advertisement is not None:
88            return advertisement.data
89        else:
90            return None
91
92    def signal_strength(self, channel: int) -> int:
93        advertisement = self._observer.observe(channel)
94        return advertisement.rssi if advertisement is not None else -128
95
96    def version(self) -> str:
97        return self._device_version

27class PybricksBroadcast(NamedTuple):
28    """
29    Data structure for a Pybricks broadcast.
30    """
31
32    channel: int
33    """The broadcast channel for the data (0 to 255)."""
34
35    data: PybricksBroadcastData
36    """The value or values to be broadcast."""