Skip to content

Commit

Permalink
backends/scanner: Add rssi to AdvertisementData
Browse files Browse the repository at this point in the history
This adds an `rssi` attribute to the AdvertisementData type.

This is done so we can eventually deprecate the `rssi` on the BLEDevice
object (#1025).

While we are touching this, also change AdvertisementData to a named
tuple so we don't have to initialize all of the values ourselves.
  • Loading branch information
dlech committed Sep 30, 2022
1 parent c8c1793 commit 422d9b0
Show file tree
Hide file tree
Showing 9 changed files with 54 additions and 32 deletions.
5 changes: 5 additions & 0 deletions CHANGELOG.rst
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,11 @@ Added
* Added support for Python 3.11.
* Added better error message for Bluetooth not authorized on macOS.
* ``BleakDeviceNotFoundError`` which should be raised if a device can not be found by ``connect``, ``pair`` and ``unpair``
* Added ``rssi`` attribute to ``AdvertisementData``.

Changed
-------
Changed ``AdvertisementData`` to a named tuple.

Fixed
-----
Expand Down
3 changes: 2 additions & 1 deletion bleak/backends/bluezdbus/scanner.py
Original file line number Diff line number Diff line change
Expand Up @@ -218,8 +218,9 @@ def _handle_advertising_data(self, path: str, props: Device1) -> None:
manufacturer_data=_manufacturer_data,
service_data=_service_data,
service_uuids=_service_uuids,
platform_data=(path, props),
tx_power=tx_power,
rssi=props.get("RSSI", -127),
platform_data=(path, props),
)

metadata = dict(
Expand Down
5 changes: 3 additions & 2 deletions bleak/backends/corebluetooth/scanner.py
Original file line number Diff line number Diff line change
Expand Up @@ -104,12 +104,13 @@ def callback(p: CBPeripheral, a: Dict[str, Any], r: int) -> None:
tx_power = a.get("kCBAdvDataTxPowerLevel")

advertisement_data = AdvertisementData(
local_name=p.name(),
local_name=a.get("kCBAdvDataLocalName"),
manufacturer_data=manufacturer_data,
service_data=service_data,
service_uuids=service_uuids,
platform_data=(p, a, r),
tx_power=tx_power,
rssi=r,
platform_data=(p, a, r),
)

metadata = dict(
Expand Down
1 change: 0 additions & 1 deletion bleak/backends/device.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@
Created on 2018-04-23 by hbldh <henrik.blidh@nedomkull.com>
"""
from ._manufacturers import MANUFACTURERS


class BLEDevice:
Expand Down
3 changes: 2 additions & 1 deletion bleak/backends/p4android/scanner.py
Original file line number Diff line number Diff line change
Expand Up @@ -256,8 +256,9 @@ def _handle_scan_result(self, result):
manufacturer_data=manufacturer_data,
service_data=service_data,
service_uuids=service_uuids,
platform_data=(result,),
tx_power=tx_power,
rssi=native_device.getRssi(),
platform_data=(result,),
)

metadata = dict(
Expand Down
62 changes: 38 additions & 24 deletions bleak/backends/scanner.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,44 +3,57 @@
import inspect
import os
import platform
from typing import Awaitable, Callable, Dict, List, Optional, Tuple, Type
from typing import Awaitable, Callable, Dict, List, NamedTuple, Optional, Tuple, Type

from ..exc import BleakError
from .device import BLEDevice


class AdvertisementData:
class AdvertisementData(NamedTuple):
"""
Wrapper around the advertisement data that each platform returns upon discovery
"""

def __init__(self, **kwargs):
"""
Keyword Args:
local_name (str): The name of the ble device advertising
manufacturer_data (dict): Manufacturer data from the device
service_data (dict): Service data from the device
service_uuids (list): UUIDs associated with the device
platform_data (tuple): Tuple of platform specific advertisement data
tx_power (int): Transmit power level of the device
"""
# The local name of the device
self.local_name: Optional[str] = kwargs.get("local_name", None)
local_name: Optional[str]
"""
The local name of the device or ``None`` if not included in advertising data.
"""

manufacturer_data: Dict[int, bytes]
"""
Dictionary of manufacturer data in bytes from the received advertisement data or empty dict if not present.
The keys are Bluetooth SIG assigned Company Identifiers and the values are bytes.
https://www.bluetooth.com/specifications/assigned-numbers/company-identifiers/
"""

service_data: Dict[str, bytes]
"""
Dictionary of service data from the received advertisement data or empty dict if not present.
"""

# Dictionary of manufacturer data in bytes
self.manufacturer_data: Dict[int, bytes] = kwargs.get("manufacturer_data", {})
service_uuids: List[str]
"""
List of service UUIDs from the received advertisement data or empty list if not present.
"""

# Dictionary of service data
self.service_data: Dict[str, bytes] = kwargs.get("service_data", {})
tx_power: Optional[int]
"""
Tx Power data from the received advertising data or ``None`` if not present.
"""

# List of UUIDs
self.service_uuids: List[str] = kwargs.get("service_uuids", [])
rssi: int
"""
The Radio Receive Signal Strength (RSSI) in dBm.
"""

# Tuple of platform specific data
self.platform_data: Tuple = kwargs.get("platform_data", ())
platform_data: Tuple
"""
Tuple of platform specific data.
# Tx Power data
self.tx_power: Optional[int] = kwargs.get("tx_power")
This is not a stable API. The actual values may change between releases.
"""

def __repr__(self) -> str:
kwargs = []
Expand All @@ -54,6 +67,7 @@ def __repr__(self) -> str:
kwargs.append(f"service_uuids={repr(self.service_uuids)}")
if self.tx_power is not None:
kwargs.append(f"tx_power={repr(self.tx_power)}")
kwargs.append(f"rssi={repr(self.rssi)}")
return f"AdvertisementData({', '.join(kwargs)})"


Expand Down
3 changes: 2 additions & 1 deletion bleak/backends/winrt/scanner.py
Original file line number Diff line number Diff line change
Expand Up @@ -176,8 +176,9 @@ def _received_handler(
manufacturer_data=mfg_data,
service_data=service_data,
service_uuids=uuids,
platform_data=(sender, raw_data),
tx_power=tx_power,
rssi=event_args.raw_signal_strength_in_d_bm,
platform_data=(sender, raw_data),
)

metadata = dict(
Expand Down
2 changes: 1 addition & 1 deletion examples/detection_callback.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@


def simple_callback(device: BLEDevice, advertisement_data: AdvertisementData):
logger.info(f"{device.address} RSSI: {device.rssi}, {advertisement_data}")
logger.info(f"{device.address}: {advertisement_data}")


async def main(service_uuids):
Expand Down
2 changes: 1 addition & 1 deletion examples/kivy/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ async def example(self):
scanned_devices.sort(key=lambda device: -device.rssi)

for device in scanned_devices:
self.line(f"{device.name} {device.rssi}dB")
self.line(f"{device.name} ({device.address})")

for device in scanned_devices:
self.line(f"Connecting to {device.name} ...")
Expand Down

0 comments on commit 422d9b0

Please sign in to comment.