diff --git a/src/bleak_retry_connector/__init__.py b/src/bleak_retry_connector/__init__.py index 94d515d..6efdbb2 100644 --- a/src/bleak_retry_connector/__init__.py +++ b/src/bleak_retry_connector/__init__.py @@ -21,6 +21,7 @@ BleakSlotManager, _get_properties, clear_cache, + device_source, get_bluez_device, get_connected_devices, get_device, @@ -60,6 +61,7 @@ "clear_cache", "get_device", "get_device_by_adapter", + "device_source", "restore_discoveries", "retry_bluetooth_connection_error", "BleakClientWithServiceCache", diff --git a/src/bleak_retry_connector/bluez.py b/src/bleak_retry_connector/bluez.py index d4f2f4e..32ad54c 100644 --- a/src/bleak_retry_connector/bluez.py +++ b/src/bleak_retry_connector/bluez.py @@ -67,6 +67,20 @@ async def get_global_bluez_manager_with_timeout() -> "BlueZManager" | None: return None +def device_source(device: BLEDevice) -> str | None: + """Return the device source.""" + return _device_details_value_or_none(device, "source") + + +def _device_details_value_or_none(device: BLEDevice, key: str) -> Any | None: + """Return a value from device details or None.""" + details = device.details + if not isinstance(details, dict) or key not in details: + return None + key_value: str = device.details[key] + return key_value + + def _reset_dbus_socket_cache() -> None: """Reset the dbus socket cache.""" setattr(get_global_bluez_manager_with_timeout, "_has_dbus_socket", None) @@ -79,12 +93,7 @@ def adapter_from_path(path: str) -> str: def path_from_ble_device(device: BLEDevice) -> str | None: """Get the adapter from a ble device.""" - if not isinstance(device.details, dict): - return None - if "path" not in device.details: - return None - path: str = device.details["path"] - return path + return _device_details_value_or_none(device, "path") def _on_characteristic_value_changed( diff --git a/tests/test_bluez.py b/tests/test_bluez.py index 3972d05..a42d2a5 100644 --- a/tests/test_bluez.py +++ b/tests/test_bluez.py @@ -4,10 +4,11 @@ import pytest from bleak.backends.bluezdbus import defs from bleak.backends.bluezdbus.manager import DeviceWatcher +from bleak.backends.device import BLEDevice import bleak_retry_connector -from bleak_retry_connector import BleakSlotManager -from bleak_retry_connector.bluez import ble_device_from_properties +from bleak_retry_connector import BleakSlotManager, device_source +from bleak_retry_connector.bluez import ble_device_from_properties, path_from_ble_device pytestmark = pytest.mark.asyncio @@ -226,3 +227,40 @@ async def test_slot_manager_mac_os(): slot_manager.release_slot(ble_device_hci0) assert slot_manager._get_allocations("hci0") == [] slot_manager.remove_adapter("hci0") + + +def test_device_source(): + ble_device_hci0_2 = BLEDevice( + "FA:23:9D:AA:45:46", + "FA:23:9D:AA:45:46", + { + "source": "aa:bb:cc:dd:ee:ff", + "path": "/org/bluez/hci0/dev_FA_23_9D_AA_45_47", + "props": {}, + }, + -127, + uuids=[], + manufacturer_data={}, + ) + + assert device_source(ble_device_hci0_2) == "aa:bb:cc:dd:ee:ff" + + +def test_path_from_ble_device(): + ble_device_hci0_2 = BLEDevice( + "FA:23:9D:AA:45:46", + "FA:23:9D:AA:45:46", + { + "source": "aa:bb:cc:dd:ee:ff", + "path": "/org/bluez/hci0/dev_FA_23_9D_AA_45_47", + "props": {}, + }, + -127, + uuids=[], + manufacturer_data={}, + ) + + assert ( + path_from_ble_device(ble_device_hci0_2) + == "/org/bluez/hci0/dev_FA_23_9D_AA_45_47" + )