pb_ble.bluezdbus

 24class BlueZBroadcaster(AbstractAsyncContextManager):
 25    """
 26    A BLE broadcaster backed by BlueZ.
 27    Supports multiple advertising sets in parallel.
 28
 29    The recommended use is as a context manager, which ensures that all
 30    registered broadcasts are stopped when exiting the context:
 31
 32    ```python
 33    bus = ...
 34    adapter = ...
 35    device_name = "my-computer"
 36
 37    async with BlueZBroadcaster(bus, adapter, device_name) as broadcaster:
 38        # Start broadcasting
 39        adv = BroadcastAdvertisement(device_name)
 40        await broadcaster.broadcast(adv)
 41        # Stop after 10 seconds
 42        await asyncio.sleep(10)
 43    ```
 44    """
 45
 46    def __init__(self, bus: MessageBus, adapter: ProxyObject, name: str):
 47        """
 48        Creates a new broadcaster.
 49
 50        :param bus: The message bus.
 51        :param adapter: The Bluetooth adapter.
 52        :param name: The name of this broadcaster.
 53        """
 54        self.bus: MessageBus = bus
 55        """The message bus used to connect to DBus."""
 56        self.adapter: ProxyObject = adapter
 57        """A DBus proxy object for the Bluetooth adapter to use for advertising."""
 58        self.name: str = name
 59        """The name of this broadcaster. Will be used as `local_name` in advertisements."""
 60        self.adv_manager: LEAdvertisingManager = LEAdvertisingManager(adapter)
 61        """The BlueZ advertising manager client."""
 62        self.path_namespace: str = f"/org/bluez/{self.name}"
 63        """Path prefix to use for DBus objects created by this broadcaster."""
 64        self.advertisements: dict[str, BroadcastAdvertisement] = {}
 65        """Active advertisements of this broadcaster."""
 66
 67    @overload
 68    async def stop_broadcast(self, adv: str): ...
 69    @overload
 70    async def stop_broadcast(self, adv: BroadcastAdvertisement): ...
 71    async def stop_broadcast(self, adv: str | BroadcastAdvertisement):
 72        """
 73        Stop broadcasting the given advertisement.
 74
 75        :param adv: The broadcast to stop. Takes either the D-Bus path of the
 76            advertisement, or a reference to the object.
 77        """
 78        path = adv.path if isinstance(adv, BroadcastAdvertisement) else adv
 79
 80        try:
 81            await self.adv_manager.unregister_advertisement(path)
 82        except DBusError:
 83            # Advertisement does not exist
 84            pass
 85        finally:
 86            self.bus.unexport(path)
 87            if path in self.advertisements:
 88                del self.advertisements[path]
 89
 90    async def stop(self):
 91        """
 92        Stops this broadcaster. Cleans up any active broadcasts.
 93        """
 94        await asyncio.gather(
 95            *[self.stop_broadcast(path) for path in self.advertisements.keys()]
 96        )
 97
 98    async def __aexit__(self, exc_type, exc, tb):
 99        await self.stop()
