Trying to apply changes from CircuitPython

Author: o-murphy

bleak/__init__.py

@@ -7,6 +7,9 @@
 __author__ = """Henrik Blidh"""
 __email__ = "[email protected]"
 
+from bleak.backends.circuitpython import patches
+patches.apply_patches()
+
 import asyncio
 import functools
 import inspect
@@ -17,22 +20,25 @@
 from collections.abc import AsyncGenerator, Awaitable, Callable, Iterable
 from types import TracebackType
 from typing import Any, Literal, Optional, TypedDict, Union, cast, overload
-
-if sys.version_info < (3, 12):
-    from typing_extensions import Buffer
-else:
-    from collections.abc import Buffer
-
-if sys.version_info < (3, 11):
-    from async_timeout import timeout as async_timeout
-    from typing_extensions import Never, Self, Unpack, assert_never
-else:
-    from asyncio import timeout as async_timeout
-    from typing import Never, Self, Unpack, assert_never
-
-from bleak.args.bluez import BlueZScannerArgs
-from bleak.args.corebluetooth import CBScannerArgs, CBStartNotifyArgs
-from bleak.args.winrt import WinRTClientArgs
+from typing import TYPE_CHECKING
+
+if sys.implementation.name == "cpython":
+    if sys.version_info < (3, 12):
+        from typing_extensions import Buffer
+    else:
+        from collections.abc import Buffer
+
+    if sys.version_info < (3, 11):
+        from async_timeout import timeout as async_timeout
+        from typing_extensions import Never, Self, Unpack, assert_never
+    else:
+        from asyncio import timeout as async_timeout
+        from typing import Never, Self, Unpack, assert_never
+
+if TYPE_CHECKING:
+    from bleak.args.bluez import BlueZScannerArgs
+    from bleak.args.corebluetooth import CBScannerArgs, CBStartNotifyArgs
+    from bleak.args.winrt import WinRTClientArgs
 from bleak.backends.characteristic import BleakGATTCharacteristic
 from bleak.backends.client import BaseBleakClient, get_platform_client_backend_type
 from bleak.backends.descriptor import BleakGATTDescriptor
@@ -50,17 +56,18 @@
 
 _logger = logging.getLogger(__name__)
 _logger.addHandler(logging.NullHandler())
-if bool(os.environ.get("BLEAK_LOGGING", False)):
-    FORMAT = "%(asctime)-15s %(name)-8s %(threadName)s %(levelname)s: %(message)s"
-    handler = logging.StreamHandler(sys.stderr)
-    handler.setLevel(logging.DEBUG)
-    handler.setFormatter(logging.Formatter(fmt=FORMAT))
-    _logger.addHandler(handler)
-    _logger.setLevel(logging.DEBUG)
+if sys.implementation.name == "cpython":
+    if bool(os.environ.get("BLEAK_LOGGING", False)):
+        FORMAT = "%(asctime)-15s %(name)-8s %(threadName)s %(levelname)s: %(message)s"
+        handler = logging.StreamHandler(sys.stderr)
+        handler.setLevel(logging.DEBUG)
+        handler.setFormatter(logging.Formatter(fmt=FORMAT))
+        _logger.addHandler(handler)
+        _logger.setLevel(logging.DEBUG)
 
 
 # prevent tasks from being garbage collected
-_background_tasks = set[asyncio.Task[None]]()
+_background_tasks: set[asyncio.Task[None]] = set()
 
 
 class BleakScanner:
