86 adapter selection

bless/backends/bluezdbus/dbus/utils.py

@@ -1,61 +1,84 @@
 import bleak.backends.bluezdbus.defs as defs  # type: ignore
 
-from typing import Dict, Optional, cast
+from typing import Dict, List, Optional
 
 from dbus_next.aio import MessageBus, ProxyObject, ProxyInterface  # type: ignore
 from dbus_next.introspection import Node  # type: ignore
 
 
-async def find_adapter(bus: MessageBus) -> Optional[str]:
+async def list_adapters(bus: MessageBus) -> List[str]:
     """
-    Returns the first object within the bluez service that has a GattManager1
-    interface
+    Returns a list of strings that represent host-controller interfaces for
+    bluetooth
 
     Parameters
     ----------
-    bus : MessageBus
+    bus : MessageBux
         The currently connected message bus to communicate with BlueZ
 
     Returns
     -------
-    Optional[str]
-        The name of the adapter interface on the dbus.
-        None if it does not find an adapter
+    List[str]
+        A list of adapter interfaces on the dbus
     """
     bluez_node: Node = await bus.introspect(defs.BLUEZ_SERVICE, "/")
     bluez_obj: ProxyObject = bus.get_proxy_object(defs.BLUEZ_SERVICE, "/", bluez_node)
-    interface: ProxyInterface = bluez_obj.get_interface(
-        defs.OBJECT_MANAGER_INTERFACE
-    )
+    interface: ProxyInterface = bluez_obj.get_interface(defs.OBJECT_MANAGER_INTERFACE)
     bt_objects: Dict = await interface.call_get_managed_objects()  # type: ignore
 
-    for objs, props in bt_objects.items():
-        if defs.GATT_MANAGER_INTERFACE in props.keys():
-            return objs
+    adapters: List[str] = [
+        objs
+        for objs, props in bt_objects.items()
+        if defs.GATT_MANAGER_INTERFACE in props.keys()
+    ]
+    return adapters
+
+
+async def find_adapter(bus: MessageBus, adapter: str = "hci0") -> str:
+    """
+    Returns the first object within the bluez service that has a GattManager1
+    interface
+
+    Parameters
+    ----------
+    bus : MessageBus
+        The currently connected message bus to communicate with BlueZ
+
+    adapter : str
+        The adapter to find. Default is 'hci0'
 
-    return None
+    Returns
+    -------
+    str
+        The dbus path to the adapter
+    """
+    adapter_strs: List[str] = await list_adapters(bus)
+    found_adapter: List[str] = [a for a in adapter_strs if adapter in a]
+    if len(found_adapter) > 0:
+        return found_adapter[0]
+    raise Exception(f"No adapter named {adapter} found")
 
 
-async def get_adapter(bus: MessageBus) -> Optional[ProxyObject]:
+async def get_adapter(bus: MessageBus, adapter: Optional[str] = None) -> ProxyObject:
     """
-    Gets the bluetooth adapter
+    Gets the bluetooth adapter specified by adapter or the default if adapter
+    is None
 
     Parameters
     ----------
     bus : MessageBus
         The connected DBus object
+    adapter: Optional[str]
+        A string that points to the HCI adapter
 
     Returns
     -------
-    Optional[ProxyObject]
+    ProxyObject
         The adapter object
-        None if not found
     """
