BlessAdvertisementData for starting a server

bless/backends/advertisement.py

@@ -0,0 +1,67 @@
+import platform
+import warnings
+from dataclasses import dataclass
+from typing import Dict, List, Optional
+
+
+@dataclass
+class BlessAdvertisementData:
+    """
+    Generic advertisement data for BLE backends.
+    """
+
+    local_name: Optional[str] = None
+    service_uuids: Optional[List[str]] = None
+    manufacturer_data: Optional[Dict[int, bytes]] = None
+    service_data: Optional[Dict[str, bytes]] = None
+    is_connectable: Optional[bool] = None
+    is_discoverable: Optional[bool] = None
+    tx_power: Optional[int] = None
+
+    def __post_init__(self) -> None:
+        """
+        Warn when fields are provided that are not used by the current OS backend.
+        """
+        system = platform.system()
+        if system == "Darwin":
+            unused = self._unused_fields({"local_name", "service_uuids"})
+        elif system == "Windows":
+            unused = self._unused_fields(
+                {"local_name", "is_connectable", "is_discoverable"}
+            )
+        elif system == "Linux":
+            unused = self._unused_fields(
+                {
+                    "local_name",
+                    "service_uuids",
+                    "manufacturer_data",
+                    "service_data",
+                    "tx_power",
+                }
+            )
+        else:
+            unused = self._unused_fields(set())
+
+        if unused:
+            unused_list = ", ".join(sorted(unused))
+            warnings.warn(
+                f"Advertisement fields not used on {system}: {unused_list}",
+                RuntimeWarning,
+            )
+
+    def _unused_fields(self, used: set) -> set:
+        """
+        Return the provided field names that are not in the OS-used set.
+        """
+        provided = {
+            "local_name": self.local_name is not None,
+            "service_uuids": self.service_uuids is not None,
+            "manufacturer_data": self.manufacturer_data is not None,
+            "service_data": self.service_data is not None,
+            "is_connectable": self.is_connectable is not None,
+            "is_discoverable": self.is_discoverable is not None,
+            "tx_power": self.tx_power is not None,
+        }
+        return {
+            name for name, present in provided.items() if present and name not in used
+        }

bless/backends/bluezdbus/dbus/application.py

@@ -8,6 +8,7 @@
 from dbus_next.service import ServiceInterface  # type: ignore
 from dbus_next.signature import Variant  # type: ignore
 
