Additional typehints for corebluetooth

bleak/backends/corebluetooth/CentralManagerDelegate.py

@@ -10,7 +10,7 @@
 import logging
 import sys
 import threading
-from typing import Any, Callable, Dict, Optional
+from typing import Any, Callable, Dict, List, Optional
 
 if sys.version_info < (3, 11):
     from async_timeout import timeout as async_timeout
@@ -103,7 +103,7 @@ def init(self) -> Optional["CentralManagerDelegate"]:
 
         return self
 
-    def __del__(self):
+    def __del__(self) -> None:
         if objc.macos_available(10, 13):
             try:
                 self.central_manager.removeObserver_forKeyPath_(self, "isScanning")
@@ -116,7 +116,7 @@ def __del__(self):
     # User defined functions
 
     @objc.python_method
-    async def start_scan(self, service_uuids) -> None:
+    async def start_scan(self, service_uuids: Optional[List[str]]) -> None:
         service_uuids = (
             NSArray.alloc().initWithArray_(
                 list(map(CBUUID.UUIDWithString_, service_uuids))
@@ -158,7 +158,7 @@ async def connect(
         self,
         peripheral: CBPeripheral,
         disconnect_callback: DisconnectCallback,
-        timeout=10.0,
+        timeout: float = 10.0,
     ) -> None:
         try:
             self._disconnect_callbacks[peripheral.identifier()] = disconnect_callback

bleak/backends/corebluetooth/PeripheralDelegate.py

@@ -6,6 +6,8 @@
 
 """
 
+from __future__ import annotations
+
 import asyncio
 import itertools
 import logging
@@ -43,7 +45,9 @@ class PeripheralDelegate(NSObject):
 
     ___pyobjc_protocols__ = [CBPeripheralDelegate]
 
-    def initWithPeripheral_(self, peripheral: CBPeripheral):
+    def initWithPeripheral_(
+        self, peripheral: CBPeripheral
+    ) -> Optional[PeripheralDelegate]:
         """macOS init function for NSObject"""
         self = objc.super(PeripheralDelegate, self).init()
 
@@ -287,7 +291,7 @@ def did_discover_characteristics_for_service(
         service: CBService,
         characteristics: NSArray,
         error: Optional[NSError],
-    ):
+    ) -> None:
         future = self._service_characteristic_discovered_futures.get(
             service.startHandle()
         )
@@ -307,7 +311,7 @@ def did_discover_characteristics_for_service(
 
     def peripheral_didDiscoverCharacteristicsForService_error_(
         self, peripheral: CBPeripheral, service: CBService, error: Optional[NSError]
-    ):
+    ) -> None:
         logger.debug("peripheral_didDiscoverCharacteristicsForService_error_")
         self._event_loop.call_soon_threadsafe(
             self.did_discover_characteristics_for_service,
@@ -323,7 +327,7 @@ def did_discover_descriptors_for_characteristic(
         peripheral: CBPeripheral,
         characteristic: CBCharacteristic,
         error: Optional[NSError],
-    ):
+    ) -> None:
         future = self._characteristic_descriptor_discover_futures.get(
             characteristic.handle()
         )
@@ -346,7 +350,7 @@ def peripheral_didDiscoverDescriptorsForCharacteristic_error_(
         peripheral: CBPeripheral,
         characteristic: CBCharacteristic,
         error: Optional[NSError],
-    ):
+    ) -> None:
         logger.debug("peripheral_didDiscoverDescriptorsForCharacteristic_error_")
         self._event_loop.call_soon_threadsafe(
             self.did_discover_descriptors_for_characteristic,
@@ -362,7 +366,7 @@ def did_update_value_for_characteristic(
         characteristic: CBCharacteristic,
         value: NSData,
         error: Optional[NSError],
-    ):
+    ) -> None:
         c_handle = characteristic.handle()
 
         future = self._characteristic_read_futures.get(c_handle)
@@ -389,7 +393,7 @@ def peripheral_didUpdateValueForCharacteristic_error_(
         peripheral: CBPeripheral,
         characteristic: CBCharacteristic,
         error: Optional[NSError],
-    ):
+    ) -> None:
         logger.debug("peripheral_didUpdateValueForCharacteristic_error_")
         self._event_loop.call_soon_threadsafe(
             self.did_update_value_for_characteristic,
@@ -406,7 +410,7 @@ def did_update_value_for_descriptor(
         descriptor: CBDescriptor,
         value: NSObject,
         error: Optional[NSError],
-    ):
+    ) -> None:
         future = self._descriptor_read_futures.get(descriptor.handle())
         if not future:
             logger.warning("Unexpected event didUpdateValueForDescriptor")
@@ -425,7 +429,7 @@ def peripheral_didUpdateValueForDescriptor_error_(
         peripheral: CBPeripheral,
         descriptor: CBDescriptor,
         error: Optional[NSError],
-    ):
+    ) -> None:
         logger.debug("peripheral_didUpdateValueForDescriptor_error_")
         self._event_loop.call_soon_threadsafe(
             self.did_update_value_for_descriptor,
@@ -441,7 +445,7 @@ def did_write_value_for_characteristic(
         peripheral: CBPeripheral,
         characteristic: CBCharacteristic,
         error: Optional[NSError],
-    ):
+    ) -> None:
         future = self._characteristic_write_futures.get(characteristic.handle(), None)
         if not future:
             return  # event only expected on write with response
@@ -459,7 +463,7 @@ def peripheral_didWriteValueForCharacteristic_error_(
         peripheral: CBPeripheral,
         characteristic: CBCharacteristic,
         error: Optional[NSError],
-    ):
+    ) -> None:
         logger.debug("peripheral_didWriteValueForCharacteristic_error_")
         self._event_loop.call_soon_threadsafe(
             self.did_write_value_for_characteristic,
@@ -474,7 +478,7 @@ def did_write_value_for_descriptor(
         peripheral: CBPeripheral,
         descriptor: CBDescriptor,
         error: Optional[NSError],
-    ):
+    ) -> None:
         future = self._descriptor_write_futures.get(descriptor.handle())
         if not future:
             logger.warning("Unexpected event didWriteValueForDescriptor")
@@ -493,7 +497,7 @@ def peripheral_didWriteValueForDescriptor_error_(
         peripheral: CBPeripheral,
         descriptor: CBDescriptor,
         error: Optional[NSError],
-    ):
+    ) -> None:
         logger.debug("peripheral_didWriteValueForDescriptor_error_")
         self._event_loop.call_soon_threadsafe(
             self.did_write_value_for_descriptor,
@@ -508,7 +512,7 @@ def did_update_notification_for_characteristic(
         peripheral: CBPeripheral,
         characteristic: CBCharacteristic,
         error: Optional[NSError],
-    ):
+    ) -> None:
         c_handle = characteristic.handle()
         future = self._characteristic_notify_change_futures.get(c_handle)
         if not future:
@@ -530,7 +534,7 @@ def peripheral_didUpdateNotificationStateForCharacteristic_error_(
         peripheral: CBPeripheral,
         characteristic: CBCharacteristic,
         error: Optional[NSError],
-    ):
+    ) -> None:
         logger.debug("peripheral_didUpdateNotificationStateForCharacteristic_error_")
         self._event_loop.call_soon_threadsafe(
             self.did_update_notification_for_characteristic,

bleak/backends/corebluetooth/client.py

@@ -80,7 +80,7 @@ def __init__(
             else None
         )
 
-    def __str__(self):
+    def __str__(self) -> str:
         return "BleakClientCoreBluetooth ({})".format(self.address)
 
     async def connect(self, **kwargs) -> bool:
@@ -111,7 +111,7 @@ async def connect(self, **kwargs) -> bool:
                 self._peripheral
             )
 
-        def disconnect_callback():
+        def disconnect_callback() -> None:
             # Ensure that `get_services` retrieves services again, rather
             # than using the cached object
             self.services = None
@@ -256,7 +256,7 @@ async def get_services(self, **kwargs) -> BleakGATTServiceCollection:
     async def read_gatt_char(
         self,
         char_specifier: Union[BleakGATTCharacteristic, int, str, uuid.UUID],
-        use_cached=False,
+        use_cached: bool = False,
         **kwargs,
     ) -> bytearray:
         """Perform read operation on the specified GATT characteristic.
@@ -287,7 +287,7 @@ async def read_gatt_char(
         return value
 
     async def read_gatt_descriptor(
-        self, handle: int, use_cached=False, **kwargs
+        self, handle: int, use_cached: bool = False, **kwargs
     ) -> bytearray:
         """Perform read operation on the specified GATT descriptor.
 

bleak/backends/corebluetooth/service.py

@@ -32,7 +32,9 @@ def characteristics(self) -> List[BleakGATTCharacteristicCoreBluetooth]:
         """List of characteristics for this service"""
         return self.__characteristics
 
-    def add_characteristic(self, characteristic: BleakGATTCharacteristicCoreBluetooth):
+    def add_characteristic(
+        self, characteristic: BleakGATTCharacteristicCoreBluetooth
+    ) -> None:
         """Add a :py:class:`~BleakGATTCharacteristicCoreBluetooth` to the service.
 
         Should not be used by end user, but rather by `bleak` itself.