feat: add utility function to get device_source (#81)

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",

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(

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"
+    )