+from bless.backends.advertisement import BlessAdvertisementData
 from bless.backends.bluezdbus.dbus.advertisement import (  # type: ignore
     Type,
     BlueZLEAdvertisement,
@@ -150,24 +151,44 @@ async def unregister(self, adapter: ProxyObject):
         iface: ProxyInterface = adapter.get_interface(defs.GATT_MANAGER_INTERFACE)
         await iface.call_unregister_application(self.path)  # type: ignore
 
-    async def start_advertising(self, adapter: ProxyObject):
+    async def start_advertising(
+        self,
+        adapter: ProxyObject,
+        advertisement_data: Optional[BlessAdvertisementData] = None,
+    ):
         """
         Start Advertising the application
 
         Parameters
         ----------
         adapter : ProxyObject
             The adapter object to start advertising on
+        advertisement_data : Optional[BlessAdvertisementData]
+            Optional advertisement payload to populate BlueZ advertisement data
         """
-        await self.set_name(adapter, self.app_name)
+        local_name: str = self.app_name
+        if advertisement_data and advertisement_data.local_name is not None:
+            local_name = advertisement_data.local_name
+        await self.set_name(adapter, local_name)
 
         advertisement: BlueZLEAdvertisement = BlueZLEAdvertisement(
             Type.PERIPHERAL, len(self.advertisements) + 1, self
         )
         self.advertisements.append(advertisement)
 
-        # Only add the first UUID
-        advertisement._service_uuids.append(self.services[0].UUID)
+        if advertisement_data and advertisement_data.local_name is not None:
+            advertisement._local_name = advertisement_data.local_name
+        if advertisement_data and advertisement_data.service_uuids is not None:
+            advertisement._service_uuids.extend(advertisement_data.service_uuids)
+        elif len(self.services) > 0:
+            # Only add the first UUID
+            advertisement._service_uuids.append(self.services[0].UUID)
+        if advertisement_data and advertisement_data.manufacturer_data is not None:
+            advertisement._manufacturer_data = advertisement_data.manufacturer_data
+        if advertisement_data and advertisement_data.service_data is not None:
+            advertisement._service_data = advertisement_data.service_data
+        if advertisement_data and advertisement_data.tx_power is not None:
+            advertisement._tx_power = advertisement_data.tx_power
 
         self.bus.export(advertisement.path, advertisement)
 

bless/backends/bluezdbus/server.py

@@ -10,6 +10,7 @@
 from dbus_next.constants import BusType  # type: ignore
 
 from bless.backends.server import BaseBlessServer  # type: ignore
+from bless.backends.advertisement import BlessAdvertisementData
 from bless.backends.bluezdbus.characteristic import BlessGATTCharacteristicBlueZDBus
 from bless.backends.bluezdbus.descriptor import BlessGATTDescriptorBlueZDBus
 from bless.backends.bluezdbus.dbus.application import (  # type: ignore
@@ -79,10 +80,17 @@ async def setup(self: "BlessServerBlueZDBus"):
             raise Exception("Could not locate bluetooth adapter")
         self.adapter: ProxyObject = cast(ProxyObject, potential_adapter)
 
-    async def start(self, **kwargs) -> bool:
+    async def start(
+        self, advertisement_data: Optional[BlessAdvertisementData] = None, **kwargs
+    ) -> bool:
         """
         Start the server
 
+        Parameters
+        ----------
+        advertisement_data : Optional[BlessAdvertisementData]
+            Optional advertisement payload to customize BlueZ advertising data
+
         Returns
         -------
         bool
@@ -97,7 +105,9 @@ async def start(self, **kwargs) -> bool:
         await self.app.register(self.adapter)
 
         # advertise
-        await self.app.start_advertising(self.adapter)
+        await self.app.start_advertising(
+            self.adapter, advertisement_data=advertisement_data
+        )
 
         return True
 

bless/backends/corebluetooth/server.py

@@ -13,18 +13,20 @@
     CBMutableDescriptor,
     CBAdvertisementDataLocalNameKey,
     CBAdvertisementDataServiceUUIDsKey,
+    CBUUID,
 )
 
 from bleak.backends.service import BleakGATTService  # type: ignore
 
 from .peripheral_manager_delegate import PeripheralManagerDelegate  # type: ignore
 from bless.backends.server import BaseBlessServer  # type: ignore
+from bless.backends.advertisement import BlessAdvertisementData
 from bless.backends.corebluetooth.service import BlessGATTServiceCoreBluetooth
 from bless.backends.corebluetooth.characteristic import (  # type: ignore
     BlessGATTCharacteristicCoreBluetooth,
 )
 from bless.backends.corebluetooth.descriptor import (  # type: ignore
-    BlessGATTDescriptorCoreBluetooth
+    BlessGATTDescriptorCoreBluetooth,
 )
 
 from bless.backends.descriptor import (  # type: ignore
@@ -71,7 +73,11 @@ def __init__(self, name: str, loop: Optional[AbstractEventLoop] = None, **kwargs
         self.peripheral_manager_delegate.write_request_func = self.write_request
 
     async def start(
-        self, timeout: float = 10, prioritize_local_name: bool = True, **kwargs
+        self,
+        advertisement_data: Optional[BlessAdvertisementData] = None,
+        timeout: float = 10,
+        prioritize_local_name: bool = True,
+        **kwargs,
     ):
         """
         Start the server
@@ -87,28 +93,42 @@ async def start(
             names associated with BLE applications. When true, the name of the
             server is prioritized over service UUIDs, and will automatrically
             be truncated if longer than 28 bytes.
+        advertisement_data : Optional[BlessAdvertisementData]
+            Optional advertisement payload to customize the local name and
+            service UUIDs advertised
         """
         for service_uuid in self.services:
             bleak_service: BleakGATTService = self.services[service_uuid]
             service_obj: CBService = bleak_service.obj
             logger.debug("Adding service: {}".format(bleak_service.uuid))
             await self.peripheral_manager_delegate.add_service(service_obj)
 
+        local_name: str = self.name
+        if advertisement_data and advertisement_data.local_name is not None:
+            local_name = advertisement_data.local_name
+
         advertisement_uuids: List
-        if (prioritize_local_name) and len(self.name) > 10:
+        if advertisement_data and advertisement_data.service_uuids is not None:
+            advertisement_uuids = [
+                CBUUID.alloc().initWithString_(uuid)
+                for uuid in advertisement_data.service_uuids
+            ]
+        elif (prioritize_local_name) and len(local_name) > 10:
             advertisement_uuids = []
         else:
             advertisement_uuids = list(
                 map(lambda x: self.services[x].obj.UUID(), self.services)
             )
 
-        advertisement_data = {
-            CBAdvertisementDataLocalNameKey: self.name,
+        advertisement_payload = {
+            CBAdvertisementDataLocalNameKey: local_name,
             CBAdvertisementDataServiceUUIDsKey: advertisement_uuids,
         }
-        logger.debug("Advertisement Data: {}".format(advertisement_data))
+        logger.debug("Advertisement Data: {}".format(advertisement_payload))
         try:
-            await self.peripheral_manager_delegate.start_advertising(advertisement_data)
+            await self.peripheral_manager_delegate.start_advertising(
+                advertisement_payload
+            )
         except TimeoutError:
             # If advertising fails as a result of bluetooth module power
             # cycling or advertisement failure, attempt to start again

bless/backends/server.py

@@ -7,6 +7,7 @@
 from typing import Any, Optional, Dict, Callable, List
 
 from bless.backends.service import BlessGATTService
+from bless.backends.advertisement import BlessAdvertisementData
 from bless.backends.attribute import (  # type: ignore
     GATTAttributePermissions
 )
@@ -52,10 +53,17 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
     # Abstract Methods
 
     @abc.abstractmethod
-    async def start(self, **kwargs) -> bool:
+    async def start(
+        self, advertisement_data: Optional[BlessAdvertisementData] = None, **kwargs
+    ) -> bool:
         """
         Start the server
 
+        Parameters
+        ----------
+        advertisement_data : Optional[BlessAdvertisementData]
+            Optional advertisement payload to customize backend advertising
+
         Returns
         -------
         bool

bless/backends/winrt/server.py

@@ -8,6 +8,7 @@
 from typing import Optional, List, Any, cast
 
 from bless.backends.server import BaseBlessServer  # type: ignore
+from bless.backends.advertisement import BlessAdvertisementData
 from bless.backends.attribute import (  # type: ignore
     GATTAttributePermissions,
 )
@@ -122,7 +123,11 @@ def __init__(
         self._adapter: BLEAdapter = BLEAdapter()
         self._name_overwrite: bool = name_overwrite
 
-    async def start(self: "BlessServerWinRT", **kwargs):
+    async def start(
+        self: "BlessServerWinRT",
+        advertisement_data: Optional[BlessAdvertisementData] = None,
+        **kwargs,
+    ):
         """
         Start the server
 
@@ -131,16 +136,27 @@ async def start(self: "BlessServerWinRT", **kwargs):
         timeout : float
             Floating point decimal in seconds for how long to wait for the
             on-board bluetooth module to power on
+        advertisement_data : Optional[BlessAdvertisementData]
+            Optional advertisement payload to customize local name and
+            connectable/discoverable settings
         """
 
-        if self._name_overwrite:
+        if advertisement_data and advertisement_data.local_name is not None:
+            self._adapter.set_local_name(advertisement_data.local_name)
+        elif self._name_overwrite:
             self._adapter.set_local_name(self.name)
 
         adv_parameters: GattServiceProviderAdvertisingParameters = (
             GattServiceProviderAdvertisingParameters()
         )
-        adv_parameters.is_discoverable = True
-        adv_parameters.is_connectable = True
+        if advertisement_data and advertisement_data.is_discoverable is not None:
+            adv_parameters.is_discoverable = advertisement_data.is_discoverable
+        else:
+            adv_parameters.is_discoverable = True
+        if advertisement_data and advertisement_data.is_connectable is not None:
+            adv_parameters.is_connectable = advertisement_data.is_connectable
+        else:
+            adv_parameters.is_connectable = True
 
         for uuid, service in self.services.items():
             winrt_service: BlessGATTServiceWinRT = cast(BlessGATTServiceWinRT, service)