feat: initial import (#1)

.github/workflows/ci.yml

@@ -36,15 +36,11 @@ jobs:
       fail-fast: false
       matrix:
         python-version:
-          - "3.8"
-          - "3.9"
           - "3.10"
           - "3.11"
           - "3.12"
         os:
           - ubuntu-latest
-          - windows-latest
-          - macOS-latest
     runs-on: ${{ matrix.os }}
     steps:
       - uses: actions/checkout@v4

.pre-commit-config.yaml

@@ -45,10 +45,6 @@ repos:
     rev: 23.11.0
     hooks:
       - id: black
-  - repo: https://github.com/codespell-project/codespell
-    rev: v2.2.6
-    hooks:
-      - id: codespell
   - repo: https://github.com/pre-commit/mirrors-mypy
     rev: v1.7.1
     hooks:

pyproject.toml

@@ -23,7 +23,11 @@ packages = [
 "Changelog" = "https://github.com/bluetooth-devices/bleak-esphome/blob/main/CHANGELOG.md"
 
 [tool.poetry.dependencies]
-python = "^3.8"
+python = ">=3.10,<3.13"
+aioesphomeapi = ">=21.0.0"
+bleak = ">=0.21.1"
+bluetooth-data-tools = ">=1.17.0"
+habluetooth = ">=1.0.0"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7.0"

src/bleak_esphome/__init__.py

@@ -1 +0,0 @@
-__version__ = "0.0.0"

src/bleak_esphome/backend/cache.py

@@ -0,0 +1,50 @@
+"""Bluetooth cache for esphome."""
+from __future__ import annotations
+
+from collections.abc import MutableMapping
+from dataclasses import dataclass, field
+
+from bleak.backends.service import BleakGATTServiceCollection
+from lru import LRU  # pylint: disable=no-name-in-module
+
+MAX_CACHED_SERVICES = 128
+
+
+@dataclass(slots=True)
+class ESPHomeBluetoothCache:
+    """Shared cache between all ESPHome bluetooth devices."""
+
+    _gatt_services_cache: MutableMapping[int, BleakGATTServiceCollection] = field(
+        default_factory=lambda: LRU(MAX_CACHED_SERVICES)
+    )
+    _gatt_mtu_cache: MutableMapping[int, int] = field(
+        default_factory=lambda: LRU(MAX_CACHED_SERVICES)
+    )
+
+    def get_gatt_services_cache(
+        self, address: int
+    ) -> BleakGATTServiceCollection | None:
+        """Get the BleakGATTServiceCollection for the given address."""
+        return self._gatt_services_cache.get(address)
+
+    def set_gatt_services_cache(
+        self, address: int, services: BleakGATTServiceCollection
+    ) -> None:
+        """Set the BleakGATTServiceCollection for the given address."""
+        self._gatt_services_cache[address] = services
+
+    def clear_gatt_services_cache(self, address: int) -> None:
+        """Clear the BleakGATTServiceCollection for the given address."""
+        self._gatt_services_cache.pop(address, None)
+
+    def get_gatt_mtu_cache(self, address: int) -> int | None:
+        """Get the mtu cache for the given address."""
+        return self._gatt_mtu_cache.get(address)
+
+    def set_gatt_mtu_cache(self, address: int, mtu: int) -> None:
+        """Set the mtu cache for the given address."""
+        self._gatt_mtu_cache[address] = mtu
+
+    def clear_gatt_mtu_cache(self, address: int) -> None:
+        """Clear the mtu cache for the given address."""
+        self._gatt_mtu_cache.pop(address, None)

src/bleak_esphome/backend/characteristic.py

@@ -0,0 +1,96 @@
+"""BleakGATTCharacteristicESPHome."""
+from __future__ import annotations
+
+import contextlib
+from uuid import UUID
+
+from aioesphomeapi.model import BluetoothGATTCharacteristic
+from bleak.backends.characteristic import BleakGATTCharacteristic
+from bleak.backends.descriptor import BleakGATTDescriptor
+
+PROPERTY_MASKS = {
+    2**n: prop
+    for n, prop in enumerate(
+        (
+            "broadcast",
+            "read",
+            "write-without-response",
+            "write",
+            "notify",
+            "indicate",
+            "authenticated-signed-writes",
+            "extended-properties",
+            "reliable-writes",
+            "writable-auxiliaries",
+        )
+    )
+}
+
+
+class BleakGATTCharacteristicESPHome(BleakGATTCharacteristic):
+    """GATT Characteristic implementation for the ESPHome backend."""
+
+    obj: BluetoothGATTCharacteristic
+
+    def __init__(
+        self,
+        obj: BluetoothGATTCharacteristic,
+        max_write_without_response_size: int,
+        service_uuid: str,
+        service_handle: int,
+    ) -> None:
+        """Init a BleakGATTCharacteristicESPHome."""
+        super().__init__(obj, max_write_without_response_size)
+        self.__descriptors: list[BleakGATTDescriptor] = []
+        self.__service_uuid: str = service_uuid
+        self.__service_handle: int = service_handle
+        char_props = self.obj.properties
+        self.__props: list[str] = [
+            prop for mask, prop in PROPERTY_MASKS.items() if char_props & mask
+        ]
+
+    @property
+    def service_uuid(self) -> str:
+        """Uuid of the Service containing this characteristic."""
+        return self.__service_uuid
+
+    @property
+    def service_handle(self) -> int:
+        """Integer handle of the Service containing this characteristic."""
+        return self.__service_handle
+
+    @property
+    def handle(self) -> int:
+        """Integer handle for this characteristic."""
+        return self.obj.handle
+
+    @property
+    def uuid(self) -> str:
+        """Uuid of this characteristic."""
+        return self.obj.uuid
+
+    @property
+    def properties(self) -> list[str]:
+        """Properties of this characteristic."""
+        return self.__props
+
+    @property
+    def descriptors(self) -> list[BleakGATTDescriptor]:
+        """List of descriptors for this service."""
+        return self.__descriptors
+
+    def get_descriptor(self, specifier: int | str | UUID) -> BleakGATTDescriptor | None:
+        """Get a descriptor by handle (int) or UUID (str or uuid.UUID)."""
+        with contextlib.suppress(StopIteration):
+            if isinstance(specifier, int):
+                return next(filter(lambda x: x.handle == specifier, self.descriptors))
+            return next(filter(lambda x: x.uuid == str(specifier), self.descriptors))
+        return None
+
+    def add_descriptor(self, descriptor: BleakGATTDescriptor) -> None:
+        """
+        Add a :py:class:`~BleakGATTDescriptor` to the characteristic.
+
+        Should not be used by end user, but rather by `bleak` itself.
+        """
+        self.__descriptors.append(descriptor)