@@ -170,7 +177,7 @@ async def advertisement_data(
 
         .. versionadded:: 0.21
         """
-        devices = asyncio.Queue[tuple[BLEDevice, AdvertisementData]]()
+        devices: asyncio.Queue[tuple[BLEDevice, AdvertisementData]] = asyncio.Queue()
 
         unregister_callback = self._backend.register_detection_callback(
             lambda bd, ad: devices.put_nowait((bd, ad))
@@ -181,7 +188,7 @@ async def advertisement_data(
         finally:
             unregister_callback()
 
-    class ExtraArgs(TypedDict, total=False):
+    class ExtraArgs(TypedDict):
         """
         Keyword args from :class:`~bleak.BleakScanner` that can be passed to
         other convenience methods.
@@ -372,7 +379,7 @@ async def find_device_by_filter(
                     async for bd, ad in scanner.advertisement_data():
                         if filterfunc(bd, ad):
                             return bd
-                    assert_never(cast(Never, "advertisement_data() should never stop"))
+                    # assert_never(cast(Never, "advertisement_data() should never stop"))
             except asyncio.TimeoutError:
                 return None
 

bleak/backends/circuitpython/__init__.py

@@ -0,0 +1,57 @@
+import sys
+
+try:
+    from typing import NamedTuple as _typing_namedtuple
+except ImportError:
+    _typing_namedtuple = None
+
+
+def circuit_namedtuple(cls=None, *, annotations=None):
+    if sys.implementation.name != "circuitpython":
+        def passthrough(cls):
+            if _typing_namedtuple is not None:
+                return _typing_namedtuple(cls.__name__, [(k, v) for k, v in getattr(cls, "__annotations__", {}).items()])
+            else:
+                from collections import namedtuple as _collections_namedtuple
+                fields = list(getattr(cls, "__annotations__", {}).keys())
+                return _collections_namedtuple(cls.__name__, fields)
+        return passthrough if cls is None else passthrough(cls)
+
+    # ---- CircuitPython ----
+    def wrap(cls):
+        nonlocal annotations
+        if annotations is None:
+            raise TypeError("You must provide annotations in CircuitPython")
+        fields = list(annotations.keys())
+
+        class NT:
+            __slots__ = fields
+
+            def __init__(self, **kwargs):
+                for f in fields:
+                    if f in kwargs:
+                        setattr(self, f, kwargs.pop(f))
+                    else:
+                        raise TypeError(f"Missing field {f}")
+                if kwargs:
+                    raise TypeError(f"Unexpected fields {list(kwargs.keys())}")
+
+            def __repr__(self):
+                values = ", ".join(f"{f}={getattr(self,f)!r}" for f in fields)
+                return f"{cls.__name__}({values})"
+
+        return NT
+
+    return wrap if cls is None else wrap(cls)
+
+def circuit_advertisement_data_patch(cls=None):
+    annotations = {
+        "local_name": "Optional[str]",
+        "manufacturer_data": "dict[int, bytes]",
+        "service_data": "dict[str, bytes]",
+        "service_uuids": "list[str]",
+        "tx_power": "Optional[int]",
+        "rssi": "int",
+        "platform_data": "tuple[Any, ...]",
+    }
+    return circuit_namedtuple(cls, annotations=annotations)

bleak/backends/circuitpython/_advertisement_data.py

@@ -0,0 +1,153 @@
+import logging
+from typing import Optional, Any
+from typing_extensions import override
+
+from _bleio import set_adapter
+from adafruit_ble import BLERadio, Advertisement, BLEConnection
+
+from bleak import BleakScanner
+from bleak.backends.circuitpython.scanner import BleakScannerCircuitPython
+from bleak.backends.client import BaseBleakClient
+from bleak.backends.descriptor import BleakGATTDescriptor
+from bleak.backends.device import BLEDevice
+from bleak.exc import BleakError, BleakDeviceNotFoundError
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.DEBUG)
+
+class BleakClientCircuitPython(BaseBleakClient):
+    def __init__(
+        self,
+        address_or_ble_device: BLEDevice,
+        services=None,
+        **kwargs,
+    ):
+        super().__init__(address_or_ble_device, **kwargs)
+        _adapter = kwargs.get("adapter")
+        if _adapter is not None:
+            set_adapter(_adapter)
+
+        self._radio: Optional[BLERadio] = None
+        self._advertisement: Optional[Advertisement] = None
+
+        if isinstance(address_or_ble_device, BLEDevice):
+            self._radio, self._advertisement = address_or_ble_device.details
+
+        self._connection: Optional[BLEConnection] = None
+        # self._services = None
+        # self._is_connected = False
+        # self._mtu = 23
+
+    @override
+    async def connect(self, pair, dangerous_use_bleak_cache=False, **kwargs):
+        logger.debug("Attempting to connect BLE device @ {}".format(self.address))
+
+        if self.is_connected:
+            raise BleakError("Client is already connected")
+
+        if not self._advertisement.connectable:
+            raise BleakError("Device is not connectable")
+
+        if pair:
+            raise NotImplementedError("Not yet implemented")
+
+        timeout = kwargs.get("timeout", self._timeout)
+
+        if self._advertisement is None:
+            logger.debug("Attempting to find BLE device @ {}".format(self.address))
+
+            device = await BleakScanner.find_device_by_address(
+                self.address, timeout=timeout, backend=BleakScannerCircuitPython
+            )
+            if device:
+                self._radio, self._advertisement = device.details
+            else:
+                raise BleakDeviceNotFoundError(
+                    self.address, f"Device @ {self.address} was not found"
+                )
+
+        if self._radio is None:
+            self._radio = BLERadio()
+
+        # TODO: disconnect_callback ?
+
+        logger.debug("Connecting to BLE device @ {}".format(self.address))
+
+        # TODO: wrap async
+        self._connection = self._radio.connect(self._advertisement.address, timeout=timeout)
+        logger.debug("Connected to BLE device @ {}".format(self.address))
+
+        logger.debug("Retrieving services from BLE device @ {}".format(self.address))
+
+        # TODO: get services
+
+        logger.debug("Services retrieved from BLE device @ {}".format(self.address))
+
+
+    async def disconnect(self) -> None:
+        """Disconnect from the peripheral device"""
+        logger.debug("Disconnecting from BLE device @ {}".format(self.address))
+        if (
+            self._radio is None
+            or self._advertisement is None
+            or not self.is_connected
+        ):
+            logger.debug("Device is not connected @ {}".format(self.address))
+            return
+
+        # TODO: wrap async
+        self._connection.disconnect()
+        logger.debug("Device disconnected @ {}".format(self.address))
+
+    @property
+    @override
+    def is_connected(self) -> bool:
+        return self._connection is not None and self._connection.connected
+
+    @property
+    @override
+    def mtu_size(self) -> int:
+        """Get ATT MTU size for active connection"""
+        # TODO: implement
+        raise NotImplementedError()
+
+    @override
+    async def pair(self, *args: Any, **kwargs: Any) -> None:
+        raise NotImplementedError()
+
+    @override
+    async def unpair(self) -> None:
+        raise NotImplementedError()
+
+    @override
+    async def read_gatt_char(self, characteristic, **kwargs):
+        ...
+
+    @override
+    async def read_gatt_descriptor(self, descriptor: BleakGATTDescriptor, **kwargs):
+        ...
+
+    @override
+    async def write_gatt_char(self, characteristic, data, response):
+        ...
+
+    @override
+    async def write_gatt_descriptor(self, descriptor, data):
+        ...
+
+    @override
+    async def start_notify(
+        self,
+        characteristic,
+        callback,
+        **kwargs,
+    ):
+        ...
+
+    @override
+    async def stop_notify(self, characteristic):
+        ...
+
+    async def get_rssi(self) -> int:
+        assert self._advertisement
+        return self._advertisement.rssi

bleak/backends/circuitpython/client.py

@@ -0,0 +1,6 @@
+from adafruit_ble import BLERadio
+
+
+class Manager:
+    def __init__(self) -> None:
+        pass

bleak/backends/circuitpython/manager.py

@@ -0,0 +1,41 @@
+"""Monkeypatch absent modules in CircuitPython"""
+import sys
+
+
+def apply_patches():
+    if sys.implementation.name == "circuitpython":
+        import builtins
+        assert builtins
+        import asyncio
+        assert asyncio
+
+        from . import _fake_enum
+        from . import _fake_types
+        from . import _fake_typing
+        from . import _inspect
+        from . import _platform
+        from . import _fake_abc
+        from . import _fake_collections_abc
+        from . import _uuid
+        from . import _deprecation_warn
+        from ._async_timeout import async_timeout as _async_timeout
+        import adafruit_logging as _logging
+        import circuitpython_functools as _functools
+
+        if 'builtins' not in sys.modules:
+            sys.modules['builtins'] = builtins
+        sys.modules["builtins"].DeprecationWarning = _deprecation_warn.DeprecationWarning
+        sys.modules["enum"] = _fake_enum
+        sys.modules["inspect"] = _inspect
+        sys.modules["platform"] = _platform
+        sys.modules["uuid"] = _uuid
+        sys.modules["types"] = _fake_types
+        sys.modules["typing"] = _fake_typing
+        sys.modules["typing_extensions"] = _fake_typing
+        sys.modules["logging"] = _logging
+        sys.modules["abc"] = _fake_abc
+        sys.modules["collections.abc"] = _fake_collections_abc
+        sys.modules['functools'] = _functools
+        sys.modules['asyncio'].timeout = _async_timeout
+
+        print("patched")