Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.rst
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ and this project adheres to `Semantic Versioning <https://semver.org/spec/v2.0.0

Added
-----
* Exposed address type (``BLEAddressType.PUBLIC`` or ``BLEAddressType.RANDOM``) in ``BLEDevice``.
* Added ``BleakAdapter`` class with ``get_connected_devices()`` to retrieve BLE devices that are already connected to the system without scanning.

Changed
Expand Down
6 changes: 4 additions & 2 deletions bleak/args/winrt.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,15 +6,17 @@

from typing import Literal, TypedDict

from bleak.backends.device import BLEAddressType


class WinRTClientArgs(TypedDict, total=False):
"""
Windows-specific arguments for :class:`BleakClient`.
"""

address_type: Literal["public", "random"]
address_type: BLEAddressType | Literal["public", "random"]
"""
Can either be ``"public"`` or ``"random"``, depending on the required address
Can either be ``BLEAddressType.PUBLIC`` or ``BLEAddressType.RANDOM``, depending on the required address
type needed to connect to your device.
"""

Expand Down
7 changes: 5 additions & 2 deletions bleak/backends/bluezdbus/adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
from bleak.args.bluez import BlueZAdapterArgs
from bleak.backends.adapter import BaseBleakAdapter
from bleak.backends.bluezdbus.manager import get_global_bluez_manager
from bleak.backends.device import BLEDevice
from bleak.backends.device import BLEAddressType, BLEDevice


class BleakAdapterBlueZDBus(BaseBleakAdapter):
Expand Down Expand Up @@ -41,13 +41,16 @@ async def get_connected_devices(
self._adapter_path, service_uuids
):
address = props["Address"]
address_type = BLEAddressType(props["AddressType"])
# BlueZ generates a name based on the address if no name is available.
# To match other backends, we replace this with None.
name = (
None
if props["Alias"] == props["Address"].replace(":", "-")
else props["Alias"]
)
devices.append(BLEDevice(address, name, {"path": path, "props": props}))
devices.append(
BLEDevice(address, name, {"path": path, "props": props}, address_type)
)

return devices
3 changes: 3 additions & 0 deletions bleak/backends/bluezdbus/scanner.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
import sys
from typing import TYPE_CHECKING

from bleak.backends.device import BLEAddressType

if TYPE_CHECKING:
if sys.platform != "linux":
assert False, "This backend is only available on Linux"
Expand Down Expand Up @@ -214,6 +216,7 @@ def _handle_advertising_data(self, path: str, props: Device1) -> None:
device = self.create_or_update_device(
path,
props["Address"],
BLEAddressType(props["AddressType"]),
device_name_from_props(props),
{"path": path, "props": props},
advertisement_data,
Expand Down
2 changes: 2 additions & 0 deletions bleak/backends/corebluetooth/scanner.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
to_optional_int,
to_optional_str,
)
from bleak.backends.device import BLEAddressType
from bleak.backends.scanner import (
AdvertisementData,
AdvertisementDataCallback,
Expand Down Expand Up @@ -164,6 +165,7 @@ def callback(
device = self.create_or_update_device(
peripheral.identifier().UUIDString(),
address,
BLEAddressType.UNKNOWN, # macOS does not provide address type information
peripheral.name(),
(peripheral, self._manager),
advertisement_data,
Expand Down
21 changes: 18 additions & 3 deletions bleak/backends/device.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,21 +4,36 @@
:py:meth:`bleak.discover`.
"""


from enum import Enum
from typing import Any, Optional
from warnings import warn


class BLEAddressType(Enum):
UNKNOWN = "unknown"
PUBLIC = "public"
RANDOM = "random"


class BLEDevice:
"""
A simple wrapper class representing a BLE server detected during scanning.
"""

__slots__ = ("address", "name", "details")
__slots__ = ("address", "address_type", "name", "details")

def __init__(self, address: str, name: Optional[str], details: Any, **kwargs: Any):
def __init__(
self,
address: str,
name: Optional[str],
details: Any,
address_type: BLEAddressType = BLEAddressType.UNKNOWN,
**kwargs: Any,
):
#: The Bluetooth address of the device on this machine (UUID on macOS).
self.address = address
#: The address type of the device (public or random).
self.address_type = address_type
#: The operating system name of the device (not necessarily the local name
#: from the advertising data), suitable for display to the user.
self.name = name
Expand Down
5 changes: 5 additions & 0 deletions bleak/backends/p4android/defs.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,11 @@
BLEAK_JNI_NAMESPACE + ".PythonBluetoothGattCallback"
)

ADDRESS_TYPE_PUBLIC = 0
ADDRESS_TYPE_RANDOM = 1
ADDRESS_TYPE_ANONYMOUS = 255
ADDRESS_TYPE_UNKNOWN = 65535


class ScanFailed(enum.IntEnum):
ALREADY_STARTED = ScanCallback.SCAN_FAILED_ALREADY_STARTED
Expand Down
11 changes: 11 additions & 0 deletions bleak/backends/p4android/scanner.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

from bleak._compat import override
from bleak._compat import timeout as async_timeout
from bleak.backends.device import BLEAddressType
from bleak.backends.p4android import defs, utils
from bleak.backends.scanner import (
AdvertisementData,
Expand Down Expand Up @@ -247,6 +248,15 @@ def _handle_scan_result(self, result) -> None:
if tx_power == -2147483648: # Integer#MIN_VALUE
tx_power = None

address_type: BLEAddressType
match native_device.getAddressType():
case defs.ADDRESS_TYPE_PUBLIC:
address_type = BLEAddressType.PUBLIC
case defs.ADDRESS_TYPE_RANDOM:
address_type = BLEAddressType.RANDOM
case _:
address_type = BLEAddressType.UNKNOWN

advertisement = AdvertisementData(
local_name=record.getDeviceName(),
manufacturer_data=manufacturer_data,
Expand All @@ -260,6 +270,7 @@ def _handle_scan_result(self, result) -> None:
device = self.create_or_update_device(
native_device.getAddress(),
native_device.getAddress(),
address_type,
native_device.getName(),
native_device,
advertisement,
Expand Down
5 changes: 3 additions & 2 deletions bleak/backends/scanner.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
from typing import Any, NamedTuple, Optional

from bleak.backends import BleakBackend, get_default_backend
from bleak.backends.device import BLEDevice
from bleak.backends.device import BLEAddressType, BLEDevice
from bleak.exc import BleakError

# prevent tasks from being garbage collected
Expand Down Expand Up @@ -238,6 +238,7 @@ def create_or_update_device(
self,
key: str,
address: str,
address_type: BLEAddressType,
name: Optional[str],
details: Any,
adv: AdvertisementData,
Expand All @@ -261,7 +262,7 @@ def create_or_update_device(

device.name = name
except KeyError:
device = BLEDevice(address, name, details)
device = BLEDevice(address, name, details, address_type)

self.seen_devices[key] = (device, adv)

Expand Down
29 changes: 25 additions & 4 deletions bleak/backends/winrt/adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,9 @@

from uuid import UUID

from winrt.system import unbox_string
from winrt.system import unbox_string, unbox_uint8
from winrt.windows.devices.bluetooth import (
BluetoothAddressType,
BluetoothCacheMode,
BluetoothConnectionStatus,
BluetoothDeviceId,
Expand All @@ -19,7 +20,7 @@

from bleak._compat import Self, override
from bleak.backends.adapter import BaseBleakAdapter
from bleak.backends.device import BLEDevice
from bleak.backends.device import BLEAddressType, BLEDevice
from bleak.backends.winrt.util import assert_mta
from bleak.uuids import normalize_uuid_16

Expand All @@ -44,7 +45,11 @@ async def get_connected_devices(
)
connected_devices = (
await DeviceInformation.find_all_async_aqs_filter_and_additional_properties(
selector, ["System.Devices.Aep.DeviceAddress"]
selector,
[
"System.Devices.Aep.DeviceAddress",
"System.Devices.Aep.Bluetooth.Le.AddressType",
],
)
)

Expand Down Expand Up @@ -82,6 +87,22 @@ async def get_connected_devices(
device_info.properties["System.Devices.Aep.DeviceAddress"]
).upper()

devices.append(BLEDevice(address, device_info.name, device_info))
address_type: BLEAddressType
if address_type_prop := device_info.properties.get(
"System.Devices.Aep.Bluetooth.Le.AddressType"
):
match unbox_uint8(address_type_prop):
case BluetoothAddressType.PUBLIC.value:
address_type = BLEAddressType.PUBLIC
case BluetoothAddressType.RANDOM.value:
address_type = BLEAddressType.RANDOM
case _:
address_type = BLEAddressType.UNKNOWN
else:
address_type = BLEAddressType.UNKNOWN

devices.append(
BLEDevice(address, device_info.name, device_info, address_type)
)

return devices
14 changes: 10 additions & 4 deletions bleak/backends/winrt/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,7 @@
from bleak.backends.characteristic import BleakGATTCharacteristic
from bleak.backends.client import BaseBleakClient, NotifyCallback
from bleak.backends.descriptor import BleakGATTDescriptor
from bleak.backends.device import BLEDevice
from bleak.backends.device import BLEAddressType, BLEDevice
from bleak.backends.service import BleakGATTService, BleakGATTServiceCollection
from bleak.backends.winrt.scanner import BleakScannerWinRT, RawAdvData
from bleak.exc import BleakDeviceNotFoundError, BleakError, BleakGATTProtocolError
Expand Down Expand Up @@ -173,7 +173,13 @@ def __init__(

# os-specific options
self._use_cached_services = winrt.get("use_cached_services")
self._address_type = winrt.get("address_type")

if address_type := winrt.get("address_type"):
self._address_type = BLEAddressType(address_type)
elif isinstance(address_or_ble_device, BLEDevice):
self._address_type = address_or_ble_device.address_type
else:
self._address_type = BLEAddressType.UNKNOWN
self._retry_on_services_changed = False

self._services_changed_token: Optional[EventRegistrationToken] = None
Expand All @@ -186,12 +192,12 @@ def __str__(self) -> str:
# Connectivity methods

async def _create_requester(self, bluetooth_address: int) -> BluetoothLEDevice:
if self._address_type is not None:
if self._address_type != BLEAddressType.UNKNOWN:
requester = await BluetoothLEDevice.from_bluetooth_address_with_bluetooth_address_type_async(
bluetooth_address,
(
BluetoothAddressType.PUBLIC
if self._address_type == "public"
if self._address_type == BLEAddressType.PUBLIC
else BluetoothAddressType.RANDOM
),
)
Expand Down
14 changes: 12 additions & 2 deletions bleak/backends/winrt/scanner.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
from typing import Literal, NamedTuple, Optional
from uuid import UUID

from winrt.windows.devices.bluetooth import BluetoothAdapter
from winrt.windows.devices.bluetooth import BluetoothAdapter, BluetoothAddressType
from winrt.windows.devices.bluetooth.advertisement import (
BluetoothLEAdvertisementReceivedEventArgs,
BluetoothLEAdvertisementType,
Expand All @@ -24,6 +24,7 @@

from bleak._compat import override
from bleak.assigned_numbers import AdvertisementDataType
from bleak.backends.device import BLEAddressType
from bleak.backends.scanner import (
AdvertisementData,
AdvertisementDataCallback,
Expand Down Expand Up @@ -132,6 +133,15 @@ def _received_handler(

bdaddr = _format_bdaddr(event_args.bluetooth_address)

address_type: BLEAddressType
match event_args.bluetooth_address_type:
case BluetoothAddressType.PUBLIC:
address_type = BLEAddressType.PUBLIC
case BluetoothAddressType.RANDOM:
address_type = BLEAddressType.RANDOM
case _:
address_type = BLEAddressType.UNKNOWN

# Unlike other platforms, Windows does not combine advertising data for
# us (regular advertisement + scan response) so we have to do it manually.

Expand Down Expand Up @@ -214,7 +224,7 @@ def _received_handler(
)

device = self.create_or_update_device(
bdaddr, bdaddr, local_name, raw_data, advertisement_data
bdaddr, bdaddr, address_type, local_name, raw_data, advertisement_data
)

self.call_detection_callbacks(device, advertisement_data)
Expand Down
Loading