pstruebi/bumble_fork
1
0
Fork 0
forked from auracaster/bumble_mirror
Code Pull Requests Activity

Replace deprecated typing aliases

Browse Source
This commit is contained in:
Josh Wu
2025-06-07 23:29:26 +08:00
parent 3a64772cc5
commit 8a0cd5d0d1
68 changed files with 366 additions and 424 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
bumble/pandora/__init__.py
View File

@@ -45,7 +45,7 @@ __all__ = [
# Add servicers hooks.
_SERVICERS_HOOKS: List[Callable[[PandoraDevice, Config, grpc.aio.Server], None]] = []
_SERVICERS_HOOKS: list[Callable[[PandoraDevice, Config, grpc.aio.Server], None]] = []
def register_servicer_hook(

4
bumble/pandora/config.py
View File

@@ -15,7 +15,7 @@
from __future__ import annotations
from bumble.pairing import PairingConfig, PairingDelegate
from dataclasses import dataclass
from typing import Any, Dict
from typing import Any
@dataclass
@@ -32,7 +32,7 @@ class Config:
PairingDelegate.DEFAULT_KEY_DISTRIBUTION
)
def load_from_dict(self, config: Dict[str, Any]) -> None:
def load_from_dict(self, config: dict[str, Any]) -> None:
io_capability_name: str = config.get(
'io_capability', 'no_output_no_input'
).upper()

12
bumble/pandora/device.py
View File

@@ -32,7 +32,7 @@ from bumble.sdp import (
DataElement,
ServiceAttribute,
)
from typing import Any, Dict, List, Optional
from typing import Any, Optional
# Default rootcanal HCI TCP address
@@ -49,13 +49,13 @@ class PandoraDevice:
# Bumble device instance & configuration.
device: Device
config: Dict[str, Any]
config: dict[str, Any]
# HCI transport name & instance.
_hci_name: str
_hci: Optional[transport.Transport] # type: ignore[name-defined]
def __init__(self, config: Dict[str, Any]) -> None:
def __init__(self, config: dict[str, Any]) -> None:
self.config = config
self.device = _make_device(config)
self._hci_name = config.get(
@@ -95,14 +95,14 @@ class PandoraDevice:
await self.close()
await self.open()
def info(self) -> Optional[Dict[str, str]]:
def info(self) -> Optional[dict[str, str]]:
return {
'public_bd_address': str(self.device.public_address),
'random_address': str(self.device.random_address),
}
def _make_device(config: Dict[str, Any]) -> Device:
def _make_device(config: dict[str, Any]) -> Device:
"""Initialize an idle Bumble device instance."""
# initialize bumble device.
@@ -117,7 +117,7 @@ def _make_device(config: Dict[str, Any]) -> Device:
# TODO(b/267540823): remove when Pandora A2dp is supported
def _make_sdp_records(rfcomm_channel: int) -> Dict[int, List[ServiceAttribute]]:
def _make_sdp_records(rfcomm_channel: int) -> dict[int, list[ServiceAttribute]]:
return {
0x00010001: [
ServiceAttribute(

55
bumble/pandora/host.py
View File

@@ -73,7 +73,6 @@ from pandora.host_pb2 import (
ConnectResponse,
DataTypes,
DisconnectRequest,
DiscoverabilityMode,
InquiryResponse,
PrimaryPhy,
ReadLocalAddressResponse,
@@ -86,9 +85,9 @@ from pandora.host_pb2 import (
WaitConnectionResponse,
WaitDisconnectionRequest,
)
from typing import AsyncGenerator, Dict, List, Optional, Set, Tuple, cast
from typing import AsyncGenerator, Optional, cast
PRIMARY_PHY_MAP: Dict[int, PrimaryPhy] = {
PRIMARY_PHY_MAP: dict[int, PrimaryPhy] = {
# Default value reported by Bumble for legacy Advertising reports.
# FIXME(uael): `None` might be a better value, but Bumble need to change accordingly.
0: PRIMARY_1M,
@@ -96,26 +95,26 @@ PRIMARY_PHY_MAP: Dict[int, PrimaryPhy] = {
3: PRIMARY_CODED,
}
SECONDARY_PHY_MAP: Dict[int, SecondaryPhy] = {
SECONDARY_PHY_MAP: dict[int, SecondaryPhy] = {
0: SECONDARY_NONE,
1: SECONDARY_1M,
2: SECONDARY_2M,
3: SECONDARY_CODED,
}
PRIMARY_PHY_TO_BUMBLE_PHY_MAP: Dict[PrimaryPhy, Phy] = {
PRIMARY_PHY_TO_BUMBLE_PHY_MAP: dict[PrimaryPhy, Phy] = {
PRIMARY_1M: Phy.LE_1M,
PRIMARY_CODED: Phy.LE_CODED,
}
SECONDARY_PHY_TO_BUMBLE_PHY_MAP: Dict[SecondaryPhy, Phy] = {
SECONDARY_PHY_TO_BUMBLE_PHY_MAP: dict[SecondaryPhy, Phy] = {
SECONDARY_NONE: Phy.LE_1M,
SECONDARY_1M: Phy.LE_1M,
SECONDARY_2M: Phy.LE_2M,
SECONDARY_CODED: Phy.LE_CODED,
}
OWN_ADDRESS_MAP: Dict[host_pb2.OwnAddressType, OwnAddressType] = {
OWN_ADDRESS_MAP: dict[host_pb2.OwnAddressType, OwnAddressType] = {
host_pb2.PUBLIC: OwnAddressType.PUBLIC,
host_pb2.RANDOM: OwnAddressType.RANDOM,
host_pb2.RESOLVABLE_OR_PUBLIC: OwnAddressType.RESOLVABLE_OR_PUBLIC,
@@ -124,7 +123,7 @@ OWN_ADDRESS_MAP: Dict[host_pb2.OwnAddressType, OwnAddressType] = {
class HostService(HostServicer):
waited_connections: Set[int]
waited_connections: set[int]
def __init__(
self, grpc_server: grpc.aio.Server, device: Device, config: Config
@@ -618,7 +617,7 @@ class HostService(HostServicer):
self.log.debug('Inquiry')
inquiry_queue: asyncio.Queue[
Optional[Tuple[Address, int, AdvertisingData, int]]
Optional[tuple[Address, int, AdvertisingData, int]]
] = asyncio.Queue()
complete_handler = self.device.on(
self.device.EVENT_INQUIRY_COMPLETE, lambda: inquiry_queue.put_nowait(None)
@@ -670,10 +669,10 @@ class HostService(HostServicer):
return empty_pb2.Empty()
def unpack_data_types(self, dt: DataTypes) -> AdvertisingData:
ad_structures: List[Tuple[int, bytes]] = []
ad_structures: list[tuple[int, bytes]] = []
uuids: List[str]
datas: Dict[str, bytes]
uuids: list[str]
datas: dict[str, bytes]
def uuid128_from_str(uuid: str) -> bytes:
"""Decode a 128-bit uuid encoded as XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX
@@ -887,50 +886,50 @@ class HostService(HostServicer):
def pack_data_types(self, ad: AdvertisingData) -> DataTypes:
dt = DataTypes()
uuids: List[UUID]
uuids: list[UUID]
s: str
i: int
ij: Tuple[int, int]
uuid_data: Tuple[UUID, bytes]
ij: tuple[int, int]
uuid_data: tuple[UUID, bytes]
data: bytes
if uuids := cast(
List[UUID],
list[UUID],
ad.get(AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS),
):
dt.incomplete_service_class_uuids16.extend(
list(map(lambda x: x.to_hex_str('-'), uuids))
)
if uuids := cast(
List[UUID],
list[UUID],
ad.get(AdvertisingData.COMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS),
):
dt.complete_service_class_uuids16.extend(
list(map(lambda x: x.to_hex_str('-'), uuids))
)
if uuids := cast(
List[UUID],
list[UUID],
ad.get(AdvertisingData.INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS),
):
dt.incomplete_service_class_uuids32.extend(
list(map(lambda x: x.to_hex_str('-'), uuids))
)
if uuids := cast(
List[UUID],
list[UUID],
ad.get(AdvertisingData.COMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS),
):
dt.complete_service_class_uuids32.extend(
list(map(lambda x: x.to_hex_str('-'), uuids))
)
if uuids := cast(
List[UUID],
list[UUID],
ad.get(AdvertisingData.INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS),
):
dt.incomplete_service_class_uuids128.extend(
list(map(lambda x: x.to_hex_str('-'), uuids))
)
if uuids := cast(
List[UUID],
list[UUID],
ad.get(AdvertisingData.COMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS),
):
dt.complete_service_class_uuids128.extend(
@@ -945,42 +944,42 @@ class HostService(HostServicer):
if i := cast(int, ad.get(AdvertisingData.CLASS_OF_DEVICE)):
dt.class_of_device = i
if ij := cast(
Tuple[int, int],
tuple[int, int],
ad.get(AdvertisingData.PERIPHERAL_CONNECTION_INTERVAL_RANGE),
):
dt.peripheral_connection_interval_min = ij[0]
dt.peripheral_connection_interval_max = ij[1]
if uuids := cast(
List[UUID],
list[UUID],
ad.get(AdvertisingData.LIST_OF_16_BIT_SERVICE_SOLICITATION_UUIDS),
):
dt.service_solicitation_uuids16.extend(
list(map(lambda x: x.to_hex_str('-'), uuids))
)
if uuids := cast(
List[UUID],
list[UUID],
ad.get(AdvertisingData.LIST_OF_32_BIT_SERVICE_SOLICITATION_UUIDS),
):
dt.service_solicitation_uuids32.extend(
list(map(lambda x: x.to_hex_str('-'), uuids))
)
if uuids := cast(
List[UUID],
list[UUID],
ad.get(AdvertisingData.LIST_OF_128_BIT_SERVICE_SOLICITATION_UUIDS),
):
dt.service_solicitation_uuids128.extend(
list(map(lambda x: x.to_hex_str('-'), uuids))
)
if uuid_data := cast(
Tuple[UUID, bytes], ad.get(AdvertisingData.SERVICE_DATA_16_BIT_UUID)
tuple[UUID, bytes], ad.get(AdvertisingData.SERVICE_DATA_16_BIT_UUID)
):
dt.service_data_uuid16[uuid_data[0].to_hex_str('-')] = uuid_data[1]
if uuid_data := cast(
Tuple[UUID, bytes], ad.get(AdvertisingData.SERVICE_DATA_32_BIT_UUID)
tuple[UUID, bytes], ad.get(AdvertisingData.SERVICE_DATA_32_BIT_UUID)
):
dt.service_data_uuid32[uuid_data[0].to_hex_str('-')] = uuid_data[1]
if uuid_data := cast(
Tuple[UUID, bytes], ad.get(AdvertisingData.SERVICE_DATA_128_BIT_UUID)
tuple[UUID, bytes], ad.get(AdvertisingData.SERVICE_DATA_128_BIT_UUID)
):
dt.service_data_uuid128[uuid_data[0].to_hex_str('-')] = uuid_data[1]
if data := cast(bytes, ad.get(AdvertisingData.PUBLIC_TARGET_ADDRESS, raw=True)):

4
bumble/pandora/l2cap.py
View File

@@ -51,7 +51,7 @@ from pandora.l2cap_pb2 import ( # pytype: disable=pyi-error
WaitDisconnectionRequest,
WaitDisconnectionResponse,
)
from typing import AsyncGenerator, Dict, Optional, Union
from typing import AsyncGenerator, Optional, Union
from dataclasses import dataclass
L2capChannel = Union[ClassicChannel, LeCreditBasedChannel]
@@ -70,7 +70,7 @@ class L2CAPService(L2CAPServicer):
)
self.device = device
self.config = config
self.channels: Dict[bytes, ChannelContext] = {}
self.channels: dict[bytes, ChannelContext] = {}
def register_event(self, l2cap_channel: L2capChannel) -> ChannelContext:
close_future = asyncio.get_running_loop().create_future()

4
bumble/pandora/security.py
View File

@@ -57,7 +57,7 @@ from pandora.security_pb2 import (
WaitSecurityRequest,
WaitSecurityResponse,
)
from typing import Any, AsyncGenerator, AsyncIterator, Callable, Dict, Optional, Union
from typing import Any, AsyncGenerator, AsyncIterator, Callable, Optional, Union
class PairingDelegate(BasePairingDelegate):
@@ -457,7 +457,7 @@ class SecurityService(SecurityServicer):
if self.need_pairing(connection, level):
pair_task = asyncio.create_task(connection.pair())
listeners: Dict[str, Callable[..., Union[None, Awaitable[None]]]] = {
listeners: dict[str, Callable[..., Union[None, Awaitable[None]]]] = {
'disconnection': set_failure('connection_died'),
'pairing_failure': set_failure('pairing_failure'),
'connection_authentication_failure': set_failure('authentication_failure'),

6
bumble/pandora/utils.py
View File

@@ -22,9 +22,9 @@ import logging
from bumble.device import Device
from bumble.hci import Address, AddressType
from google.protobuf.message import Message # pytype: disable=pyi-error
from typing import Any, Dict, Generator, MutableMapping, Optional, Tuple
from typing import Any, Generator, MutableMapping, Optional
ADDRESS_TYPES: Dict[str, AddressType] = {
ADDRESS_TYPES: dict[str, AddressType] = {
"public": Address.PUBLIC_DEVICE_ADDRESS,
"random": Address.RANDOM_DEVICE_ADDRESS,
"public_identity": Address.PUBLIC_IDENTITY_ADDRESS,
@@ -43,7 +43,7 @@ class BumbleServerLoggerAdapter(logging.LoggerAdapter): # type: ignore
def process(
self, msg: str, kwargs: MutableMapping[str, Any]
) -> Tuple[str, MutableMapping[str, Any]]:
) -> tuple[str, MutableMapping[str, Any]]:
assert self.extra
service_name = self.extra['service_name']
assert isinstance(service_name, str)