Adding use_cached_services keyword to WinRT backend

This allows Bleak to use the Windows Bluetooth cache for services, characteristics and descriptors. Since Windows 11 took 10 seconds to fetch services on each connect, this will provide a faster connect on devices that do not change.

Fixes hbldh#686

Also modified Characteristic class in WinRT to use description property from super class in case of empty string description from Windows object.

The SensorTag example is a bit expanded to also light up a LED.

bleak/backends/winrt/characteristic.py

@@ -94,7 +94,11 @@ def uuid(self) -> str:
     @property
     def description(self) -> str:
         """Description for this characteristic"""
-        return self.obj.user_description
+        return (
+            self.obj.user_description
+            if self.obj.user_description
+            else super().description
+        )
 
     @property
     def properties(self) -> List[str]:

bleak/backends/winrt/client.py

@@ -113,10 +113,14 @@ class BleakClientWinRT(BaseBleakClient):
     a package that enables Python developers to access Windows Runtime APIs directly from Python.
 
     Args:
-        address_or_ble_device (`BLEDevice` or str): The Bluetooth address of the BLE peripheral to connect to or the `BLEDevice` object representing it.
+        address_or_ble_device (``BLEDevice`` or str): The Bluetooth address of the BLE peripheral
+        to connect to or the ``BLEDevice`` object representing it.
 
     Keyword Args:
         timeout (float): Timeout for required ``BleakScanner.find_device_by_address`` call. Defaults to 10.0.
+        use_cached_services (bool): ``True`` allows Windows to fetch the services, characteristics and descriptors
+            from the Windows cache instead of reading them from the device. Can be very much faster for known devices.
+            Defaults to ``False``.
 
     """
 
@@ -141,6 +145,7 @@ def __init__(self, address_or_ble_device: Union[BLEDevice, str], **kwargs):
             and kwargs["address_type"] in ("public", "random")
             else None
         )
+        self._use_cached_services = kwargs.get("use_cached_services", False)
 
         self._session_status_changed_token: Optional[EventRegistrationToken] = None
 
@@ -154,12 +159,17 @@ async def connect(self, **kwargs) -> bool:
 
         Keyword Args:
             timeout (float): Timeout for required ``BleakScanner.find_device_by_address`` call. Defaults to 10.0.
+            use_cached_services (bool): ``True`` allows Windows to fetch the services, characteristics and descriptors
+                from the Windows cache instead of reading them from the device. Can be very much faster for known
+                devices. Defaults to what was specified in the constructor, where default is ``False``.
 
         Returns:
             Boolean representing connection status.
 
         """
-
+        use_cached_services = kwargs.get(
+            "use_cached_services", self._use_cached_services
+        )
         # Try to find the desired device.
         timeout = kwargs.get("timeout", self._timeout)
         if self._device_info is None:
@@ -274,7 +284,7 @@ def session_status_changed_event_handler(
             self._session_active_events.remove(event)
 
         # Obtain services, which also leads to connection being established.
-        await self.get_services()
+        await self.get_services(use_cached_services=use_cached_services)
 
         return True
 
@@ -435,6 +445,12 @@ async def unpair(self) -> bool:
     async def get_services(self, **kwargs) -> BleakGATTServiceCollection:
         """Get all services registered for this GATT server.
 
+        Keyword Args:
+
+            use_cached_services (bool): ``True`` allows Windows to fetch the services, characteristics and descriptors
+                from the Windows cache instead of reading them from the device. Can be very much faster for known
+                devices. Defaults to what was specified in the constructor, where default is ``False``.
+
         Returns:
            A :py:class:`bleak.backends.service.BleakGATTServiceCollection` with this device's services tree.
 
@@ -444,9 +460,14 @@ async def get_services(self, **kwargs) -> BleakGATTServiceCollection:
             return self.services
         else:
             logger.debug("Get Services...")
+            use_cached_services = kwargs.get(
+                "use_cached_services", self._use_cached_services
+            )
             services: Sequence[GattDeviceService] = _ensure_success(
                 await self._requester.get_gatt_services_async(
-                    BluetoothCacheMode.UNCACHED
+                    BluetoothCacheMode.CACHED
+                    if use_cached_services
+                    else BluetoothCacheMode.UNCACHED
                 ),
                 "services",
                 "Could not get GATT services",
@@ -463,7 +484,9 @@ async def get_services(self, **kwargs) -> BleakGATTServiceCollection:
 
                 characteristics: Sequence[GattCharacteristic] = _ensure_success(
                     await service.get_characteristics_async(
-                        BluetoothCacheMode.UNCACHED
+                        BluetoothCacheMode.CACHED
+                        if use_cached_services
+                        else BluetoothCacheMode.UNCACHED
                     ),
                     "characteristics",
                     f"Could not get GATT characteristics for {service}",
@@ -476,7 +499,9 @@ async def get_services(self, **kwargs) -> BleakGATTServiceCollection:
 
                     descriptors: Sequence[GattDescriptor] = _ensure_success(
                         await characteristic.get_descriptors_async(
-                            BluetoothCacheMode.UNCACHED
+                            BluetoothCacheMode.CACHED
+                            if use_cached_services
+                            else BluetoothCacheMode.UNCACHED
                         ),
                         "descriptors",
                         f"Could not get GATT descriptors for {service}",

examples/sensortag.py

@@ -132,10 +132,11 @@ async def main(address):
         battery_level = await client.read_gatt_char(BATTERY_LEVEL_UUID)
         print("Battery Level: {0}%".format(int(battery_level[0])))
 
-        async def keypress_handler(sender, data):
+        async def notification_handler(sender, data):
             print("{0}: {1}".format(sender, data))
 
-        write_value = bytearray([0xA0])
+        # Turn on the red light on the Sensor Tag by writing to I/O Data and I/O Config.
+        write_value = bytearray([0x01])
         value = await client.read_gatt_char(IO_DATA_CHAR_UUID)
         print("I/O Data Pre-Write Value: {0}".format(value))
 
@@ -145,7 +146,19 @@ async def keypress_handler(sender, data):
         print("I/O Data Post-Write Value: {0}".format(value))
         assert value == write_value
 
-        await client.start_notify(KEY_PRESS_UUID, keypress_handler)
+        write_value = bytearray([0x01])
+        value = await client.read_gatt_char(IO_CONFIG_CHAR_UUID)
+        print("I/O Config Pre-Write Value: {0}".format(value))
+
+        await client.write_gatt_char(IO_CONFIG_CHAR_UUID, write_value)
+
+        value = await client.read_gatt_char(IO_CONFIG_CHAR_UUID)
+        print("I/O Config Post-Write Value: {0}".format(value))
+        assert value == write_value
+
+        # Try notifications with key presses.
+
+        await client.start_notify(KEY_PRESS_UUID, notification_handler)
         await asyncio.sleep(5.0)
         await client.stop_notify(KEY_PRESS_UUID)