100
101    async def broadcast(self, adv: BroadcastAdvertisement):
102        """
103        Start broadcasting the given advertisement.
104
105        :param adv: The reference to the advertisement object.
106        :raises ValueError: If a D-Bus object already exists on the given path.
107        :raises DBusError: If the given advertisement is invalid, or is already
108            registered with BlueZ.
109        """
110        # TODO construct advertisement in here to ensure local_name
111        assert adv._local_name == self.name, f"{adv.name} != {self.name}"
112
113        # cleanup on release of advertisement
114        on_release = adv.on_release
115
116        def release_advertisement(path):
117            try:
118                self.bus.unexport(path)
119                if path in self.advertisements:
120                    del self.advertisements[path]
121            finally:
122                on_release(path)
123
124        adv.on_release = release_advertisement
125
126        log.info("Broadcasting: %s", adv)
127
128        try:
129            self.bus.export(adv.path, adv)
130        except ValueError:
131            # Already exported
132            raise
133
134        try:
135            await self.adv_manager.register_advertisement(adv)
136        except DBusError:
137            # org.bluez.Error.InvalidArguments
138            # org.bluez.Error.AlreadyExists
139            # org.bluez.Error.InvalidLength
140            # org.bluez.Error.NotPermitted
141            raise
142
143        self.advertisements[adv.path] = adv
144
145    def is_broadcasting(self, adv: BroadcastAdvertisement | None = None) -> bool:
146        """
147        Checks whether this broadcaster is active.
148
149        :param adv: The reference to the advertisement object to check,
150            defaults to None (check if any broadcast is active).
151        :return: True if the given (or any) broadcast is active.
152        """
153        if adv is not None:
154            return adv.path in self.advertisements
155        else:
156            return len(self.advertisements) > 0

 37class BlueZPybricksObserver(AbstractAsyncContextManager):
 38    """
 39    A BLE observer backed by BlueZ.
 40    Keeps a cache of observed Pybricks messages.
 41
 42    The recommended use is as a context manager, which ensures that the
 43    underlying BLE scanner is stopped when exiting the context:
 44
 45    ```python
 46    async with BlueZPybricksObserver(channels=[1, 2, 3]) as observer:
 47        # Observe for 10 seconds
 48        await asyncio.sleep(10)
 49        # Check results for channel 2
 50        message = observer.observe(2)
 51        print(message)
 52    ```
 53    """
 54
 55    def __init__(
 56        self,
 57        adapter_name: str | None = None,
 58        scanning_mode: ScanningMode = "passive",
 59        channels: Sequence[int] | None = None,
 60        rssi_threshold: int | None = None,
 61        device_pattern: str | None = None,
 62        message_ttl: int = 60,
 63    ):
 64        """
 65        Creates a new observer.
 66
 67        :param adapter_name: The Bluetooth adapter to use for scanning,
 68            defaults to `None` (auto-discover default device).
 69        :param scanning_mode: The scanning mode to use, defaults to "passive".
 70        :param channels: Channels to observe, defaults to None (all channels).
 71        :param rssi_threshold: Minimum required signal strength of observed
 72            broadcasts in dBm, defaults to None (no RSSI filtering).
 73        :param device_pattern: Pattern that the device name of the sender must
 74            start with, defaults to `None` (no name filtering).
 75        :param message_ttl: Time in seconds to cache observed broadcasts for,
 76            defaults to 60.
 77        """
 78        self.channels = channels or []
 79        """List of channels that this observer is monitoring."""
 80
 81        for channel in self.channels:
 82            if (
 83                not isinstance(channel, int)
 84                or PYBRICKS_MIN_CHANNEL < channel > PYBRICKS_MAX_CHANNEL
 85            ):
 86                raise ValueError(
 87                    f"Observe channel must be list of integers from {PYBRICKS_MIN_CHANNEL} to {PYBRICKS_MAX_CHANNEL}."
 88                )
 89
 90        self.rssi_threshold = rssi_threshold
 91        """The configured RSSI threshold for broadcasts."""
 92        self.device_pattern = device_pattern
 93        """The configured device name pattern match for broadcasts."""
 94        self.advertisements: TTLCache = TTLCache(
 95            maxsize=len(self.channels) or PYBRICKS_MAX_CHANNEL, ttl=message_ttl
 96        )
 97        """Cache of observed broadcasts."""
 98
 99        # Filters used for active scanning
