Merge pull request #168 from kevincar/128-connect-multiple-clients-to-ble-server

128 connect multiple clients to ble server

Author: kevincar

bless/backends/bluezdbus/characteristic.py

@@ -130,6 +130,7 @@ def value(self) -> bytearray:
     def value(self, val: bytearray):
         """Set the value of the characteristic"""
         self._value = val
+        self.obj.Value = self._value
 
     @property
     def uuid(self) -> str:

bless/backends/bluezdbus/dbus/application.py

@@ -25,13 +25,26 @@
 
 LOGGER = logging.getLogger(__name__)
 
+ReadCallback = Callable[[BlueZGattCharacteristic, Dict[str, Any]], bytes]
+WriteCallback = Callable[[BlueZGattCharacteristic, bytes, Dict[str, Any]], None]
+SubscribeCallback = Callable[[BlueZGattCharacteristic, Dict[str, Any]], None]
+
 
 class BlueZGattApplication(ServiceInterface):
     """
     org.bluez.GattApplication1 interface implementation
     """
 
-    def __init__(self, name: str, destination: str, bus: MessageBus):
+    def __init__(
+        self,
+        name: str,
+        destination: str,
+        bus: MessageBus,
+        on_read: ReadCallback,
+        on_write: WriteCallback,
+        on_subscribe: SubscribeCallback,
+        on_unsubscribe: SubscribeCallback,
+    ):
         """
         Initialize a new GattApplication1
 
@@ -49,6 +62,10 @@ def __init__(self, name: str, destination: str, bus: MessageBus):
         self.app_name: str = name
         self.destination: str = destination
         self.bus: MessageBus = bus
+        self.Read: ReadCallback = on_read
+        self.Write: WriteCallback = on_write
+        self.StartNotify: SubscribeCallback = on_subscribe
+        self.StopNotify: SubscribeCallback = on_unsubscribe
 
         # Valid path must be ASCII characters "[A-Z][a-z][0-9]_"
         # see https://dbus.freedesktop.org/doc/dbus-specification.html#message-protocol-marshaling-object-path  # noqa E501
@@ -57,21 +74,6 @@ def __init__(self, name: str, destination: str, bus: MessageBus):
         self.advertisements: List[BlueZLEAdvertisement] = []
         self.services: List[BlueZGattService] = []
 
-        self.Read: Optional[
-            Callable[[BlueZGattCharacteristic, Dict[str, Any]], bytes]
-        ] = None
-        self.Write: Optional[
-            Callable[[BlueZGattCharacteristic, bytes, Dict[str, Any]], None]
-        ] = None
-        self.StartNotify: Optional[
-            Callable[[BlueZGattCharacteristic, Dict[str, Any]], None]
-        ] = None
-        self.StopNotify: Optional[
-            Callable[[BlueZGattCharacteristic, Dict[str, Any]], None]
-        ] = None
-
-        self.subscribed_characteristics: List[str] = []
-
         super(BlueZGattApplication, self).__init__(self.destination)
 
     @staticmethod
@@ -270,7 +272,16 @@ async def is_connected(self) -> bool:
             Whether a central device is subscribed to one of our
             characteristics
         """
