Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 15 additions & 3 deletions bleak/backends/corebluetooth/CentralManagerDelegate.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -42,6 +42,7 @@
from Foundation import (
NSUUID,
NSArray,
NSData,
NSDictionary,
NSError,
NSKeyValueChangeNewKey,
Expand All @@ -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.
Expand Down Expand Up @@ -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]] = {}
Expand Down Expand Up @@ -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",
Expand Down
30 changes: 17 additions & 13 deletions bleak/backends/corebluetooth/scanner.py
Original file line number Diff line number Diff line change
@@ -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):
Expand All @@ -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,
Expand Down Expand Up @@ -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 = [
Expand All @@ -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
Expand All @@ -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),
)
Expand Down
45 changes: 45 additions & 0 deletions bleak/backends/corebluetooth/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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)
6 changes: 5 additions & 1 deletion typings/Foundation/__init__.pyi
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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")

Expand Down
Loading