100        filters: BlueZDiscoveryFilters = BlueZDiscoveryFilters()
101
102        if device_pattern:
103            filters["Pattern"] = device_pattern
104        if rssi_threshold:
105            filters["RSSI"] = rssi_threshold
106
107        # Patterns used for passive scanning
108        or_patterns: list[OrPattern | tuple[int, AdvertisementDataType, bytes]]
109
110        if self.channels:
111            or_patterns = [
112                OrPattern(
113                    0,
114                    AdvertisementDataType.MANUFACTURER_SPECIFIC_DATA,
115                    pack("<HB", LEGO_CID, channel),
116                )
117                for channel in self.channels
118            ]
119        else:
120            or_patterns = [
121                OrPattern(
122                    0,
123                    AdvertisementDataType.MANUFACTURER_SPECIFIC_DATA,
124                    pack("<H", LEGO_CID),
125                )
126            ]
127
128        log.debug(
129            "Observer init: scanning_mode=%s, device_pattern=%s, rssi_threshold=%s",
130            scanning_mode,
131            device_pattern,
132            rssi_threshold,
133        )
134
135        self._scanner = BleakScanner(
136            detection_callback=self._callback,
137            scanning_mode=scanning_mode,
138            bluez=BlueZScannerArgs(
139                filters=filters,
140                or_patterns=or_patterns,
141            ),
142            adapter=adapter_name,
143        )
144
145    def _callback(self, device: BLEDevice, ad: AdvertisementData):
146        """
147        @public
148        The callback function for detected BLE advertisements from the
149        scanner.
150
151        Performs filtering of advertisements, decodes the Pybricks
152        broadcast message contained in the advertisement, and stores
153        it in the broadcast cache.
154
155        Depending on the selected scanning mode, certain filters must
156        be applied here:
157
158        1. Filter advertisements based on the configured `rssi_threshold`
159            and `device_pattern` (required in "passive" mode).
160        2. Filter advertisements that contain invalid manufacturer data,
161            such as non-Pybricks advertisements (required in "active" mode).
162        4. Filter messages on the incorrect Pybricks channel (required in
163            "active" mode).
164
165        :param device: The device sending the advertisement.
166        :param ad: The advertisement data.
167        """
168        if self.rssi_threshold is not None and ad.rssi < self.rssi_threshold:
169            log.debug("Filtered AD due to RSSI below threshold: %i", ad.rssi)
170            return
171
172        if (ad.local_name and self.device_pattern) and not ad.local_name.startswith(
173            self.device_pattern
174        ):
175            log.debug("Filtered AD due to invalid device name: %s", ad.local_name)
176            return
177
178        if LEGO_CID not in ad.manufacturer_data:
179            log.debug(
180                "Filtered AD due to invalid manufacturer data: %s",
181                ad.manufacturer_data.keys(),
182            )
183            return
184
185        message = ad.manufacturer_data[LEGO_CID]
186        channel, data = decode_message(message)
187
188        if self.channels and channel not in self.channels:
189            log.debug("Filtered broadcast due to wrong channel: %i", channel)
190            return
191
192        log.info(
193            "Pybricks broadcast on channel %i: %s (rssi %s)", channel, data, ad.rssi
194        )
195        self.advertisements[channel] = ObservedAdvertisement(data, ad.rssi)
196
197    def observe(self, channel: int) -> ObservedAdvertisement | None:
198        """
199        Retrieves the last observed data for a given channel.
200
201        :param channel: The channel to observe (0 to 255).
202        :return: The received data in the same format as it was sent, or `None`
203            if no recent data is available.
204        """
205        if self.channels and channel not in self.channels:
206            raise ValueError(f"Channel {channel} not allocated.")
207
208        return self.advertisements.get(channel, None)
209
210    async def __aenter__(self):
211        log.info("Observing on channels %s...", self.channels or "ALL")
212        await self._scanner.start()
213        return self
214
215    async def __aexit__(self, exc_type, exc, tb):
216        await self._scanner.stop()

28class ObservedAdvertisement(NamedTuple):
29    """
30    Data structure for an observed broadcast.
31    """
32
33    data: PybricksBroadcastData
34    rssi: int

 99class LEAdvertisement(ServiceInterface):