-        return len(self.subscribed_characteristics) > 0
+        return (
+            sum(
+                [
+                    characteristic.Notifying
+                    for service in self.services
+                    for characteristic in service.characteristics
+                ]
+            )
+            > 0
+        )
 
     async def _register_object(
         self, o: Union[BlueZGattService, BlueZGattCharacteristic, BlueZGattDescriptor]

bless/backends/bluezdbus/dbus/characteristic.py

@@ -1,20 +1,26 @@
-from enum import Enum
+import asyncio
+import logging
+import os
 
 import bleak.backends.bluezdbus.defs as defs  # type: ignore
 
-from typing import List, TYPE_CHECKING, Any, Dict
-
-from dbus_next.service import ServiceInterface, method, dbus_property  # type: ignore
+from dbus_next import DBusError  # type: ignore
 from dbus_next.constants import PropertyAccess  # type: ignore
+from dbus_next.service import ServiceInterface, method, dbus_property  # type: ignore
 from dbus_next.signature import Variant  # type: ignore
+from enum import Enum
+from typing import List, TYPE_CHECKING, Any, Dict
 
 from .descriptor import BlueZGattDescriptor, DescriptorFlags  # type: ignore
+from .session import NotifySession  # type: ignore
 
 if TYPE_CHECKING:
     from bless.backends.bluezdbus.dbus.service import (  # type: ignore # noqa: F401
         BlueZGattService,
     )
 
+logger = logging.getLogger(name=__name__)
+
 
 class Flags(Enum):
     BROADCAST = "broadcast"
@@ -68,7 +74,8 @@ def __init__(
         self._service: "BlueZGattService" = service  # noqa: F821
 
         self._value: bytes = b""
-        self._notifying: bool = "notify" in self._flags or "indicate" in self._flags
+        self._notifying_calls: int = 0
+        self._subscribed_centrals: Dict[str, NotifySession] = {}
         self.descriptors: List["BlueZGattDescriptor"] = []  # noqa: F821
 
         super(BlueZGattCharacteristic, self).__init__(self.interface_name)
@@ -87,17 +94,22 @@ def Value(self) -> "ay":  # type: ignore # noqa: F821 N802
 
     @Value.setter  # type: ignore
     def Value(self, value: "ay"):  # type: ignore # noqa: F821 N802
+        if isinstance(value, bytearray):
+            value = bytes(value)
         self._value = value
-        self.emit_properties_changed(changed_properties={"Value": self._value})
 
     @dbus_property(access=PropertyAccess.READ)
     def Notifying(self) -> "b":  # type: ignore # noqa: F821 N802
-        return self._notifying
+        return self._notifying_calls > 0 or self.NotifyAcquired
 
     @dbus_property(access=PropertyAccess.READ)  # noqa: F722
     def Flags(self) -> "as":  # type: ignore # noqa: F821 F722 N802
         return self._flags
 
+    @dbus_property(access=PropertyAccess.READ)  # noqa: F722
+    def NotifyAcquired(self) -> "b":  # type: ignore # noqa: F821
+        return len(self._subscribed_centrals) > 0
+
     @method()  # noqa: F722
     def ReadValue(self, options: "a{sv}") -> "ay":  # type: ignore # noqa: F722 F821 N802 E501
         """
@@ -137,27 +149,102 @@ def WriteValue(self, value: "ay", options: "a{sv}"):  # type: ignore # noqa
             raise NotImplementedError()
         f(self, value, options)
 
+    @method()
+    async def AcquireNotify(self, options: "a{sv}") -> "hq":  # type: ignore # noqa
+        """
+        Called when a central device subscribes to the
+        characteristic
+        """
+        mtu: int = options["mtu"].value
+        potential_device: Variant = options["device"]
+        device_path: str = potential_device.value
+
+        # Can only process this if we are not already subscribed
+        if self.Notifying and not self.NotifyAcquired:
+            logger.error("AcquireNotify attempted after StartNotify called")
+            raise DBusError(
+                "org.bluez.Error.NotPermitted", "AcquireNotify not permitted"
+            )
+
+        session: NotifySession = NotifySession(
+            device_path, mtu, self._service.app.bus, self.ReleaseNotify
+        )
+        rx: int = await session.start()
+        address: str = await session.get_device_address()
+        logger.debug(f"AcquireNotify on {self.UUID} from {address} on FD {rx}")
+
+        f = self._service.app.StartNotify
+        if f is None:
+            raise NotImplementedError()
+        f(self, {"device": address})
+        self._subscribed_centrals[address] = session
+
+        async def close_rx():
+            logger.debug("Closing RX")
+            await asyncio.sleep(2)
+            os.close(rx)
+            # asyncio.get_running_loop().call_soon_threadsafe(os.close, rx)
+
+        asyncio.create_task(close_rx())
+        return [rx, mtu]
+
+    async def ReleaseNotify(self, session: NotifySession):
+        address: str = await session.get_device_address()
+        logger.debug(f"ReleaseNotify on {self.UUID} from {address}")
+        f = self._service.app.StopNotify
+        if f is None:
+            raise NotImplementedError()
+        f(self, {"device": address})
+        del self._subscribed_centrals[address]
+
     @method()
     def StartNotify(self):  # noqa: N802
         """
         Begin a subscription to the characteristic
         """
+        if self.NotifyAcquired:
+            logger.info(
+                "StartNotify called. "
+                + "AcquireNotify already called. "
+                + "Ignoring call to Start Notify"
+            )
+            return
+
+        logger.debug(f"StartNotify on {self.UUID}")
         f = self._service.app.StartNotify
         if f is None:
             raise NotImplementedError()
         f(self, {})
-        self._service.app.subscribed_characteristics.append(self._uuid)
+        self._notifying_calls += 1
 
     @method()
-    def StopNotify(self):  # noqa: N802
+    async def StopNotify(self):  # noqa: N802
         """
         Stop a subscription to the characteristic
         """
+        if self.NotifyAcquired:
+            logger.error("StopNotify called but notifications are aquried!")
+            return
+
         f = self._service.app.StopNotify
         if f is None:
             raise NotImplementedError()
         f(self, {})
-        self._service.app.subscribed_characteristics.remove(self._uuid)
+        self._notifying_calls -= 1
+
+    def update_value(self) -> None:
+        """
+        This method does not actually alter that value of the characteristic,
+        but rather sends updates to subscribed centrals.
+        """
+        if self.NotifyAcquired is True:
+            for central_id, session in self._subscribed_centrals.items():
+                logger.debug(f"Sending update to {central_id}")
+                if not session.send_update(self._value):
+                    logger.warn(f"Failed to send update to {central_id}")
+
+        else:
+            self.emit_properties_changed(changed_properties={"Value": self._value})
 
     async def add_descriptor(
         self, uuid: str, flags: List[DescriptorFlags], value: Any