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

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

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

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