-    oadapter_path: Optional[str] = await find_adapter(bus)
-    if oadapter_path is None:
-        return None
-
-    adapter_path: str = cast(str, oadapter_path)
+    adapter_path: str = await find_adapter(
+        bus, adapter if adapter is not None else "hci0"
+    )
     adapter_node: Node = await bus.introspect(defs.BLUEZ_SERVICE, adapter_path)
     adapter_obj: ProxyObject = bus.get_proxy_object(
         defs.BLUEZ_SERVICE, adapter_path, adapter_node

bless/backends/bluezdbus/server.py

@@ -38,13 +38,14 @@ class BlessServerBlueZDBus(BaseBlessServer):
 
     """
 
-    def __init__(self, name: str, loop: AbstractEventLoop = None, **kwargs):
+    def __init__(self, name: str, loop: Optional[AbstractEventLoop] = None, **kwargs):
         super(BlessServerBlueZDBus, self).__init__(loop=loop, **kwargs)
         self.name: str = name
+        self._adapter: Optional[str] = kwargs.get("adapter", None)
 
         self.setup_task: asyncio.Task = self.loop.create_task(self.setup())
 
-    async def setup(self):
+    async def setup(self: "BlessServerBlueZDBus"):
         """
         Asyncronous side of init
         """
@@ -61,7 +62,9 @@ async def setup(self):
         self.app.StartNotify = lambda x: None
         self.app.StopNotify = lambda x: None
 
-        potential_adapter: Optional[ProxyObject] = await get_adapter(self.bus)
+        potential_adapter: Optional[ProxyObject] = await get_adapter(
+            self.bus, self._adapter
+        )
         if potential_adapter is None:
             raise Exception("Could not locate bluetooth adapter")
         self.adapter: ProxyObject = cast(ProxyObject, potential_adapter)

bless/backends/corebluetooth/peripheral_manager_delegate.py

@@ -33,7 +33,7 @@ class PeripheralManagerDelegate(  # type: ignore
         NSObject,
         protocols=[CBPeripheralManagerDelegate]
         ):
-    def init(self):
+    def init(self: "PeripheralManagerDelegate"):
         self = objc.super(PeripheralManagerDelegate, self).init()
 
         self.event_loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()
@@ -55,7 +55,7 @@ def init(self):
         # that the bluetooth module is powered on
         self._powered_on_event.wait()
 
-        self._central_subscriptions = {}
+        self._central_subscriptions: Dict = {}
 
         if not self.compliant():
             LOGGER.warning("PeripheralManagerDelegate is not compliant")

bless/backends/corebluetooth/server.py

@@ -17,7 +17,6 @@
 from bleak.backends.service import BleakGATTService  # type: ignore
 
 from .peripheral_manager_delegate import PeripheralManagerDelegate  # type: ignore
-from bless.exceptions import BlessError
 from bless.backends.server import BaseBlessServer  # type: ignore
 from bless.backends.corebluetooth.service import BlessGATTServiceCoreBluetooth
 from bless.backends.corebluetooth.characteristic import (  # type: ignore
@@ -49,7 +48,7 @@ class BlessServerCoreBluetooth(BaseBlessServer):
         The delegated class to manage this peripheral device
     """
 
-    def __init__(self, name: str, loop: AbstractEventLoop = None, **kwargs):
+    def __init__(self, name: str, loop: Optional[AbstractEventLoop] = None, **kwargs):
         super(BlessServerCoreBluetooth, self).__init__(loop=loop, **kwargs)
 
         self.name: str = name
@@ -84,9 +83,6 @@ async def start(
             logger.debug("Adding service: {}".format(bleak_service.uuid))
             await self.peripheral_manager_delegate.add_service(service_obj)
 
-        if not self.read_request_func or not self.write_request_func:
-            raise BlessError("Callback functions must be initialized first")
-
         advertisement_uuids: List
         if (prioritize_local_name) and len(self.name) > 10:
             advertisement_uuids = []

bless/backends/server.py

@@ -28,7 +28,7 @@ class BaseBlessServer(abc.ABC):
         Used to manage services and characteristics that this server advertises
     """
 
-    def __init__(self, loop: AbstractEventLoop = None, **kwargs):
+    def __init__(self, loop: Optional[AbstractEventLoop] = None, **kwargs):
         self.loop: AbstractEventLoop = loop if loop else asyncio.get_event_loop()
 
         self._callbacks: Dict[str, Callable[[Any], Any]] = {}

bless/backends/winrt/ble/adapter.py

@@ -15,7 +15,7 @@
 
 
 class BLEAdapter:
-    def __init__(self):
+    def __init__(self: "BLEAdapter"):
         self._adapter_name: str = get_bluetooth_adapter()
         self._device_guid: str = "{a5dcbf10-6530-11d2-901f-00c04fb951ed}"
         self._device_name: str = self._adapter_name.replace("\\", "#")
@@ -49,7 +49,7 @@ def _set_registry_name(self, local_name: str):
         )
         CloseKey(key)
 
-    def _restart_device(self):
+    def _restart_device(self: "BLEAdapter"):
         os_major_version: int = win32api.GetVersionEx()[0]
         control_code: int = 0x220fd4 if os_major_version < 6 else 0x411008
         reload_command: int = 4

bless/backends/winrt/server.py

@@ -64,7 +64,7 @@ class BlessServerWinRT(BaseBlessServer):
     def __init__(
         self,
         name: str,
-        loop: AbstractEventLoop = None,
+        loop: Optional[AbstractEventLoop] = None,
         name_overwrite: bool = False,
         **kwargs,
     ):
@@ -96,7 +96,7 @@ def __init__(
         self._adapter: BLEAdapter = BLEAdapter()
         self._name_overwrite: bool = name_overwrite
 
-    async def start(self, **kwargs):
+    async def start(self: "BlessServerWinRT", **kwargs):
         """
         Start the server
 
@@ -122,7 +122,7 @@ async def start(self, **kwargs):
         self._advertising = True
         self._advertising_started.wait()
 
-    async def stop(self):
+    async def stop(self: "BlessServerWinRT"):
         """
         Stop the server
         """

test/backends/bluezdbus/test_utils.py

@@ -0,0 +1,24 @@
+import sys
+import pytest
+
+from typing import List
+
+if sys.platform.lower() != "linux":
+    pytest.skip("Only for linux", allow_module_level=True)
+
+from dbus_next.aio import MessageBus
+from dbus_next.constants import BusType
+
+from bless.backends.bluezdbus.dbus.utils import list_adapters
+
+hardware_only = pytest.mark.skipif("os.environ.get('TEST_HARDWARE') is None")
+
+
+@hardware_only
+class TestUtils:
+
+    @pytest.mark.asyncio
+    async def test_list_adapters(self):
+        bus: MessageBus = await MessageBus(bus_type=BusType.SYSTEM).connect()
+        adapters: List[str] = await list_adapters(bus)
+        assert len(adapters) > 0