Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: add new multi platform adapter manager #23

Merged
merged 6 commits into from
Nov 16, 2022
Merged
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
61 changes: 31 additions & 30 deletions src/bluetooth_adapters/__init__.py
Original file line number Diff line number Diff line change
@@ -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)
29 changes: 29 additions & 0 deletions src/bluetooth_adapters/adapters.py
Original file line number Diff line number Diff line change
@@ -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."""
10 changes: 10 additions & 0 deletions src/bluetooth_adapters/const.py
Original file line number Diff line number Diff line change
@@ -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"
42 changes: 42 additions & 0 deletions src/bluetooth_adapters/dbus.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
from __future__ import annotations

import asyncio
import logging
from functools import cache
Expand All @@ -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]]:
Expand Down
2 changes: 2 additions & 0 deletions src/bluetooth_adapters/history.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
from __future__ import annotations

from dataclasses import dataclass
from typing import Any

Expand Down
20 changes: 20 additions & 0 deletions src/bluetooth_adapters/models.py
Original file line number Diff line number Diff line change
@@ -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"
20 changes: 20 additions & 0 deletions src/bluetooth_adapters/systems/__init__.py
Original file line number Diff line number Diff line change
@@ -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()
51 changes: 51 additions & 0 deletions src/bluetooth_adapters/systems/linux.py
Original file line number Diff line number Diff line change
@@ -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
25 changes: 25 additions & 0 deletions src/bluetooth_adapters/systems/macos.py
Original file line number Diff line number Diff line change
@@ -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
27 changes: 27 additions & 0 deletions src/bluetooth_adapters/systems/windows.py
Original file line number Diff line number Diff line change
@@ -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
14 changes: 14 additions & 0 deletions src/bluetooth_adapters/util.py
Original file line number Diff line number Diff line change
@@ -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
Loading