100    """
101    Implementation of the `org.bluez.LEAdvertisement1` D-Bus interface.
102    """
103
104    INTERFACE_NAME: str = "org.bluez.LEAdvertisement1"
105
106    def __init__(
107        self,
108        advertising_type: Type,
109        local_name: str,
110        index: int = 0,
111        includes: set[Include] = set(),
112    ):
113        if index < 0:
114            raise ValueError("index must be positive")
115
116        self.index = index
117        self.path = f"/org/bluez/{local_name}/advertisement{index:03}"
118
119        self._type: str = advertising_type.value
120        self._service_uuids: list[str] = []
121        self._manufacturer_data: dict[int, Variant] = {}  # uint16 -> bytes
122        self._solicit_uuids: list[str] = []
123        self._service_data: dict[str, Variant] = {}  # uint16 | str -> bytes
124        self._data: dict[int, Variant] = {}  # EXPERIMENTAL # uint8 -> bytes
125        self._discoverable: bool = False  # EXPERIMENTAL
126        self._discoverable_timeout: int = 0  # EXPERIMENTAL # uint16
127        self._includes: list[str] = [i.value for i in includes]
128        self._local_name: str = local_name
129        self._appearance: int = 0x00  # uint16
130        self._duration: int = 2  # uint16
131        self._timeout: int = 0  # uint16
132        self._secondary_channel: str = SecondaryChannel.ONE.value  # EXPERIMENTAL
133        self._min_interval: int = 100  # EXPERIMENTAL # uint32
134        self._max_interval: int = 1000  # EXPERIMENTAL # uint32
135        self._tx_power: int = 7  # EXPERIMENTAL # int16
136
137        super().__init__(self.INTERFACE_NAME)
138
139    def _enable_props(self, *prop_names: str):
140        """
141        Enables the given properties.
142
143        This should be used by subclasses to opt-into experimental properties.
144
145        :param prop_names: List of D-Bus property names to enable.
146        :raises ValueError: If an unknown property was passed.
147        """
148        for prop_name in prop_names:
149            prop: _Property | None = next(
150                (
151                    p
152                    for p in ServiceInterface._get_properties(self)
153                    if p.name == prop_name
154                ),
155                None,
156            )
157            if prop is None:
158                raise ValueError(f"Unknown property: {prop_name}")
159            else:
160                prop.disabled = False
161
162    def _disable_props(self, *prop_names: str):
163        """
164        Disables the given properties.
165
166        This can be used by subclasses to opt-out of exposing certain properties.
167
168        :param prop_names: List of D-Bus property names to disable.
169        :raises ValueError: If an unknown property was passed.
170        """
171        for prop_name in prop_names:
172            prop: _Property | None = next(
173                (
174                    p
175                    for p in ServiceInterface._get_properties(self)
176                    if p.name == prop_name
177                ),
178                None,
179            )
180            if prop is None:
181                raise ValueError(f"Unknown property: {prop_name}")
182            else:
183                prop.disabled = True
184
185    @dbus_method()
186    def Release(self) -> None:
187        logger.debug("Released advertisement: %s", self)
188
189    @dbus_property(access=PropertyAccess.READ)
190    def Type(self) -> DBusStr:
191        """
192        Determines the type of advertising packet requested.
193        """
194        return self._type
195
196    @Type.setter  # type: ignore[no-redef]
197    def Type(self, type: DBusStr) -> None:
198        self._type = type
199
200    @dbus_property()
201    def ServiceUUIDs(self) -> DBusArrayString:
202        """
203        List of UUIDs to include in the "Service UUID" field of the Advertising Data.
204        """
205        return self._service_uuids
206
207    @ServiceUUIDs.setter  # type: ignore[no-redef]
208    def ServiceUUIDs(self, service_uuids: DBusArrayString) -> None:
209        self._service_uuids = service_uuids
210
211    @dbus_property()
212    def ManufacturerData(self) -> DBusUInt16Dict:
213        """
214        Manufacturer Data fields to include in the Advertising Data.
215        Keys are the Manufacturer ID to associate with the data.
216        """
217        return self._manufacturer_data
218
219    @ManufacturerData.setter  # type: ignore[no-redef]
220    def ManufacturerData(self, data: DBusUInt16Dict) -> None:
221        self._manufacturer_data = data
222
223    @dbus_property()
224    def SolicitUUIDs(self) -> DBusArrayString:
225        """
226        Array of UUIDs to include in "Service Solicitation" Advertisement Data.
227        """
228        return self._solicit_uuids
229
230    @SolicitUUIDs.setter  # type: ignore[no-redef]
231    def SolicitUUIDs(self, uuids: DBusArrayString) -> None:
232        self._solicit_uuids = uuids
233
234    @dbus_property()
235    def ServiceData(self) -> DBusDict:
236        """
237        Service Data elements to include. The keys are the UUID to associate with the data.
238        """
239        return self._service_data
240
241    @ServiceData.setter  # type: ignore[no-redef]
242    def ServiceData(self, data: DBusDict) -> None:
243        self._service_data = data
244
245    @dbus_property(disabled=True)
246    def Data(self) -> DBusByteDict:
247        """
248        Advertising Data to include.
249        Key is the advertising type and value is the data as byte array.
250        """
251        return self._data
252
253    @Data.setter  # type: ignore[no-redef]
254    def Data(self, data: DBusByteDict) -> None:
255        self._data = data
256
257    @dbus_property(disabled=True)
258    def Discoverable(self) -> DBusBool:
259        """
260        Advertise as general discoverable.
261        When present this will override adapter Discoverable property.
262        """
263        return self._discoverable
264
265    @Discoverable.setter  # type: ignore[no-redef]
266    def Discoverable(self, discoverable: DBusBool) -> None:
267        self._discoverable = discoverable
268
269    @dbus_property(disabled=True)
270    def DiscoverableTimeout(self) -> DBusUInt16:
271        """
272        The discoverable timeout in seconds.
273        A value of zero means that the timeout is disabled and it will stay in discoverable/limited mode forever.
274        """
275        return self._discoverable_timeout
276
277    @DiscoverableTimeout.setter  # type: ignore[no-redef]
278    def DiscoverableTimeout(self, timeout: DBusUInt16) -> None:
279        self._discoverable_timeout = timeout
280
281    @dbus_property()
282    def Includes(self) -> DBusArrayString:
283        """
284        List of features to be included in the advertising packet.
285        """
286        return self._includes
287
288    @Includes.setter  # type: ignore[no-redef]
289    def Includes(self, includes: DBusArrayString) -> None:
290        self._includes = includes
291
292    @dbus_property()
293    def LocalName(self) -> DBusStr:
294        """
295        Local name to be used in the advertising report.
296        If the string is too big to fit into the packet it will be truncated.
297        """
298        return self._local_name
299
300    @LocalName.setter  # type: ignore[no-redef]
301    def LocalName(self, name: DBusStr) -> None:
302        self._local_name = name
303
304    @dbus_property()
305    def Appearance(self) -> DBusUInt16:
306        """
307        Appearance to be used in the advertising report.
308        """
309        return self._appearance
310
311    @Appearance.setter  # type: ignore[no-redef]
312    def Appearance(self, appearance: DBusUInt16) -> None:
313        self._appearance = appearance
314
315    @dbus_property()
316    def Duration(self) -> DBusUInt16:
317        """
318        Rotation duration of the advertisement in seconds.
319        If there are other applications advertising no duration is set the default is 2 seconds.
320        """
321        return self._duration
322
323    @Duration.setter  # type: ignore[no-redef]
324    def Duration(self, seconds: DBusUInt16) -> None:
325        self._duration = seconds
326
327    @dbus_property()
328    def Timeout(self) -> DBusUInt16:
329        """
330        Timeout of the advertisement in seconds.
331        This defines the lifetime of the advertisement.
332        """
333        return self._timeout
334
335    @Timeout.setter  # type: ignore[no-redef]
336    def Timeout(self, seconds: DBusUInt16) -> None:
337        self._timeout = seconds
338
339    @dbus_property(disabled=True)
340    def SecondaryChannel(self) -> DBusStr:
341        """
342        Secondary channel to be used.
343        Primary channel is always set to "1M" except when "Coded" is set.
344        """
345        return self._secondary_channel
346
347    @SecondaryChannel.setter  # type: ignore[no-redef]
348    def SecondaryChannel(self, channel: DBusStr) -> None:
349        self._secondary_channel = channel
350
351    @dbus_property(disabled=True)
352    def MinInterval(self) -> DBusUInt32:
353        """
354        Minimum advertising interval to be used by the advertising set, in milliseconds.
355        Acceptable values are in the range [20ms, 10,485s].
356        If the provided MinInterval is larger than the provided MaxInterval,
357        the registration will return failure.
358        """
359        return self._min_interval
360
361    @MinInterval.setter  # type: ignore[no-redef]
362    def MinInterval(self, milliseconds: DBusUInt32) -> None:
363        self._min_interval = milliseconds
364
365    @dbus_property(disabled=True)
366    def MaxInterval(self) -> DBusUInt32:
367        """
368        Maximum advertising interval to be used by the advertising set, in milliseconds.
369        Acceptable values are in the range [20ms, 10,485s].
370        If the provided MinInterval is larger than the provided MaxInterval,
371        the registration will return failure.
372        """
373        return self._max_interval
374
375    @MaxInterval.setter  # type: ignore[no-redef]
376    def MaxInterval(self, milliseconds: DBusUInt32) -> None:
377        self._max_interval = milliseconds
378
379    @dbus_property(disabled=True)
380    def TxPower(self) -> DBusInt16:
381        """
382        Requested transmission power of this advertising set.
383        The provided value is used only if the "CanSetTxPower" feature is enabled on the org.bluez.LEAdvertisingManager(5).
384        The provided value must be in range [-127 to +20], where units are in dBm.
385        """
386        return self._tx_power
387
388    @TxPower.setter  # type: ignore[no-redef]
389    def TxPower(self, dbm: DBusInt16) -> None:
390        self._tx_power = dbm