feat: add new multi platform adapter manager (#23)

src/bluetooth_adapters/__init__.py

@@ -1,46 +1,47 @@
 __version__ = "0.7.0"
 
-from typing import Any
 
+from .adapters import BluetoothAdapters
+from .const import (
+    DEFAULT_ADDRESS,
+    MACOS_DEFAULT_BLUETOOTH_ADAPTER,
+    UNIX_DEFAULT_BLUETOOTH_ADAPTER,
+    WINDOWS_DEFAULT_BLUETOOTH_ADAPTER,
+)
 from .dbus import (
-    _adapters_from_managed_objects,
+    BlueZDBusObjects,
     get_bluetooth_adapter_details,
     get_bluetooth_adapters,
     get_dbus_managed_objects,
 )
-from .history import AdvertisementHistory, load_history_from_managed_objects
+from .history import AdvertisementHistory
+from .models import (
+    ADAPTER_ADDRESS,
+    ADAPTER_HW_VERSION,
+    ADAPTER_PASSIVE_SCAN,
+    ADAPTER_SW_VERSION,
+    AdapterDetails,
+)
+from .systems import get_adapters
+from .util import adapter_human_name, adapter_unique_name
 
 __all__ = [
     "AdvertisementHistory",
+    "BluetoothAdapters",
     "BlueZDBusObjects",
+    "adapter_human_name",
+    "adapter_unique_name",
     "get_bluetooth_adapters",
     "get_bluetooth_adapter_details",
     "get_dbus_managed_objects",
+    "get_adapters",
+    "AdapterDetails",
+    "ADAPTER_ADDRESS",
+    "ADAPTER_SW_VERSION",
+    "ADAPTER_HW_VERSION",
+    "ADAPTER_PASSIVE_SCAN",
+    "WINDOWS_DEFAULT_BLUETOOTH_ADAPTER",
+    "MACOS_DEFAULT_BLUETOOTH_ADAPTER",
+    "UNIX_DEFAULT_BLUETOOTH_ADAPTER",
+    "DEFAULT_ADDRESS",
 ]
-
-
-class BlueZDBusObjects:
-    """Fetch and parse BlueZObjects."""
-
-    def __init__(self) -> None:
-        """Init the manager."""
-        self._managed_objects: dict[str, Any] = {}
-
-    async def load(self) -> None:
-        """Load from the bus."""
-        self._managed_objects = await get_dbus_managed_objects()
-
-    @property
-    def adapters(self) -> list[str]:
-        """Get adapters."""
-        return list(self.adapter_details)
-
-    @property
-    def adapter_details(self) -> dict[str, dict[str, Any]]:
-        """Get adapters."""
-        return _adapters_from_managed_objects(self._managed_objects)
-
-    @property
-    def history(self) -> dict[str, AdvertisementHistory]:
-        """Get history from managed objects."""
-        return load_history_from_managed_objects(self._managed_objects)

src/bluetooth_adapters/adapters.py

@@ -0,0 +1,29 @@
+"""Base class for Bluetooth adapters."""
+from __future__ import annotations
+
+from abc import abstractproperty
+
+from .history import AdvertisementHistory
+from .models import AdapterDetails
+
+
+class BluetoothAdapters:
+    """Class for getting the bluetooth adapters on a system."""
+
+    async def refresh(self) -> None:
+        """Refresh the adapters."""
+
+    @property
+    def history(self) -> dict[str, AdvertisementHistory]:
+        """Get the history."""
+        return {}
+
+    @abstractproperty
+    @property
+    def adapters(self) -> dict[str, AdapterDetails]:
+        """Get the adapter details."""
+
+    @abstractproperty
+    @property
+    def default_adapter(self) -> str:
+        """Get the default adapter."""

src/bluetooth_adapters/const.py

@@ -0,0 +1,10 @@
+"""Constants for bluetooth adapters."""
+
+from typing import Final
+
+WINDOWS_DEFAULT_BLUETOOTH_ADAPTER: Final = "bluetooth"
+MACOS_DEFAULT_BLUETOOTH_ADAPTER: Final = "Core Bluetooth"
+UNIX_DEFAULT_BLUETOOTH_ADAPTER: Final = "hci0"
+
+# Some operating systems hide the adapter address for privacy reasons (ex MacOS)
+DEFAULT_ADDRESS: Final = "00:00:00:00:00:00"

src/bluetooth_adapters/dbus.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import asyncio
 import logging
 from functools import cache
@@ -8,11 +10,51 @@
 from dbus_fast import BusType, Message, MessageType, unpack_variants
 from dbus_fast.aio import MessageBus
 
+from .history import AdvertisementHistory, load_history_from_managed_objects
+
 _LOGGER = logging.getLogger(__name__)
 
 REPLY_TIMEOUT = 8
 
 
+class BlueZDBusObjects:
+    """Fetch and parse BlueZObjects."""
+
+    def __init__(self) -> None:
+        """Init the manager."""
+        self._packed_managed_objects: dict[str, Any] = {}
+        self._unpacked_managed_objects: dict[str, Any] = {}
+
+    async def load(self) -> None:
+        """Load from the bus."""
+        self._packed_managed_objects = await _get_dbus_managed_objects()
+        self._unpacked_managed_objects = {}
+
+    @property
+    def adapters(self) -> list[str]:
+        """Get adapters."""
+        return list(self.adapter_details)
+
+    @property
+    def unpacked_managed_objects(self) -> dict[str, Any]:
+        """Get unpacked managed objects."""
+        if not self._unpacked_managed_objects:
+            self._unpacked_managed_objects = unpack_variants(
+                self._packed_managed_objects
+            )
+        return self._unpacked_managed_objects
+
+    @property
+    def adapter_details(self) -> dict[str, dict[str, Any]]:
+        """Get adapters."""
+        return _adapters_from_managed_objects(self.unpacked_managed_objects)
+
+    @property
+    def history(self) -> dict[str, AdvertisementHistory]:
+        """Get history from managed objects."""
+        return load_history_from_managed_objects(self.unpacked_managed_objects)
+
+
 def _adapters_from_managed_objects(
     managed_objects: dict[str, Any]
 ) -> dict[str, dict[str, Any]]:

src/bluetooth_adapters/history.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from dataclasses import dataclass
 from typing import Any
 

src/bluetooth_adapters/models.py

@@ -0,0 +1,20 @@
+"""Models for bluetooth adapters."""
+
+from __future__ import annotations
+
+from typing import Final, TypedDict
+
+
+class AdapterDetails(TypedDict, total=False):
+    """Adapter details."""
+
+    address: str
+    sw_version: str
+    hw_version: str | None
+    passive_scan: bool
+
+
+ADAPTER_ADDRESS: Final = "address"
+ADAPTER_SW_VERSION: Final = "sw_version"
+ADAPTER_HW_VERSION: Final = "hw_version"
+ADAPTER_PASSIVE_SCAN: Final = "passive_scan"

src/bluetooth_adapters/systems/__init__.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+import platform
+
+from ..adapters import BluetoothAdapters
+
+
+def get_adapters() -> BluetoothAdapters:
+    """Get the adapters."""
+    if platform.system() == "Windows":
+        from .windows import WindowsAdapters
+
+        return WindowsAdapters()
+    if platform.system() == "Darwin":
+        from .macos import MacOSAdapters
+
+        return MacOSAdapters()
+    from .linux import LinuxAdapters
+
+    return LinuxAdapters()

src/bluetooth_adapters/systems/linux.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+from ..adapters import BluetoothAdapters
+from ..const import UNIX_DEFAULT_BLUETOOTH_ADAPTER
+from ..dbus import BlueZDBusObjects
+from ..history import AdvertisementHistory
+from ..models import AdapterDetails
+
+
+class LinuxAdapters(BluetoothAdapters):
+    """Class for getting the bluetooth adapters on a Linux system."""
+
+    def __init__(self) -> None:
+        """Initialize the adapter."""
+        self._bluez = BlueZDBusObjects()
+        self._adapters: dict[str, AdapterDetails] | None = None
+
+    async def refresh(self) -> None:
+        """Refresh the adapters."""
+        await self._bluez.load()
+        self._adapters = None
+
+    @property
+    def history(self) -> dict[str, AdvertisementHistory]:
+        """Get the bluez history."""
+        return self._bluez.history
+
+    @property
+    def adapters(self) -> dict[str, AdapterDetails]:
+        """Get the adapter details."""
+        if self._adapters is None:
+            adapters: dict[str, AdapterDetails] = {}
+            adapter_details = self._bluez.adapter_details
+            for adapter, details in adapter_details.items():
+                if adapter1 := details.get("org.bluez.Adapter1"):
+                    adapters[adapter] = AdapterDetails(
+                        address=adapter1["Address"],
+                        sw_version=adapter1[
+                            "Name"
+                        ],  # This is actually the BlueZ version
+                        hw_version=adapter1.get("Modalias"),
+                        passive_scan="org.bluez.AdvertisementMonitorManager1"
+                        in details,
+                    )
+            self._adapters = adapters
+        return self._adapters
+
+    @property
+    def default_adapter(self) -> str:
+        """Get the default adapter."""
+        return UNIX_DEFAULT_BLUETOOTH_ADAPTER

src/bluetooth_adapters/systems/macos.py

@@ -0,0 +1,25 @@
+import platform
+
+from ..adapters import BluetoothAdapters
+from ..const import DEFAULT_ADDRESS, MACOS_DEFAULT_BLUETOOTH_ADAPTER
+from ..models import AdapterDetails
+
+
+class MacOSAdapters(BluetoothAdapters):
+    """Class for getting the bluetooth adapters on a MacOS system."""
+
+    @property
+    def adapters(self) -> dict[str, AdapterDetails]:
+        """Get the adapter details."""
+        return {
+            MACOS_DEFAULT_BLUETOOTH_ADAPTER: AdapterDetails(
+                address=DEFAULT_ADDRESS,
+                sw_version=platform.release(),
+                passive_scan=False,
+            )
+        }
+
+    @property
+    def default_adapter(self) -> str:
+        """Get the default adapter."""
+        return MACOS_DEFAULT_BLUETOOTH_ADAPTER

src/bluetooth_adapters/systems/windows.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+import platform
+
+from ..adapters import BluetoothAdapters
+from ..const import DEFAULT_ADDRESS, WINDOWS_DEFAULT_BLUETOOTH_ADAPTER
+from ..models import AdapterDetails
+
+
+class WindowsAdapters(BluetoothAdapters):
+    """Class for getting the bluetooth adapters on a Windows system."""
+
+    @property
+    def adapters(self) -> dict[str, AdapterDetails]:
+        """Get the adapter details."""
+        return {
+            WINDOWS_DEFAULT_BLUETOOTH_ADAPTER: AdapterDetails(
+                address=DEFAULT_ADDRESS,
+                sw_version=platform.release(),
+                passive_scan=False,
+            )
+        }
+
+    @property
+    def default_adapter(self) -> str:
+        """Get the default adapter."""
+        return WINDOWS_DEFAULT_BLUETOOTH_ADAPTER

src/bluetooth_adapters/util.py

@@ -0,0 +1,14 @@
+"""Utils."""
+from __future__ import annotations
+
+from .const import DEFAULT_ADDRESS
+
+
+def adapter_human_name(adapter: str, address: str) -> str:
+    """Return a human readable name for the adapter."""
+    return adapter if address == DEFAULT_ADDRESS else f"{adapter} ({address})"
+
+
+def adapter_unique_name(adapter: str, address: str) -> str:
+    """Return a unique name for the adapter."""
+    return adapter if address == DEFAULT_ADDRESS else address