diff --git a/bleak/backends/corebluetooth/CentralManagerDelegate.py b/bleak/backends/corebluetooth/CentralManagerDelegate.py index 44072e28d..bbcc2c356 100644 --- a/bleak/backends/corebluetooth/CentralManagerDelegate.py +++ b/bleak/backends/corebluetooth/CentralManagerDelegate.py @@ -16,7 +16,7 @@ import asyncio import logging from collections.abc import Callable -from typing import Any, Optional +from typing import Any, Optional, TypedDict, cast if sys.version_info < (3, 11): from async_timeout import timeout as async_timeout @@ -42,6 +42,7 @@ from Foundation import ( NSUUID, NSArray, + NSData, NSDictionary, NSError, NSKeyValueChangeNewKey, @@ -65,6 +66,17 @@ DisconnectCallback = Callable[[], None] +class CBAdvertisementData(TypedDict, total=False): + kCBAdvDataLocalName: NSString + kCBAdvDataManufacturerData: NSData + kCBAdvDataServiceData: dict[CBUUID, NSData] + kCBAdvDataServiceUUIDs: NSArray[CBUUID] + kCBAdvertisementDataOverflowServiceUUIDsKey: NSArray[CBUUID] + kCBAdvDataTxPowerLevel: NSNumber + kCBAdvertisementDataIsConnectable: NSNumber + kCBAdvDataOverflowServiceUUIDs: NSArray[CBUUID] + + class ObjcCentralManagerDelegate(NSObject, protocols=[CBCentralManagerDelegate]): """ CoreBluetooth central manager delegate for bridging callbacks to asyncio. @@ -225,7 +237,7 @@ def __init__(self) -> None: self.callbacks: dict[ int, - Callable[[CBPeripheral, NSDictionary[str, Any], NSNumber], None] | None, + Callable[[CBPeripheral, CBAdvertisementData, NSNumber], None] | None, ] = {} self._disconnect_callbacks: dict[NSUUID, DisconnectCallback] = {} self._disconnect_futures: dict[NSUUID, asyncio.Future[None]] = {} @@ -403,7 +415,7 @@ def did_discover_peripheral( for callback in self.callbacks.values(): if callback: - callback(peripheral, advertisementData, RSSI) + callback(peripheral, cast(CBAdvertisementData, advertisementData), RSSI) logger.debug( "Discovered device %s: %s @ RSSI: %d (kCBAdvData %r) and Central: %r", diff --git a/bleak/backends/corebluetooth/scanner.py b/bleak/backends/corebluetooth/scanner.py index 5d3249980..614154ff7 100644 --- a/bleak/backends/corebluetooth/scanner.py +++ b/bleak/backends/corebluetooth/scanner.py @@ -1,12 +1,12 @@ import sys -from typing import TYPE_CHECKING, cast +from typing import TYPE_CHECKING if TYPE_CHECKING: if sys.platform != "darwin": assert False, "This backend is only available on macOS" import logging -from typing import Any, Literal, Optional +from typing import Any, Literal, Optional, cast from warnings import warn if sys.version_info < (3, 12): @@ -16,11 +16,18 @@ import objc from CoreBluetooth import CBPeripheral -from Foundation import NSBundle, NSDictionary, NSNumber +from Foundation import NSBundle, NSNumber from bleak.args.corebluetooth import CBScannerArgs as _CBScannerArgs -from bleak.backends.corebluetooth.CentralManagerDelegate import CentralManagerDelegate -from bleak.backends.corebluetooth.utils import cb_uuid_to_str +from bleak.backends.corebluetooth.CentralManagerDelegate import ( + CBAdvertisementData, + CentralManagerDelegate, +) +from bleak.backends.corebluetooth.utils import ( + cb_uuid_to_str, + to_optional_int, + to_optional_str, +) from bleak.backends.scanner import ( AdvertisementData, AdvertisementDataCallback, @@ -106,7 +113,7 @@ async def start(self) -> None: self.seen_devices = {} def callback( - peripheral: CBPeripheral, adv_data: NSDictionary[str, Any], rssi: NSNumber + peripheral: CBPeripheral, adv_data: CBAdvertisementData, rssi: NSNumber ) -> None: service_uuids = [ @@ -117,9 +124,9 @@ def callback( return # Process service data - service_data_dict_raw = adv_data.get("kCBAdvDataServiceData", {}) service_data = { - cb_uuid_to_str(k): bytes(v) for k, v in service_data_dict_raw.items() + cb_uuid_to_str(k): bytes(v) + for k, v in adv_data.get("kCBAdvDataServiceData", {}).items() } # Process manufacturer data into a more friendly format @@ -132,15 +139,12 @@ def callback( manufacturer_value = bytes(manufacturer_binary_data[2:]) manufacturer_data[manufacturer_id] = manufacturer_value - # set tx_power data if available - tx_power = adv_data.get("kCBAdvDataTxPowerLevel") - advertisement_data = AdvertisementData( - local_name=adv_data.get("kCBAdvDataLocalName"), + local_name=to_optional_str(adv_data.get("kCBAdvDataLocalName")), manufacturer_data=manufacturer_data, service_data=service_data, service_uuids=service_uuids, - tx_power=tx_power, + tx_power=to_optional_int(adv_data.get("kCBAdvDataTxPowerLevel")), rssi=int(rssi), platform_data=(peripheral, adv_data, rssi), ) diff --git a/bleak/backends/corebluetooth/utils.py b/bleak/backends/corebluetooth/utils.py index d462c4aba..4e2bec9eb 100644 --- a/bleak/backends/corebluetooth/utils.py +++ b/bleak/backends/corebluetooth/utils.py @@ -5,7 +5,10 @@ if sys.platform != "darwin": assert False, "This backend is only available on macOS" +from typing import Optional, overload + from CoreBluetooth import CBUUID +from Foundation import NSNumber, NSString from bleak.uuids import normalize_uuid_str @@ -23,3 +26,45 @@ def cb_uuid_to_str(uuid: CBUUID) -> str: The UUID as a lower case Python string (``xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxx``) """ return normalize_uuid_str(uuid.UUIDString()) + + +@overload +def to_optional_str(value: NSString) -> str: ... +@overload +def to_optional_str(value: None) -> None: ... + + +def to_optional_str(value: Optional[NSString]) -> Optional[str]: + """Converts an NSString to a Python string or None. + + Args: + value: The NSString or None. + + Returns: + The Python string or None. + """ + if value is None: + return None + + return str(value) + + +@overload +def to_optional_int(value: NSNumber) -> int: ... +@overload +def to_optional_int(value: None) -> None: ... + + +def to_optional_int(value: Optional[NSNumber]) -> Optional[int]: + """Converts an NSNumber to a Python int or None. + + Args: + value: The NSNumber or None. + + Returns: + The Python int or None. + """ + if value is None: + return None + + return int(value) diff --git a/typings/Foundation/__init__.pyi b/typings/Foundation/__init__.pyi index c5b9e8c67..eee834e7d 100644 --- a/typings/Foundation/__init__.pyi +++ b/typings/Foundation/__init__.pyi @@ -1,6 +1,6 @@ import sys from collections.abc import Iterator, Mapping, Sequence -from typing import Any, NewType, Optional, TypeVar, overload +from typing import Any, NewType, Optional, SupportsIndex, TypeVar, overload if sys.version_info < (3, 12): from typing_extensions import Buffer @@ -49,6 +49,10 @@ class NSData(NSObject, Buffer): def initWithBytes_length_(self, bytes: Buffer, length: int) -> Self: ... def length(self) -> int: ... def getBytes_length_(self, buffer: bytes, length: int) -> None: ... + @overload + def __getitem__(self, index: SupportsIndex) -> int: ... + @overload + def __getitem__(self, index: slice) -> memoryview: ... T = TypeVar("T")