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

update black formatter to version 24

Browse Source
This commit is contained in:
Gilles Boccon-Gibod
2024-04-01 14:44:46 -07:00
parent ec82242462
commit 58d4ab913a
30 changed files with 155 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
bumble/att.py
View File

@@ -655,7 +655,7 @@ class ATT_Write_Command(ATT_PDU):
@ATT_PDU.subclass(
[
('attribute_handle', HANDLE_FIELD_SPEC),
('attribute_value', '*')
('attribute_value', '*'),
# ('authentication_signature', 'TODO')
]
)

7
bumble/avdtp.py
View File

@@ -1545,9 +1545,10 @@ class Protocol(EventEmitter):
assert False # Should never reach this
async def get_capabilities(
self, seid: int
) -> Union[Get_Capabilities_Response, Get_All_Capabilities_Response,]:
async def get_capabilities(self, seid: int) -> Union[
Get_Capabilities_Response,
Get_All_Capabilities_Response,
]:
if self.version > (1, 2):
return await self.send_command(Get_All_Capabilities_Command(seid))

8
bumble/avrcp.py
View File

@@ -1745,9 +1745,11 @@ class Protocol(pyee.EventEmitter):
avc.CommandFrame.CommandType.CONTROL,
avc.Frame.SubunitType.PANEL,
0,
avc.PassThroughFrame.StateFlag.PRESSED
if pressed
else avc.PassThroughFrame.StateFlag.RELEASED,
(
avc.PassThroughFrame.StateFlag.PRESSED
if pressed
else avc.PassThroughFrame.StateFlag.RELEASED
),
key,
b'',
)

18
bumble/controller.py
View File

@@ -134,15 +134,15 @@ class Controller:
self.hci_sink = None
self.link = link
self.central_connections: Dict[
Address, Connection
] = {} # Connections where this controller is the central
self.peripheral_connections: Dict[
Address, Connection
] = {} # Connections where this controller is the peripheral
self.classic_connections: Dict[
Address, Connection
] = {} # Connections in BR/EDR
self.central_connections: Dict[Address, Connection] = (
{}
) # Connections where this controller is the central
self.peripheral_connections: Dict[Address, Connection] = (
{}
) # Connections where this controller is the peripheral
self.classic_connections: Dict[Address, Connection] = (
{}
) # Connections in BR/EDR
self.central_cis_links: Dict[int, CisLink] = {} # CIS links by handle
self.peripheral_cis_links: Dict[int, CisLink] = {} # CIS links by handle

58
bumble/device.py
View File

