auracaster/bumble_mirror
3
0
Fork 1
mirror of https://github.com/google/bumble.git synced 2026-04-16 00:25:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Replace long if-else with match-case

Browse Source
This commit is contained in:
Josh Wu
2026-02-05 15:56:38 +08:00
parent d810d93aaf
commit 3d5648cdc3
11 changed files with 451 additions and 432 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
bumble/a2dp.py
View File

@@ -267,11 +267,12 @@ class MediaCodecInformation:
def create(
cls, media_codec_type: int, data: bytes
) -> MediaCodecInformation | bytes:
if media_codec_type == CodecType.SBC:
match media_codec_type:
case CodecType.SBC:
return SbcMediaCodecInformation.from_bytes(data)
elif media_codec_type == CodecType.MPEG_2_4_AAC:
case CodecType.MPEG_2_4_AAC:
return AacMediaCodecInformation.from_bytes(data)
elif media_codec_type == CodecType.NON_A2DP:
case CodecType.NON_A2DP:
vendor_media_codec_information = (
VendorSpecificMediaCodecInformation.from_bytes(data)
)

22
bumble/at.py
View File

@@ -27,7 +27,7 @@ def tokenize_parameters(buffer: bytes) -> list[bytes]:
are ignored [..], unless they are embedded in numeric or string constants"
Raises AtParsingError in case of invalid input string."""
tokens = []
tokens: list[bytearray] = []
in_quotes = False
token = bytearray()
for b in buffer:
@@ -40,22 +40,23 @@ def tokenize_parameters(buffer: bytes) -> list[bytes]:
tokens.append(token[1:-1])
token = bytearray()
else:
if char == b' ':
match char:
case b' ':
pass
elif char == b',' or char == b')':
case b',' | b')':
tokens.append(token)
tokens.append(char)
token = bytearray()
elif char == b'(':
case b'(':
if len(token) > 0:
raise AtParsingError("open_paren following regular character")
tokens.append(char)
elif char == b'"':
case b'"':
if len(token) > 0:
raise AtParsingError("quote following regular character")
in_quotes = True
token.extend(char)
else:
case _:
token.extend(char)
tokens.append(token)
@@ -71,17 +72,18 @@ def parse_parameters(buffer: bytes) -> list[bytes | list]:
current: bytes | list = b''
for token in tokens:
if token == b',':
match token:
case b',':
accumulator[-1].append(current)
current = b''
elif token == b'(':
case b'(':
accumulator.append([])
elif token == b')':
case b')':
if len(accumulator) < 2:
raise AtParsingError("close_paren without matching open_paren")
accumulator[-1].append(current)
current = accumulator.pop()
else:
case _:
current = token
accumulator[-1].append(current)

21
bumble/att.py
View File

@@ -954,11 +954,12 @@ class Attribute(utils.EventEmitter, Generic[_T]):
self.permissions = permissions
# Convert the type to a UUID object if it isn't already
if isinstance(attribute_type, str):
match attribute_type:
case str():
self.type = UUID(attribute_type)
elif isinstance(attribute_type, bytes):
case bytes():
self.type = UUID.from_bytes(attribute_type)
else:
case _:
self.type = attribute_type
self.value = value
@@ -994,7 +995,8 @@ class Attribute(utils.EventEmitter, Generic[_T]):
)
value: _T | None
if isinstance(self.value, AttributeValue):
match self.value:
case AttributeValue():
try:
read_value = self.value.read(connection)
if inspect.isawaitable(read_value):
@@ -1005,7 +1007,7 @@ class Attribute(utils.EventEmitter, Generic[_T]):
raise ATT_Error(
error_code=error.error_code, att_handle=self.handle
) from error
elif isinstance(self.value, AttributeValueV2):
case AttributeValueV2():
try:
read_value = self.value.read(bearer)
if inspect.isawaitable(read_value):
@@ -1016,7 +1018,7 @@ class Attribute(utils.EventEmitter, Generic[_T]):
raise ATT_Error(
error_code=error.error_code, att_handle=self.handle
) from error
else:
case _:
value = self.value
self.emit(self.EVENT_READ, connection, b'' if value is None else value)
@@ -1049,7 +1051,8 @@ class Attribute(utils.EventEmitter, Generic[_T]):
decoded_value = self.decode_value(value)
if isinstance(self.value, AttributeValue):
match self.value:
case AttributeValue():
try:
result = self.value.write(connection, decoded_value)
if inspect.isawaitable(result):
@@ -1058,7 +1061,7 @@ class Attribute(utils.EventEmitter, Generic[_T]):
raise ATT_Error(
error_code=error.error_code, att_handle=self.handle
) from error
elif isinstance(self.value, AttributeValueV2):
case AttributeValueV2():
try:
result = self.value.write(bearer, decoded_value)
if inspect.isawaitable(result):
@@ -1067,7 +1070,7 @@ class Attribute(utils.EventEmitter, Generic[_T]):
raise ATT_Error(
error_code=error.error_code, att_handle=self.handle
) from error
else:
case _:
self.value = decoded_value
self.emit(self.EVENT_WRITE, connection, decoded_value)

50
bumble/controller.py
View File

@@ -403,13 +403,14 @@ class Controller:
)
# If the packet is a command, invoke the handler for this packet
if isinstance(packet, hci.HCI_Command):
match packet:
case hci.HCI_Command():
self.on_hci_command_packet(packet)
elif isinstance(packet, hci.HCI_AclDataPacket):
case hci.HCI_AclDataPacket():
self.on_hci_acl_data_packet(packet)
elif isinstance(packet, hci.HCI_Event):
case hci.HCI_Event():
self.on_hci_event_packet(packet)
else:
case _:
logger.warning(f'!!! unknown packet type {packet.hci_packet_type}')
def on_hci_command_packet(self, command: hci.HCI_Command) -> None:
@@ -517,25 +518,27 @@ class Controller:
logger.error("Cannot find a connection for %s", sender_address)
return
if isinstance(packet, ll.TerminateInd):
match packet:
case ll.TerminateInd():
self.on_le_disconnected(connection, packet.error_code)
elif isinstance(packet, ll.CisReq):
case ll.CisReq():
self.on_le_cis_request(connection, packet.cig_id, packet.cis_id)
elif isinstance(packet, ll.CisRsp):
case ll.CisRsp():
self.on_le_cis_established(packet.cig_id, packet.cis_id)
connection.send_ll_control_pdu(ll.CisInd(packet.cig_id, packet.cis_id))
elif isinstance(packet, ll.CisInd):
case ll.CisInd():
self.on_le_cis_established(packet.cig_id, packet.cis_id)
elif isinstance(packet, ll.CisTerminateInd):
case ll.CisTerminateInd():
self.on_le_cis_disconnected(packet.cig_id, packet.cis_id)
elif isinstance(packet, ll.EncReq):
case ll.EncReq():
self.on_le_encrypted(connection)
def on_ll_advertising_pdu(self, packet: ll.AdvertisingPdu) -> None:
logger.debug("[%s] <<< Advertising PDU: %s", self.name, packet)
if isinstance(packet, ll.ConnectInd):
match packet:
case ll.ConnectInd():
self.on_le_connect_ind(packet)
elif isinstance(packet, (ll.AdvInd, ll.AdvExtInd)):
case ll.AdvInd() | ll.AdvExtInd():
self.on_advertising_pdu(packet)
def on_le_connect_ind(self, packet: ll.ConnectInd) -> None:
@@ -894,50 +897,51 @@ class Controller:
return future
def on_lmp_packet(self, sender_address: hci.Address, packet: lmp.Packet):
if isinstance(packet, (lmp.LmpAccepted, lmp.LmpAcceptedExt)):
match packet:
case lmp.LmpAccepted() | lmp.LmpAcceptedExt():
if future := self.classic_pending_commands.setdefault(
sender_address, {}
).get(packet.response_opcode):
future.set_result(hci.HCI_SUCCESS)
else:
logger.error("!!! Unhandled packet: %s", packet)
elif isinstance(packet, (lmp.LmpNotAccepted, lmp.LmpNotAcceptedExt)):
case lmp.LmpNotAccepted() | lmp.LmpNotAcceptedExt():
if future := self.classic_pending_commands.setdefault(
sender_address, {}
).get(packet.response_opcode):
future.set_result(packet.error_code)
else:
logger.error("!!! Unhandled packet: %s", packet)
elif isinstance(packet, (lmp.LmpHostConnectionReq)):
case lmp.LmpHostConnectionReq():
self.on_classic_connection_request(
sender_address, hci.HCI_Connection_Complete_Event.LinkType.ACL
)
elif isinstance(packet, (lmp.LmpScoLinkReq)):
case lmp.LmpScoLinkReq():
self.on_classic_connection_request(
sender_address, hci.HCI_Connection_Complete_Event.LinkType.SCO
)
elif isinstance(packet, (lmp.LmpEscoLinkReq)):
case lmp.LmpEscoLinkReq():
self.on_classic_connection_request(
sender_address, hci.HCI_Connection_Complete_Event.LinkType.ESCO
)
elif isinstance(packet, (lmp.LmpDetach)):
case lmp.LmpDetach():
self.on_classic_disconnected(
sender_address, hci.HCI_REMOTE_USER_TERMINATED_CONNECTION_ERROR
)
elif isinstance(packet, (lmp.LmpSwitchReq)):
case lmp.LmpSwitchReq():
self.on_classic_role_change_request(sender_address)
elif isinstance(packet, (lmp.LmpRemoveScoLinkReq, lmp.LmpRemoveEscoLinkReq)):
case lmp.LmpRemoveScoLinkReq() | lmp.LmpRemoveEscoLinkReq():
self.on_classic_sco_disconnected(sender_address, packet.error_code)
elif isinstance(packet, lmp.LmpNameReq):
case lmp.LmpNameReq():
self.on_classic_remote_name_request(sender_address, packet.name_offset)
elif isinstance(packet, lmp.LmpNameRes):
case lmp.LmpNameRes():
self.on_classic_remote_name_response(
sender_address,
packet.name_offset,
packet.name_length,
packet.name_fregment,
)
else:
case _:
logger.error("!!! Unhandled packet: %s", packet)
def on_classic_connection_request(

52
bumble/core.py
View File

@@ -280,13 +280,14 @@ class UUID:
if not force_128:
return self.uuid_bytes
if len(self.uuid_bytes) == 2:
match len(self.uuid_bytes):
case 2:
return self.BASE_UUID + self.uuid_bytes + bytes([0, 0])
elif len(self.uuid_bytes) == 4:
case 4:
return self.BASE_UUID + self.uuid_bytes
elif len(self.uuid_bytes) == 16:
case 16:
return self.uuid_bytes
else:
case _:
assert False, "unreachable"
def to_pdu_bytes(self) -> bytes:
@@ -1769,64 +1770,69 @@ class AdvertisingData:
@classmethod
def ad_data_to_string(cls, ad_type: int, ad_data: bytes) -> str:
if ad_type == AdvertisingData.FLAGS:
match ad_type:
case AdvertisingData.FLAGS:
ad_type_str = 'Flags'
ad_data_str = AdvertisingData.flags_to_string(ad_data[0], short=True)
elif ad_type == AdvertisingData.COMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS:
case AdvertisingData.COMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS:
ad_type_str = 'Complete List of 16-bit Service Class UUIDs'
ad_data_str = AdvertisingData.uuid_list_to_string(ad_data, 2)
elif ad_type == AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS:
case AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS:
ad_type_str = 'Incomplete List of 16-bit Service Class UUIDs'
ad_data_str = AdvertisingData.uuid_list_to_string(ad_data, 2)
elif ad_type == AdvertisingData.COMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS:
case AdvertisingData.COMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS:
ad_type_str = 'Complete List of 32-bit Service Class UUIDs'
ad_data_str = AdvertisingData.uuid_list_to_string(ad_data, 4)
elif ad_type == AdvertisingData.INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS:
case AdvertisingData.INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS:
ad_type_str = 'Incomplete List of 32-bit Service Class UUIDs'
ad_data_str = AdvertisingData.uuid_list_to_string(ad_data, 4)
elif ad_type == AdvertisingData.COMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS:
case AdvertisingData.COMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS:
ad_type_str = 'Complete List of 128-bit Service Class UUIDs'
ad_data_str = AdvertisingData.uuid_list_to_string(ad_data, 16)
elif ad_type == AdvertisingData.INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS:
case AdvertisingData.INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS:
ad_type_str = 'Incomplete List of 128-bit Service Class UUIDs'
ad_data_str = AdvertisingData.uuid_list_to_string(ad_data, 16)
elif ad_type == AdvertisingData.SERVICE_DATA_16_BIT_UUID:
case AdvertisingData.SERVICE_DATA_16_BIT_UUID:
ad_type_str = 'Service Data'
uuid = UUID.from_bytes(ad_data[:2])
ad_data_str = f'service={uuid}, data={ad_data[2:].hex()}'
elif ad_type == AdvertisingData.SERVICE_DATA_32_BIT_UUID:
case AdvertisingData.SERVICE_DATA_32_BIT_UUID:
ad_type_str = 'Service Data'
uuid = UUID.from_bytes(ad_data[:4])
ad_data_str = f'service={uuid}, data={ad_data[4:].hex()}'
elif ad_type == AdvertisingData.SERVICE_DATA_128_BIT_UUID:
case AdvertisingData.SERVICE_DATA_128_BIT_UUID:
ad_type_str = 'Service Data'
uuid = UUID.from_bytes(ad_data[:16])
ad_data_str = f'service={uuid}, data={ad_data[16:].hex()}'
elif ad_type == AdvertisingData.SHORTENED_LOCAL_NAME:
case AdvertisingData.SHORTENED_LOCAL_NAME:
ad_type_str = 'Shortened Local Name'
ad_data_str = f'"{ad_data.decode("utf-8")}"'
elif ad_type == AdvertisingData.COMPLETE_LOCAL_NAME:
case AdvertisingData.COMPLETE_LOCAL_NAME:
ad_type_str = 'Complete Local Name'
try:
ad_data_str = f'"{ad_data.decode("utf-8")}"'
except UnicodeDecodeError:
ad_data_str = ad_data.hex()
elif ad_type == AdvertisingData.TX_POWER_LEVEL:
case AdvertisingData.TX_POWER_LEVEL:
ad_type_str = 'TX Power Level'
ad_data_str = str(ad_data[0])
elif ad_type == AdvertisingData.MANUFACTURER_SPECIFIC_DATA:
case AdvertisingData.MANUFACTURER_SPECIFIC_DATA:
ad_type_str = 'Manufacturer Specific Data'
company_id = struct.unpack_from('<H', ad_data, 0)[0]
company_name = COMPANY_IDENTIFIERS.get(company_id, f'0x{company_id:04X}')
company_name = COMPANY_IDENTIFIERS.get(
company_id, f'0x{company_id:04X}'
)
ad_data_str = f'company={company_name}, data={ad_data[2:].hex()}'
elif ad_type == AdvertisingData.APPEARANCE:
case AdvertisingData.APPEARANCE:
ad_type_str = 'Appearance'
appearance = Appearance.from_int(struct.unpack_from('<H', ad_data, 0)[0])
appearance = Appearance.from_int(
struct.unpack_from('<H', ad_data, 0)[0]
)
ad_data_str = str(appearance)
elif ad_type == AdvertisingData.BROADCAST_NAME:
case AdvertisingData.BROADCAST_NAME:
ad_type_str = 'Broadcast Name'
ad_data_str = ad_data.decode('utf-8')
else:
case _:
ad_type_str = AdvertisingData.Type(ad_type).name
ad_data_str = ad_data.hex()

41
bumble/drivers/intel.py
View File

@@ -201,10 +201,11 @@ def _parse_tlv(data: bytes) -> list[tuple[ValueType, Any]]:
value = data[2 : 2 + value_length]
typed_value: Any
if value_type == ValueType.END:
match value_type:
case ValueType.END:
break
if value_type in (ValueType.CNVI, ValueType.CNVR):
case ValueType.CNVI | ValueType.CNVR:
(v,) = struct.unpack("<I", value)
typed_value = (
(((v >> 0) & 0xF) << 12)
@@ -212,38 +213,38 @@ def _parse_tlv(data: bytes) -> list[tuple[ValueType, Any]]:
| (((v >> 8) & 0xF) << 4)
| (((v >> 24) & 0xF) << 8)
)
elif value_type == ValueType.HARDWARE_INFO:
case ValueType.HARDWARE_INFO:
(v,) = struct.unpack("<I", value)
typed_value = HardwareInfo(
HardwarePlatform((v >> 8) & 0xFF), HardwareVariant((v >> 16) & 0x3F)
)
elif value_type in (
ValueType.USB_VENDOR_ID,
ValueType.USB_PRODUCT_ID,
ValueType.DEVICE_REVISION,
case (
ValueType.USB_VENDOR_ID
| ValueType.USB_PRODUCT_ID
| ValueType.DEVICE_REVISION
):
(typed_value,) = struct.unpack("<H", value)
elif value_type == ValueType.CURRENT_MODE_OF_OPERATION:
case ValueType.CURRENT_MODE_OF_OPERATION:
typed_value = ModeOfOperation(value[0])
elif value_type in (
ValueType.BUILD_TYPE,
ValueType.BUILD_NUMBER,
ValueType.SECURE_BOOT,
ValueType.OTP_LOCK,
ValueType.API_LOCK,
ValueType.DEBUG_LOCK,
ValueType.SECURE_BOOT_ENGINE_TYPE,
case (
ValueType.BUILD_TYPE
| ValueType.BUILD_NUMBER
| ValueType.SECURE_BOOT
| ValueType.OTP_LOCK
| ValueType.API_LOCK
| ValueType.DEBUG_LOCK
| ValueType.SECURE_BOOT_ENGINE_TYPE
):
typed_value = value[0]
elif value_type == ValueType.TIMESTAMP:
case ValueType.TIMESTAMP:
typed_value = Timestamp(value[0], value[1])
elif value_type == ValueType.FIRMWARE_BUILD:
case ValueType.FIRMWARE_BUILD:
typed_value = FirmwareBuild(value[0], Timestamp(value[1], value[2]))
elif value_type == ValueType.BLUETOOTH_ADDRESS:
case ValueType.BLUETOOTH_ADDRESS:
typed_value = hci.Address(
value, address_type=hci.Address.PUBLIC_DEVICE_ADDRESS
)
else:
case _:
typed_value = value
result.append((value_type, typed_value))

83
bumble/hci.py
View File

@@ -31,6 +31,7 @@ from typing import (
ClassVar,
Generic,
Literal,
SupportsBytes,
TypeVar,
cast,
)
@@ -1860,44 +1861,46 @@ class HCI_Object:
field_type = field_type['parser']
# Parse the field
if field_type == '*':
match field_type:
case '*':
# The rest of the bytes
field_value = data[offset:]
return (field_value, len(field_value))
if field_type == 'v':
case 'v':
# Variable-length bytes field, with 1-byte length at the beginning
field_length = data[offset]
offset += 1
field_value = data[offset : offset + field_length]
return (field_value, field_length + 1)
if field_type == 1:
case 1:
# 8-bit unsigned
return (data[offset], 1)
if field_type == -1:
case -1:
# 8-bit signed
return (struct.unpack_from('b', data, offset)[0], 1)
if field_type == 2:
case 2:
# 16-bit unsigned
return (struct.unpack_from('<H', data, offset)[0], 2)
if field_type == '>2':
case '>2':
# 16-bit unsigned big-endian
return (struct.unpack_from('>H', data, offset)[0], 2)
if field_type == -2:
case -2:
# 16-bit signed
return (struct.unpack_from('<h', data, offset)[0], 2)
if field_type == 3:
case 3:
# 24-bit unsigned
padded = data[offset : offset + 3] + bytes([0])
return (struct.unpack('<I', padded)[0], 3)
if field_type == 4:
case 4:
# 32-bit unsigned
return (struct.unpack_from('<I', data, offset)[0], 4)
if field_type == '>4':
case '>4':
# 32-bit unsigned big-endian
return (struct.unpack_from('>I', data, offset)[0], 4)
if isinstance(field_type, int) and 4 < field_type <= 256:
case int() if 4 < field_type <= 256:
# Byte array (from 5 up to 256 bytes)
return (data[offset : offset + field_type], field_type)
if callable(field_type):
new_offset, field_value = field_type(data, offset)
return (field_value, new_offset - offset)
@@ -1954,61 +1957,59 @@ class HCI_Object:
# Serialize the field
if serializer:
field_bytes = serializer(field_value)
elif field_type == 1:
return serializer(field_value)
match field_type:
case 1:
# 8-bit unsigned
field_bytes = bytes([field_value])
elif field_type == -1:
return bytes([field_value])
case -1:
# 8-bit signed
field_bytes = struct.pack('b', field_value)
elif field_type == 2:
return struct.pack('b', field_value)
case 2:
# 16-bit unsigned
field_bytes = struct.pack('<H', field_value)
elif field_type == '>2':
return struct.pack('<H', field_value)
case '>2':
# 16-bit unsigned big-endian
field_bytes = struct.pack('>H', field_value)
elif field_type == -2:
return struct.pack('>H', field_value)
case -2:
# 16-bit signed
field_bytes = struct.pack('<h', field_value)
elif field_type == 3:
return struct.pack('<h', field_value)
case 3:
# 24-bit unsigned
field_bytes = struct.pack('<I', field_value)[0:3]
elif field_type == 4:
return struct.pack('<I', field_value)[0:3]
case 4:
# 32-bit unsigned
field_bytes = struct.pack('<I', field_value)
elif field_type == '>4':
return struct.pack('<I', field_value)
case '>4':
# 32-bit unsigned big-endian
field_bytes = struct.pack('>I', field_value)
elif field_type == '*':
return struct.pack('>I', field_value)
case '*':
if isinstance(field_value, int):
if 0 <= field_value <= 255:
field_bytes = bytes([field_value])
return bytes([field_value])
else:
raise InvalidArgumentError('value too large for *-typed field')
else:
field_bytes = bytes(field_value)
elif field_type == 'v':
return bytes(field_value)
case 'v':
# Variable-length bytes field, with 1-byte length at the beginning
field_bytes = bytes(field_value)
field_length = len(field_bytes)
field_bytes = bytes([field_length]) + field_bytes
elif isinstance(field_value, (bytes, bytearray)) or hasattr(
field_value, '__bytes__'
):
return bytes([field_length]) + field_bytes
if isinstance(field_value, (bytes, bytearray, SupportsBytes)):
field_bytes = bytes(field_value)
if isinstance(field_type, int) and 4 < field_type <= 256:
# Truncate or pad with zeros if the field is too long or too short
if len(field_bytes) < field_type:
field_bytes += bytes(field_type - len(field_bytes))
return field_bytes + bytes(field_type - len(field_bytes))
elif len(field_bytes) > field_type:
field_bytes = field_bytes[:field_type]
else:
return field_bytes[:field_type]
return field_bytes
raise InvalidArgumentError(
f"don't know how to serialize type {type(field_value)}"
)
return field_bytes
@staticmethod
def dict_to_bytes(hci_object, object_fields):
result = bytearray()

25
bumble/host.py
View File

@@ -22,7 +22,7 @@ import collections
import dataclasses
import logging
from collections.abc import Awaitable, Callable
from typing import TYPE_CHECKING, Any, TypeVar, cast, overload
from typing import TYPE_CHECKING, Any, TypeVar, overload
from bumble import drivers, hci, utils
from bumble.colors import color
@@ -1002,17 +1002,18 @@ class Host(utils.EventEmitter):
self.snooper.snoop(bytes(packet), Snooper.Direction.CONTROLLER_TO_HOST)
# If the packet is a command, invoke the handler for this packet
if packet.hci_packet_type == hci.HCI_COMMAND_PACKET:
self.on_hci_command_packet(cast(hci.HCI_Command, packet))
elif packet.hci_packet_type == hci.HCI_EVENT_PACKET:
self.on_hci_event_packet(cast(hci.HCI_Event, packet))
elif packet.hci_packet_type == hci.HCI_ACL_DATA_PACKET:
self.on_hci_acl_data_packet(cast(hci.HCI_AclDataPacket, packet))
elif packet.hci_packet_type == hci.HCI_SYNCHRONOUS_DATA_PACKET:
self.on_hci_sco_data_packet(cast(hci.HCI_SynchronousDataPacket, packet))
elif packet.hci_packet_type == hci.HCI_ISO_DATA_PACKET:
self.on_hci_iso_data_packet(cast(hci.HCI_IsoDataPacket, packet))
else:
match packet:
case hci.HCI_Command():
self.on_hci_command_packet(packet)
case hci.HCI_Event():
self.on_hci_event_packet(packet)
case hci.HCI_AclDataPacket():
self.on_hci_acl_data_packet(packet)
case hci.HCI_SynchronousDataPacket():
self.on_hci_sco_data_packet(packet)
case hci.HCI_IsoDataPacket():
self.on_hci_iso_data_packet(packet)
case _:
logger.warning(f'!!! unknown packet type {packet.hci_packet_type}')
def on_hci_command_packet(self, command: hci.HCI_Command) -> None:

20
bumble/profiles/ascs.py
View File

@@ -664,7 +664,8 @@ class AudioStreamControlService(gatt.TemplateService):
responses = []
logger.debug(f'*** ASCS Write {operation} ***')
if isinstance(operation, ASE_Config_Codec):
match operation:
case ASE_Config_Codec():
for ase_id, *args in zip(
operation.ase_id,
operation.target_latency,
@@ -673,7 +674,7 @@ class AudioStreamControlService(gatt.TemplateService):
operation.codec_specific_configuration,
):
responses.append(self.on_operation(operation.op_code, ase_id, args))
elif isinstance(operation, ASE_Config_QOS):
case ASE_Config_QOS():
for ase_id, *args in zip(
operation.ase_id,
operation.cig_id,
@@ -687,20 +688,17 @@ class AudioStreamControlService(gatt.TemplateService):
operation.presentation_delay,
):
responses.append(self.on_operation(operation.op_code, ase_id, args))
elif isinstance(operation, (ASE_Enable, ASE_Update_Metadata)):
case ASE_Enable() | ASE_Update_Metadata():
for ase_id, *args in zip(
operation.ase_id,
operation.metadata,
):
responses.append(self.on_operation(operation.op_code, ase_id, args))
elif isinstance(
operation,
(
ASE_Receiver_Start_Ready,
ASE_Disable,
ASE_Receiver_Stop_Ready,
ASE_Release,
),
case (
ASE_Receiver_Start_Ready()
| ASE_Disable()
| ASE_Receiver_Stop_Ready()
| ASE_Release()
):
for ase_id in operation.ase_id:
responses.append(self.on_operation(operation.op_code, ase_id, []))

11
bumble/profiles/bap.py
View File

@@ -333,16 +333,17 @@ class CodecSpecificCapabilities:
value = int.from_bytes(data[offset : offset + length - 1], 'little')
offset += length - 1
if type == CodecSpecificCapabilities.Type.SAMPLING_FREQUENCY:
match type:
case CodecSpecificCapabilities.Type.SAMPLING_FREQUENCY:
supported_sampling_frequencies = SupportedSamplingFrequency(value)
elif type == CodecSpecificCapabilities.Type.FRAME_DURATION:
case CodecSpecificCapabilities.Type.FRAME_DURATION:
supported_frame_durations = SupportedFrameDuration(value)
elif type == CodecSpecificCapabilities.Type.AUDIO_CHANNEL_COUNT:
case CodecSpecificCapabilities.Type.AUDIO_CHANNEL_COUNT:
supported_audio_channel_count = bits_to_channel_counts(value)
elif type == CodecSpecificCapabilities.Type.OCTETS_PER_FRAME:
case CodecSpecificCapabilities.Type.OCTETS_PER_FRAME:
min_octets_per_sample = value & 0xFFFF
max_octets_per_sample = value >> 16
elif type == CodecSpecificCapabilities.Type.CODEC_FRAMES_PER_SDU:
case CodecSpecificCapabilities.Type.CODEC_FRAMES_PER_SDU:
supported_max_codec_frames_per_sdu = value
# It is expected here that if some fields are missing, an error should be raised.

9
bumble/profiles/gap.py
View File

@@ -55,13 +55,14 @@ class GenericAccessService(TemplateService):
def __init__(
self, device_name: str, appearance: Appearance | tuple[int, int] | int = 0
):
if isinstance(appearance, int):
match appearance:
case int():
appearance_int = appearance
elif isinstance(appearance, tuple):
case tuple():
appearance_int = (appearance[0] << 6) | appearance[1]
elif isinstance(appearance, Appearance):
case Appearance():
appearance_int = int(appearance)
else:
case _:
raise TypeError()
self.device_name_characteristic = Characteristic(