@@ -276,12 +276,12 @@ class Advertisement:
data_bytes: bytes = b''
# Constants
TX_POWER_NOT_AVAILABLE: ClassVar[
int
] = HCI_LE_Extended_Advertising_Report_Event.TX_POWER_INFORMATION_NOT_AVAILABLE
RSSI_NOT_AVAILABLE: ClassVar[
int
] = HCI_LE_Extended_Advertising_Report_Event.RSSI_NOT_AVAILABLE
TX_POWER_NOT_AVAILABLE: ClassVar[int] = (
HCI_LE_Extended_Advertising_Report_Event.TX_POWER_INFORMATION_NOT_AVAILABLE
)
RSSI_NOT_AVAILABLE: ClassVar[int] = (
HCI_LE_Extended_Advertising_Report_Event.RSSI_NOT_AVAILABLE
)
def __post_init__(self) -> None:
self.data = AdvertisingData.from_bytes(self.data_bytes)
@@ -558,7 +558,9 @@ class AdvertisingParameters:
)
primary_advertising_interval_min: int = DEVICE_DEFAULT_ADVERTISING_INTERVAL
primary_advertising_interval_max: int = DEVICE_DEFAULT_ADVERTISING_INTERVAL
primary_advertising_channel_map: HCI_LE_Set_Extended_Advertising_Parameters_Command.ChannelMap = (
primary_advertising_channel_map: (
HCI_LE_Set_Extended_Advertising_Parameters_Command.ChannelMap
) = (
AdvertisingChannelMap.CHANNEL_37
| AdvertisingChannelMap.CHANNEL_38
| AdvertisingChannelMap.CHANNEL_39
@@ -1138,14 +1140,12 @@ class Connection(CompositeEventEmitter):
@overload
async def create_l2cap_channel(
self, spec: l2cap.ClassicChannelSpec
) -> l2cap.ClassicChannel:
...
) -> l2cap.ClassicChannel: ...
@overload
async def create_l2cap_channel(
self, spec: l2cap.LeCreditBasedChannelSpec
) -> l2cap.LeCreditBasedChannel:
...
) -> l2cap.LeCreditBasedChannel: ...
async def create_l2cap_channel(
self, spec: Union[l2cap.ClassicChannelSpec, l2cap.LeCreditBasedChannelSpec]
@@ -1723,16 +1723,14 @@ class Device(CompositeEventEmitter):
self,
connection: Connection,
spec: l2cap.ClassicChannelSpec,
) -> l2cap.ClassicChannel:
...
) -> l2cap.ClassicChannel: ...
@overload
async def create_l2cap_channel(
self,
connection: Connection,
spec: l2cap.LeCreditBasedChannelSpec,
) -> l2cap.LeCreditBasedChannel:
...
) -> l2cap.LeCreditBasedChannel: ...
async def create_l2cap_channel(
self,
@@ -1753,16 +1751,14 @@ class Device(CompositeEventEmitter):
self,
spec: l2cap.ClassicChannelSpec,
handler: Optional[Callable[[l2cap.ClassicChannel], Any]] = None,
) -> l2cap.ClassicChannelServer:
...
) -> l2cap.ClassicChannelServer: ...
@overload
def create_l2cap_server(
self,
spec: l2cap.LeCreditBasedChannelSpec,
handler: Optional[Callable[[l2cap.LeCreditBasedChannel], Any]] = None,
) -> l2cap.LeCreditBasedChannelServer:
...
) -> l2cap.LeCreditBasedChannelServer: ...
def create_l2cap_server(
self,
@@ -3289,17 +3285,19 @@ class Device(CompositeEventEmitter):
handler = self.on(
'remote_name',
lambda address, remote_name: pending_name.set_result(remote_name)
if address == peer_address
else None,
lambda address, remote_name: (
pending_name.set_result(remote_name)
if address == peer_address
else None
),
)
failure_handler = self.on(
'remote_name_failure',
lambda address, error_code: pending_name.set_exception(
HCI_Error(error_code)
)
if address == peer_address
else None,
lambda address, error_code: (
pending_name.set_exception(HCI_Error(error_code))
if address == peer_address
else None
),
)
try:
@@ -3475,9 +3473,9 @@ class Device(CompositeEventEmitter):
LE features supported by the remote device.
"""
with closing(EventWatcher()) as watcher:
read_feature_future: asyncio.Future[
LeFeatureMask
] = asyncio.get_running_loop().create_future()
read_feature_future: asyncio.Future[LeFeatureMask] = (
asyncio.get_running_loop().create_future()
)
def on_le_remote_features(handle: int, features: int):
if handle == connection.handle:

1
bumble/gap.py
View File

@@ -36,6 +36,7 @@ logger = logging.getLogger(__name__)
# Classes
# -----------------------------------------------------------------------------
# -----------------------------------------------------------------------------
class GenericAccessService(Service):
def __init__(self, device_name, appearance=(0, 0)):

14
bumble/gatt.py
View File

@@ -342,9 +342,11 @@ class Service(Attribute):
uuid = UUID(uuid)
super().__init__(
GATT_PRIMARY_SERVICE_ATTRIBUTE_TYPE
if primary
else GATT_SECONDARY_SERVICE_ATTRIBUTE_TYPE,
(
GATT_PRIMARY_SERVICE_ATTRIBUTE_TYPE
if primary
else GATT_SECONDARY_SERVICE_ATTRIBUTE_TYPE
),
Attribute.READABLE,
uuid.to_pdu_bytes(),
)
@@ -560,9 +562,9 @@ class CharacteristicAdapter:
def __init__(self, characteristic: Union[Characteristic, AttributeProxy]):
self.wrapped_characteristic = characteristic
self.subscribers: Dict[
Callable, Callable
] = {} # Map from subscriber to proxy subscriber
self.subscribers: Dict[Callable, Callable] = (
{}
) # Map from subscriber to proxy subscriber
if isinstance(characteristic, Characteristic):
self.read_value = self.read_encoded_value

4
bumble/gatt_client.py
View File

@@ -352,9 +352,7 @@ class Client:
if c.uuid == uuid
]
def get_attribute_grouping(
self, attribute_handle: int
) -> Optional[
def get_attribute_grouping(self, attribute_handle: int) -> Optional[
Union[
ServiceProxy,
Tuple[ServiceProxy, CharacteristicProxy],

6
bumble/gatt_server.py
View File

@@ -445,9 +445,9 @@ class Server(EventEmitter):
assert self.pending_confirmations[connection.handle] is None
# Create a future value to hold the eventual response
pending_confirmation = self.pending_confirmations[
connection.handle
] = asyncio.get_running_loop().create_future()
pending_confirmation = self.pending_confirmations[connection.handle] = (
asyncio.get_running_loop().create_future()
)
try:
self.send_gatt_pdu(connection.handle, indication.to_bytes())

20
bumble/hci.py
View File

@@ -4249,9 +4249,11 @@ class HCI_LE_Set_Extended_Scan_Parameters_Command(HCI_Command):
fields.append(
(
f'{scanning_phy_str}.scan_type: ',
'PASSIVE'
if self.scan_types[i] == self.PASSIVE_SCANNING
else 'ACTIVE',
(
'PASSIVE'
if self.scan_types[i] == self.PASSIVE_SCANNING
else 'ACTIVE'
),
)
)
fields.append(
@@ -5010,9 +5012,9 @@ class HCI_LE_Advertising_Report_Event(HCI_LE_Meta_Event):
return f'{color(self.subevent_name(self.subevent_code), "magenta")}:\n{reports}'
HCI_LE_Meta_Event.subevent_classes[
HCI_LE_ADVERTISING_REPORT_EVENT
] = HCI_LE_Advertising_Report_Event
HCI_LE_Meta_Event.subevent_classes[HCI_LE_ADVERTISING_REPORT_EVENT] = (
HCI_LE_Advertising_Report_Event
)
# -----------------------------------------------------------------------------
@@ -5264,9 +5266,9 @@ class HCI_LE_Extended_Advertising_Report_Event(HCI_LE_Meta_Event):
return f'{color(self.subevent_name(self.subevent_code), "magenta")}:\n{reports}'
HCI_LE_Meta_Event.subevent_classes[
HCI_LE_EXTENDED_ADVERTISING_REPORT_EVENT
] = HCI_LE_Extended_Advertising_Report_Event
HCI_LE_Meta_Event.subevent_classes[HCI_LE_EXTENDED_ADVERTISING_REPORT_EVENT] = (
HCI_LE_Extended_Advertising_Report_Event
)
# -----------------------------------------------------------------------------

1
bumble/hid.py
View File

@@ -48,6 +48,7 @@ HID_INTERRUPT_PSM = 0x0013
class Message:
message_type: MessageType
# Report types
class ReportType(enum.IntEnum):
OTHER_REPORT = 0x00

8
bumble/keys.py
View File

@@ -128,10 +128,10 @@ class PairingKeys:
def print(self, prefix=''):
keys_dict = self.to_dict()
for (container_property, value) in keys_dict.items():
for container_property, value in keys_dict.items():
if isinstance(value, dict):
print(f'{prefix}{color(container_property, "cyan")}:')
for (key_property, key_value) in value.items():
for key_property, key_value in value.items():
print(f'{prefix} {color(key_property, "green")}: {key_value}')
else:
print(f'{prefix}{color(container_property, "cyan")}: {value}')
@@ -158,7 +158,7 @@ class KeyStore:
async def get_resolving_keys(self):
all_keys = await self.get_all()
resolving_keys = []
for (name, keys) in all_keys:
for name, keys in all_keys:
if keys.irk is not None:
if keys.address_type is None:
address_type = Address.RANDOM_DEVICE_ADDRESS
@@ -171,7 +171,7 @@ class KeyStore:
async def print(self, prefix=''):
entries = await self.get_all()
separator = ''
for (name, keys) in entries:
for name, keys in entries:
print(separator + prefix + color(name, 'yellow'))
keys.print(prefix=prefix + ' ')
separator = '\n'

40
bumble/pandora/host.py
View File

@@ -287,9 +287,9 @@ class HostService(HostServicer):
self.log.debug(f"WaitDisconnection: {connection_handle}")
if connection := self.device.lookup_connection(connection_handle):
disconnection_future: asyncio.Future[
None
] = asyncio.get_running_loop().create_future()
disconnection_future: asyncio.Future[None] = (
asyncio.get_running_loop().create_future()
)
def on_disconnection(_: None) -> None:
disconnection_future.set_result(None)
@@ -370,9 +370,9 @@ class HostService(HostServicer):
scan_response_data=scan_response_data,
)
pending_connection: asyncio.Future[
bumble.device.Connection
] = asyncio.get_running_loop().create_future()
pending_connection: asyncio.Future[bumble.device.Connection] = (
asyncio.get_running_loop().create_future()
)
if request.connectable:
@@ -516,9 +516,9 @@ class HostService(HostServicer):
await asyncio.sleep(1)
continue
pending_connection: asyncio.Future[
bumble.device.Connection
] = asyncio.get_running_loop().create_future()
pending_connection: asyncio.Future[bumble.device.Connection] = (
asyncio.get_running_loop().create_future()
)
self.log.debug('Wait for LE connection...')
connection = await pending_connection
@@ -563,12 +563,14 @@ class HostService(HostServicer):
legacy=request.legacy,
active=not request.passive,
own_address_type=request.own_address_type,
scan_interval=int(request.interval)
if request.interval
else DEVICE_DEFAULT_SCAN_INTERVAL,
scan_window=int(request.window)
if request.window
else DEVICE_DEFAULT_SCAN_WINDOW,
scan_interval=(
int(request.interval)
if request.interval
else DEVICE_DEFAULT_SCAN_INTERVAL
),
scan_window=(
int(request.window) if request.window else DEVICE_DEFAULT_SCAN_WINDOW
),
scanning_phys=scanning_phys,
)
@@ -782,9 +784,11 @@ class HostService(HostServicer):
*struct.pack('<H', dt.peripheral_connection_interval_min),
*struct.pack(
'<H',
dt.peripheral_connection_interval_max
if dt.peripheral_connection_interval_max
else dt.peripheral_connection_interval_min,
(
dt.peripheral_connection_interval_max
if dt.peripheral_connection_interval_max
else dt.peripheral_connection_interval_min
),
),
]
),

6
bumble/pandora/security.py
View File

@@ -383,9 +383,9 @@ class SecurityService(SecurityServicer):
connection.transport
] == request.level_variant()
wait_for_security: asyncio.Future[
str
] = asyncio.get_running_loop().create_future()
wait_for_security: asyncio.Future[str] = (
asyncio.get_running_loop().create_future()
)
authenticate_task: Optional[asyncio.Future[None]] = None
pair_task: Optional[asyncio.Future[None]] = None

4
bumble/profiles/device_information_service.py
View File

@@ -59,7 +59,7 @@ class DeviceInformationService(TemplateService):
firmware_revision: Optional[str] = None,
software_revision: Optional[str] = None,
system_id: Optional[Tuple[int, int]] = None, # (OUI, Manufacturer ID)
ieee_regulatory_certification_data_list: Optional[bytes] = None
ieee_regulatory_certification_data_list: Optional[bytes] = None,
# TODO: pnp_id
):
characteristics = [
@@ -107,7 +107,7 @@ class DeviceInformationServiceProxy(ProfileServiceProxy):
def __init__(self, service_proxy):
self.service_proxy = service_proxy
for (field, uuid) in (
for field, uuid in (
('manufacturer_name', GATT_MANUFACTURER_NAME_STRING_CHARACTERISTIC),
('model_number', GATT_MODEL_NUMBER_STRING_CHARACTERISTIC),
('serial_number', GATT_SERIAL_NUMBER_STRING_CHARACTERISTIC),

20
bumble/sdp.py
View File

@@ -825,11 +825,13 @@ class Client:
)
attribute_id_list = DataElement.sequence(
[
DataElement.unsigned_integer(
attribute_id[0], value_size=attribute_id[1]
(
DataElement.unsigned_integer(
attribute_id[0], value_size=attribute_id[1]
)
if isinstance(attribute_id, tuple)
else DataElement.unsigned_integer_16(attribute_id)
)
if isinstance(attribute_id, tuple)
else DataElement.unsigned_integer_16(attribute_id)
for attribute_id in attribute_ids
]
)
@@ -881,11 +883,13 @@ class Client:
attribute_id_list = DataElement.sequence(
[
DataElement.unsigned_integer(
attribute_id[0], value_size=attribute_id[1]
(
DataElement.unsigned_integer(
attribute_id[0], value_size=attribute_id[1]
)
if isinstance(attribute_id, tuple)
else DataElement.unsigned_integer_16(attribute_id)
)
if isinstance(attribute_id, tuple)
else DataElement.unsigned_integer_16(attribute_id)
for attribute_id in attribute_ids
]
)

6
bumble/smp.py
View File

@@ -737,9 +737,9 @@ class Session:
# Create a future that can be used to wait for the session to complete
if self.is_initiator:
self.pairing_result: Optional[
asyncio.Future[None]
] = asyncio.get_running_loop().create_future()
self.pairing_result: Optional[asyncio.Future[None]] = (
asyncio.get_running_loop().create_future()
)
else:
self.pairing_result = None

6
bumble/transport/common.py
View File

@@ -59,15 +59,13 @@ class TransportLostError(Exception):
# Typing Protocols
# -----------------------------------------------------------------------------
class TransportSink(Protocol):
def on_packet(self, packet: bytes) -> None:
...
def on_packet(self, packet: bytes) -> None: ...
class TransportSource(Protocol):
terminated: asyncio.Future[None]
def set_packet_sink(self, sink: TransportSink) -> None:
...
def set_packet_sink(self, sink: TransportSink) -> None: ...
# -----------------------------------------------------------------------------

2
bumble/transport/usb.py
View File

@@ -449,7 +449,7 @@ async def open_usb_transport(spec: str) -> Transport:
# Look for the first interface with the right class and endpoints
def find_endpoints(device):
# pylint: disable-next=too-many-nested-blocks
for (configuration_index, configuration) in enumerate(device):
for configuration_index, configuration in enumerate(device):
interface = None
for interface in configuration:
setting = None

18
bumble/utils.py
View File

@@ -117,12 +117,12 @@ class EventWatcher:
self.handlers = []
@overload
def on(self, emitter: EventEmitter, event: str) -> Callable[[_Handler], _Handler]:
...
def on(
self, emitter: EventEmitter, event: str
) -> Callable[[_Handler], _Handler]: ...
@overload
def on(self, emitter: EventEmitter, event: str, handler: _Handler) -> _Handler:
...
def on(self, emitter: EventEmitter, event: str, handler: _Handler) -> _Handler: ...
def on(
self, emitter: EventEmitter, event: str, handler: Optional[_Handler] = None
@@ -144,12 +144,14 @@ class EventWatcher:
return wrapper if handler is None else wrapper(handler)
@overload
def once(self, emitter: EventEmitter, event: str) -> Callable[[_Handler], _Handler]:
...
def once(
self, emitter: EventEmitter, event: str
) -> Callable[[_Handler], _Handler]: ...
@overload
def once(self, emitter: EventEmitter, event: str, handler: _Handler) -> _Handler:
...
def once(
self, emitter: EventEmitter, event: str, handler: _Handler
) -> _Handler: ...
def once(
self, emitter: EventEmitter, event: str, handler: Optional[_Handler] = None