Compare commits

...

3 Commits

Author SHA1 Message Date
Gilles Boccon-Gibod
b2893f26b6 fix types 2026-03-06 18:23:20 -08:00
Gilles Boccon-Gibod
90560cdea1 revert libusb-package version change 2026-03-06 17:54:03 -08:00
Gilles Boccon-Gibod
794a4a3ef0 add basic support for SCO packets over USB 2026-03-03 10:28:56 -08:00
11 changed files with 1189 additions and 540 deletions

View File

@@ -45,8 +45,10 @@ from bumble.hci import (
HCI_Read_Local_Supported_Codecs_Command,
HCI_Read_Local_Supported_Codecs_V2_Command,
HCI_Read_Local_Version_Information_Command,
HCI_Read_Voice_Setting_Command,
LeFeature,
SpecificationVersion,
VoiceSetting,
map_null_terminated_utf8_string,
)
from bumble.host import Host
@@ -214,6 +216,16 @@ async def get_codecs_info(host: Host) -> None:
if not response2.vendor_specific_codec_ids:
print(' No Vendor-specific codecs')
if host.supports_command(HCI_Read_Voice_Setting_Command.op_code):
response3 = await host.send_sync_command(HCI_Read_Voice_Setting_Command())
voice_setting = VoiceSetting.from_int(response3.voice_setting)
print(color('Voice Setting:', 'yellow'))
print(f' Air Coding Format: {voice_setting.air_coding_format.name}')
print(f' Linear PCM Bit Position: {voice_setting.linear_pcm_bit_position}')
print(f' Input Sample Size: {voice_setting.input_sample_size.name}')
print(f' Input Data Format: {voice_setting.input_data_format.name}')
print(f' Input Coding Format: {voice_setting.input_coding_format.name}')
# -----------------------------------------------------------------------------
async def async_main(

View File

@@ -16,6 +16,8 @@
# Imports
# -----------------------------------------------------------------------------
import asyncio
import statistics
import struct
import time
import click
@@ -25,7 +27,9 @@ from bumble.colors import color
from bumble.hci import (
HCI_READ_LOOPBACK_MODE_COMMAND,
HCI_WRITE_LOOPBACK_MODE_COMMAND,
Address,
HCI_Read_Loopback_Mode_Command,
HCI_SynchronousDataPacket,
HCI_Write_Loopback_Mode_Command,
LoopbackMode,
)
@@ -36,34 +40,59 @@ from bumble.transport import open_transport
class Loopback:
"""Send and receive ACL data packets in local loopback mode"""
def __init__(self, packet_size: int, packet_count: int, transport: str):
def __init__(
self,
packet_size: int,
packet_count: int,
connection_type: str,
mode: str,
interval: int,
transport: str,
):
self.transport = transport
self.packet_size = packet_size
self.packet_count = packet_count
self.connection_handle: int | None = None
self.connection_type = connection_type
self.connection_event = asyncio.Event()
self.mode = mode
self.interval = interval
self.done = asyncio.Event()
self.expected_cid = 0
self.expected_counter = 0
self.bytes_received = 0
self.start_timestamp = 0.0
self.last_timestamp = 0.0
self.send_timestamps: list[float] = []
self.rtts: list[float] = []
def on_connection(self, connection_handle: int, *args):
"""Retrieve connection handle from new connection event"""
if not self.connection_event.is_set():
# save first connection handle for ACL
# subsequent connections are SCO
# The first connection handle is of type ACL,
# subsequent connections are of type SCO
if self.connection_type == "sco" and self.connection_handle is None:
self.connection_handle = connection_handle
return
self.connection_handle = connection_handle
self.connection_event.set()
def on_sco_connection(
self, address: Address, connection_handle: int, link_type: int
):
self.on_connection(connection_handle)
def on_l2cap_pdu(self, connection_handle: int, cid: int, pdu: bytes):
"""Calculate packet receive speed"""
now = time.time()
print(f'<<< Received packet {cid}: {len(pdu)} bytes')
(counter,) = struct.unpack_from("H", pdu, 0)
rtt = now - self.send_timestamps[counter]
self.rtts.append(rtt)
print(f'<<< Received packet {counter}: {len(pdu)} bytes, RTT={rtt:.4f}')
assert connection_handle == self.connection_handle
assert cid == self.expected_cid
self.expected_cid += 1
if cid == 0:
assert counter == self.expected_counter
self.expected_counter += 1
if counter == 0:
self.start_timestamp = now
else:
elapsed_since_start = now - self.start_timestamp
@@ -71,20 +100,52 @@ class Loopback:
self.bytes_received += len(pdu)
instant_rx_speed = len(pdu) / elapsed_since_last
average_rx_speed = self.bytes_received / elapsed_since_start
print(
color(
f'@@@ RX speed: instant={instant_rx_speed:.4f},'
f' average={average_rx_speed:.4f}',
'cyan',
if self.mode == 'throughput':
print(
color(
f'@@@ RX speed: instant={instant_rx_speed:.4f},'
f' average={average_rx_speed:.4f},',
'cyan',
)
)
)
self.last_timestamp = now
if self.expected_cid == self.packet_count:
if self.expected_counter == self.packet_count:
print(color('@@@ Received last packet', 'green'))
self.done.set()
def on_sco_packet(self, connection_handle: int, packet) -> None:
print("---", connection_handle, packet)
async def send_acl_packet(self, host: Host, packet: bytes) -> None:
assert self.connection_handle
host.send_l2cap_pdu(self.connection_handle, 0, packet)
async def send_sco_packet(self, host: Host, packet: bytes) -> None:
assert self.connection_handle
host.send_hci_packet(
HCI_SynchronousDataPacket(
connection_handle=self.connection_handle,
packet_status=HCI_SynchronousDataPacket.Status.CORRECTLY_RECEIVED_DATA,
data_total_length=len(packet),
data=packet,
)
)
async def send_loop(self, host: Host, sender) -> None:
for counter in range(0, self.packet_count):
print(
color(
f'>>> Sending {self.connection_type.upper()} '
f'packet {counter}: {self.packet_size} bytes',
'yellow',
)
)
self.send_timestamps.append(time.time())
await sender(host, struct.pack("H", counter) + bytes(self.packet_size - 2))
await asyncio.sleep(self.interval / 1000 if self.mode == "rtt" else 0)
async def run(self) -> None:
"""Run a loopback throughput test"""
print(color('>>> Connecting to HCI...', 'green'))
@@ -126,8 +187,11 @@ class Loopback:
return
# set event callbacks
host.on('connection', self.on_connection)
host.on('classic_connection', self.on_connection)
host.on('le_connection', self.on_connection)
host.on('sco_connection', self.on_sco_connection)
host.on('l2cap_pdu', self.on_l2cap_pdu)
host.on('sco_packet', self.on_sco_packet)
loopback_mode = LoopbackMode.LOCAL
@@ -148,32 +212,37 @@ class Loopback:
print(color('=== Start sending', 'magenta'))
start_time = time.time()
bytes_sent = 0
for cid in range(0, self.packet_count):
# using the cid as an incremental index
host.send_l2cap_pdu(
self.connection_handle, cid, bytes(self.packet_size)
)
print(
color(
f'>>> Sending packet {cid}: {self.packet_size} bytes', 'yellow'
)
)
bytes_sent += self.packet_size # don't count L2CAP or HCI header sizes
await asyncio.sleep(0) # yield to allow packet receive
if self.connection_type == "acl":
sender = self.send_acl_packet
elif self.connection_type == "sco":
sender = self.send_sco_packet
else:
raise ValueError(f'Unknown connection type: {self.connection_type}')
await self.send_loop(host, sender)
await self.done.wait()
print(color('=== Done!', 'magenta'))
bytes_sent = self.packet_size * self.packet_count
elapsed = time.time() - start_time
average_tx_speed = bytes_sent / elapsed
print(
color(
f'@@@ TX speed: average={average_tx_speed:.4f} ({bytes_sent} bytes'
f' in {elapsed:.2f} seconds)',
'green',
if self.mode == 'throughput':
print(
color(
f'@@@ TX speed: average={average_tx_speed:.4f} '
f'({bytes_sent} bytes in {elapsed:.2f} seconds)',
'green',
)
)
if self.mode == 'rtt':
print(
color(
f'RTTs: min={min(self.rtts):.4f}, '
f'max={max(self.rtts):.4f}, '
f'avg={statistics.mean(self.rtts):.4f}',
'blue',
)
)
)
# -----------------------------------------------------------------------------
@@ -194,11 +263,43 @@ class Loopback:
default=10,
help='Packet count',
)
@click.option(
'--connection-type',
'-t',
metavar='TYPE',
type=click.Choice(['acl', 'sco']),
default='acl',
help='Connection type',
)
@click.option(
'--mode',
'-m',
metavar='MODE',
type=click.Choice(['throughput', 'rtt']),
default='throughput',
help='Test mode',
)
@click.option(
'--interval',
type=int,
default=100,
help='Inter-packet interval (ms) [RTT mode only]',
)
@click.argument('transport')
def main(packet_size, packet_count, transport):
def main(packet_size, packet_count, connection_type, mode, interval, transport):
bumble.logging.setup_basic_logging()
loopback = Loopback(packet_size, packet_count, transport)
asyncio.run(loopback.run())
if connection_type == "sco" and packet_size > 255:
print("ERROR: the maximum packet size for SCO is 255")
return
async def run():
loopback = Loopback(
packet_size, packet_count, connection_type, mode, interval, transport
)
await loopback.run()
asyncio.run(run())
# -----------------------------------------------------------------------------

View File

@@ -111,9 +111,14 @@ def show_device_details(device):
if (endpoint.getAddress() & USB_ENDPOINT_IN == 0)
else 'IN'
)
endpoint_details = (
f', Max Packet Size = {endpoint.getMaxPacketSize()}'
if endpoint_type == 'ISOCHRONOUS'
else ''
)
print(
f' Endpoint 0x{endpoint.getAddress():02X}: '
f'{endpoint_type} {endpoint_direction}'
f'{endpoint_type} {endpoint_direction}{endpoint_details}'
)

View File

@@ -1423,6 +1423,9 @@ class ScoLink(utils.CompositeEventEmitter):
acl_connection: Connection
handle: int
link_type: int
rx_packet_length: int
tx_packet_length: int
air_mode: hci.CodecID
sink: Callable[[hci.HCI_SynchronousDataPacket], Any] | None = None
EVENT_DISCONNECTION: ClassVar[str] = "disconnection"
@@ -5968,7 +5971,7 @@ class Device(utils.CompositeEventEmitter):
def on_connection_request(
self, bd_addr: hci.Address, class_of_device: int, link_type: int
):
logger.debug(f'*** Connection request: {bd_addr}')
logger.debug(f'*** Connection request: {bd_addr} link_type={link_type}')
# Handle SCO request.
if link_type in (
@@ -5978,6 +5981,7 @@ class Device(utils.CompositeEventEmitter):
if connection := self.find_connection_by_bd_addr(
bd_addr, transport=PhysicalTransport.BR_EDR
):
connection.emit(self.EVENT_SCO_REQUEST, link_type)
self.emit(self.EVENT_SCO_REQUEST, connection, link_type)
else:
logger.error(f'SCO request from a non-connected device {bd_addr}')
@@ -6337,8 +6341,7 @@ class Device(utils.CompositeEventEmitter):
logger.warning('peer name is not valid UTF-8')
if connection:
connection.emit(connection.EVENT_REMOTE_NAME_FAILURE, error)
else:
self.emit(self.EVENT_REMOTE_NAME_FAILURE, address, error)
self.emit(self.EVENT_REMOTE_NAME_FAILURE, address, error)
# [Classic only]
@host_event_handler
@@ -6355,7 +6358,13 @@ class Device(utils.CompositeEventEmitter):
@with_connection_from_address
@utils.experimental('Only for testing.')
def on_sco_connection(
self, acl_connection: Connection, sco_handle: int, link_type: int
self,
acl_connection: Connection,
sco_handle: int,
link_type: int,
rx_packet_length: int,
tx_packet_length: int,
air_mode: int,
) -> None:
logger.debug(
f'*** SCO connected: {acl_connection.peer_address}, '
@@ -6367,7 +6376,11 @@ class Device(utils.CompositeEventEmitter):
acl_connection=acl_connection,
handle=sco_handle,
link_type=link_type,
rx_packet_length=rx_packet_length,
tx_packet_length=tx_packet_length,
air_mode=hci.CodecID(air_mode),
)
acl_connection.emit(self.EVENT_SCO_CONNECTION, sco_link)
self.emit(self.EVENT_SCO_CONNECTION, sco_link)
# [Classic only]
@@ -6378,7 +6391,8 @@ class Device(utils.CompositeEventEmitter):
self, acl_connection: Connection, status: int
) -> None:
logger.debug(f'*** SCO connection failure: {acl_connection.peer_address}***')
self.emit(self.EVENT_SCO_CONNECTION_FAILURE)
acl_connection.emit(self.EVENT_SCO_CONNECTION_FAILURE, status)
self.emit(self.EVENT_SCO_CONNECTION_FAILURE, status)
# [Classic only]
@host_event_handler
@@ -6841,15 +6855,18 @@ class Device(utils.CompositeEventEmitter):
@with_connection_from_address
def on_classic_pairing(self, connection: Connection) -> None:
connection.emit(connection.EVENT_CLASSIC_PAIRING)
self.emit(connection.EVENT_CLASSIC_PAIRING, connection)
# [Classic only]
@host_event_handler
@with_connection_from_address
def on_classic_pairing_failure(self, connection: Connection, status: int) -> None:
connection.emit(connection.EVENT_CLASSIC_PAIRING_FAILURE, status)
self.emit(connection.EVENT_CLASSIC_PAIRING_FAILURE, connection, status)
def on_pairing_start(self, connection: Connection) -> None:
connection.emit(connection.EVENT_PAIRING_START)
self.emit(connection.EVENT_PAIRING_START, connection)
def on_pairing(
self,

View File

@@ -1769,6 +1769,61 @@ class CodingFormat:
)
@dataclasses.dataclass(frozen=True)
class VoiceSetting:
class AirCodingFormat(enum.IntEnum):
CVSD = 0
U_LAW = 1
A_LAW = 2
TRANSPARENT_DATA = 3
class InputSampleSize(enum.IntEnum):
SIZE_8_BITS = 0
SIZE_16_BITS = 1
class InputDataFormat(enum.IntEnum):
ONES_COMPLEMENT = 0
TWOS_COMPLEMENT = 1
SIGN_AND_MAGNITUDE = 2
UNSIGNED = 3
class InputCodingFormat(enum.IntEnum):
LINEAR = 0
U_LAW = 1
A_LAW = 2
RESERVED = 3
air_coding_format: AirCodingFormat = AirCodingFormat.CVSD
linear_pcm_bit_position: int = 0
input_sample_size: InputSampleSize = InputSampleSize.SIZE_8_BITS
input_data_format: InputDataFormat = InputDataFormat.ONES_COMPLEMENT
input_coding_format: InputCodingFormat = InputCodingFormat.LINEAR
@classmethod
def from_int(cls, value: int) -> VoiceSetting:
air_coding_format = cls.AirCodingFormat(value & 0b11)
linear_pcm_bit_position = (value >> 2) & 0b111
input_sample_size = cls.InputSampleSize((value >> 5) & 0b1)
input_data_format = cls.InputDataFormat((value >> 6) & 0b11)
input_coding_format = cls.InputCodingFormat((value >> 8) & 0b11)
return cls(
air_coding_format=air_coding_format,
linear_pcm_bit_position=linear_pcm_bit_position,
input_sample_size=input_sample_size,
input_data_format=input_data_format,
input_coding_format=input_coding_format,
)
def __int__(self) -> int:
return (
self.air_coding_format
| (self.linear_pcm_bit_position << 2)
| (self.input_sample_size << 5)
| (self.input_data_format << 6)
| (self.input_coding_format << 8)
)
# -----------------------------------------------------------------------------
class HCI_Constant:
@staticmethod
@@ -2907,6 +2962,23 @@ class HCI_Read_Clock_Offset_Command(HCI_AsyncCommand):
connection_handle: int = field(metadata=metadata(2))
# -----------------------------------------------------------------------------
@HCI_Command.command
@dataclasses.dataclass
class HCI_Accept_Synchronous_Connection_Request_Command(HCI_AsyncCommand):
'''
See Bluetooth spec @ 7.1.27 Accept Synchronous Connection Request Command
'''
bd_addr: Address = field(metadata=metadata(Address.parse_address))
transmit_bandwidth: int = field(metadata=metadata(4))
receive_bandwidth: int = field(metadata=metadata(4))
max_latency: int = field(metadata=metadata(2))
voice_setting: int = field(metadata=metadata(2))
retransmission_effort: int = field(metadata=metadata(1))
packet_type: int = field(metadata=metadata(2))
# -----------------------------------------------------------------------------
@HCI_Command.command
@dataclasses.dataclass
@@ -3965,6 +4037,23 @@ class HCI_Read_Local_OOB_Extended_Data_Command(
'''
# -----------------------------------------------------------------------------
@HCI_SyncCommand.sync_command(HCI_StatusReturnParameters)
@dataclasses.dataclass
class HCI_Configure_Data_Path_Command(HCI_SyncCommand[HCI_StatusReturnParameters]):
'''
See Bluetooth spec @ 7.3.101 Configure Data Path Command
'''
class DataPathDirection(SpecableEnum):
INPUT = 0x00
OUTPUT = 0x01
data_path_direction: DataPathDirection = field(metadata=metadata(1))
data_path_id: int = field(metadata=metadata(1))
vendor_specific_config: bytes = field(metadata=metadata('*'))
# -----------------------------------------------------------------------------
@dataclasses.dataclass
class HCI_Read_Local_Version_Information_ReturnParameters(HCI_StatusReturnParameters):
@@ -7355,7 +7444,7 @@ class HCI_Connection_Complete_Event(HCI_Event):
status: int = field(metadata=metadata(STATUS_SPEC))
connection_handle: int = field(metadata=metadata(2))
bd_addr: Address = field(metadata=metadata(Address.parse_address))
link_type: int = field(metadata=LinkType.type_metadata(1))
link_type: LinkType = field(metadata=LinkType.type_metadata(1))
encryption_enabled: int = field(metadata=metadata(1))
@@ -7751,12 +7840,6 @@ class HCI_Synchronous_Connection_Complete_Event(HCI_Event):
SCO = 0x00
ESCO = 0x02
class AirMode(SpecableEnum):
U_LAW_LOG = 0x00
A_LAW_LOG_AIR_MORE = 0x01
CVSD = 0x02
TRANSPARENT_DATA = 0x03
status: int = field(metadata=metadata(STATUS_SPEC))
connection_handle: int = field(metadata=metadata(2))
bd_addr: Address = field(metadata=metadata(Address.parse_address))
@@ -7765,7 +7848,7 @@ class HCI_Synchronous_Connection_Complete_Event(HCI_Event):
retransmission_window: int = field(metadata=metadata(1))
rx_packet_length: int = field(metadata=metadata(2))
tx_packet_length: int = field(metadata=metadata(2))
air_mode: int = field(metadata=AirMode.type_metadata(1))
air_mode: int = field(metadata=CodecID.type_metadata(1))
# -----------------------------------------------------------------------------
@@ -7997,7 +8080,9 @@ class HCI_AclDataPacket(HCI_Packet):
bc_flag = (h >> 14) & 3
data = packet[5:]
if len(data) != data_total_length:
raise InvalidPacketError('invalid packet length')
raise InvalidPacketError(
f'invalid packet length {len(data)} != {data_total_length}'
)
return cls(
connection_handle=connection_handle,
pb_flag=pb_flag,
@@ -8030,10 +8115,16 @@ class HCI_SynchronousDataPacket(HCI_Packet):
See Bluetooth spec @ 5.4.3 HCI SCO Data Packets
'''
class Status(enum.IntEnum):
CORRECTLY_RECEIVED_DATA = 0b00
POSSIBLY_INVALID_DATA = 0b01
NO_DATA = 0b10
DATA_PARTIALLY_LOST = 0b11
hci_packet_type = HCI_SYNCHRONOUS_DATA_PACKET
connection_handle: int
packet_status: int
packet_status: Status
data_total_length: int
data: bytes
@@ -8042,7 +8133,7 @@ class HCI_SynchronousDataPacket(HCI_Packet):
# Read the header
h, data_total_length = struct.unpack_from('<HB', packet, 1)
connection_handle = h & 0xFFF
packet_status = (h >> 12) & 0b11
packet_status = cls.Status((h >> 12) & 0b11)
data = packet[4:]
if len(data) != data_total_length:
raise InvalidPacketError(
@@ -8066,7 +8157,7 @@ class HCI_SynchronousDataPacket(HCI_Packet):
return (
f'{color("SCO", "blue")}: '
f'handle=0x{self.connection_handle:04x}, '
f'ps={self.packet_status}, '
f'ps={self.packet_status.name}, '
f'data_total_length={self.data_total_length}, '
f'data={self.data.hex()}'
)
@@ -8094,8 +8185,8 @@ class HCI_IsoDataPacket(HCI_Packet):
def __post_init__(self) -> None:
self.ts_flag = self.time_stamp is not None
@staticmethod
def from_bytes(packet: bytes) -> HCI_IsoDataPacket:
@classmethod
def from_bytes(cls, packet: bytes) -> HCI_IsoDataPacket:
time_stamp: int | None = None
packet_sequence_number: int | None = None
iso_sdu_length: int | None = None
@@ -8124,7 +8215,7 @@ class HCI_IsoDataPacket(HCI_Packet):
pos += 4
iso_sdu_fragment = packet[pos:]
return HCI_IsoDataPacket(
return cls(
connection_handle=connection_handle,
pb_flag=pb_flag,
ts_flag=ts_flag,

View File

@@ -166,7 +166,7 @@ class AgFeature(enum.IntFlag):
VOICE_RECOGNITION_TEXT = 0x2000
class AudioCodec(enum.IntEnum):
class AudioCodec(utils.OpenIntEnum):
"""
Audio Codec IDs (normative).
@@ -178,7 +178,7 @@ class AudioCodec(enum.IntEnum):
LC3_SWB = 0x03 # Support for LC3-SWB audio codec
class HfIndicator(enum.IntEnum):
class HfIndicator(utils.OpenIntEnum):
"""
HF Indicators (normative).
@@ -207,7 +207,7 @@ class CallHoldOperation(enum.Enum):
)
class ResponseHoldStatus(enum.IntEnum):
class ResponseHoldStatus(utils.OpenIntEnum):
"""
Response Hold status (normative).
@@ -235,7 +235,7 @@ class AgIndicator(enum.Enum):
BATTERY_CHARGE = 'battchg'
class CallSetupAgIndicator(enum.IntEnum):
class CallSetupAgIndicator(utils.OpenIntEnum):
"""
Values for the Call Setup AG indicator (normative).
@@ -248,7 +248,7 @@ class CallSetupAgIndicator(enum.IntEnum):
REMOTE_ALERTED = 3 # Remote party alerted in an outgoing call
class CallHeldAgIndicator(enum.IntEnum):
class CallHeldAgIndicator(utils.OpenIntEnum):
"""
Values for the Call Held AG indicator (normative).
@@ -262,7 +262,7 @@ class CallHeldAgIndicator(enum.IntEnum):
CALL_ON_HOLD_NO_ACTIVE_CALL = 2 # Call on hold, no active call
class CallInfoDirection(enum.IntEnum):
class CallInfoDirection(utils.OpenIntEnum):
"""
Call Info direction (normative).
@@ -273,7 +273,7 @@ class CallInfoDirection(enum.IntEnum):
MOBILE_TERMINATED_CALL = 1
class CallInfoStatus(enum.IntEnum):
class CallInfoStatus(utils.OpenIntEnum):
"""
Call Info status (normative).
@@ -288,7 +288,7 @@ class CallInfoStatus(enum.IntEnum):
WAITING = 5
class CallInfoMode(enum.IntEnum):
class CallInfoMode(utils.OpenIntEnum):
"""
Call Info mode (normative).
@@ -301,7 +301,7 @@ class CallInfoMode(enum.IntEnum):
UNKNOWN = 9
class CallInfoMultiParty(enum.IntEnum):
class CallInfoMultiParty(utils.OpenIntEnum):
"""
Call Info Multi-Party state (normative).
@@ -388,7 +388,7 @@ class CallLineIdentification:
)
class VoiceRecognitionState(enum.IntEnum):
class VoiceRecognitionState(utils.OpenIntEnum):
"""
vrec values provided in AT+BVRA command.
@@ -401,7 +401,7 @@ class VoiceRecognitionState(enum.IntEnum):
ENHANCED_READY = 2
class CmeError(enum.IntEnum):
class CmeError(utils.OpenIntEnum):
"""
CME ERROR codes (partial listed).
@@ -1624,7 +1624,7 @@ class AgProtocol(utils.EventEmitter):
# -----------------------------------------------------------------------------
class ProfileVersion(enum.IntEnum):
class ProfileVersion(utils.OpenIntEnum):
"""
Profile version (normative).
@@ -2076,6 +2076,7 @@ _ESCO_PARAMETERS_MSBC_T1 = EscoParameters(
max_latency=0x0008,
packet_type=(
HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.EV3
| HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_2_EV3
| HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_3_EV3
| HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_2_EV5
| HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_3_EV5
@@ -2091,7 +2092,6 @@ _ESCO_PARAMETERS_MSBC_T2 = EscoParameters(
max_latency=0x000D,
packet_type=(
HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.EV3
| HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_2_EV3
| HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_3_EV3
| HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_2_EV5
| HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_3_EV5

View File

@@ -686,6 +686,8 @@ class Host(utils.EventEmitter):
self.pending_response, timeout=response_timeout
)
return response
except asyncio.TimeoutError:
raise
except Exception:
logger.exception(color("!!! Exception while sending command:", "red"))
raise
@@ -866,7 +868,7 @@ class Host(utils.EventEmitter):
self.send_hci_packet(
hci.HCI_SynchronousDataPacket(
connection_handle=connection_handle,
packet_status=0,
packet_status=hci.HCI_SynchronousDataPacket.Status.CORRECTLY_RECEIVED_DATA,
data_total_length=len(sdu),
data=sdu,
)
@@ -1177,11 +1179,28 @@ class Host(utils.EventEmitter):
def on_hci_connection_complete_event(
self, event: hci.HCI_Connection_Complete_Event
):
if event.link_type == hci.HCI_Connection_Complete_Event.LinkType.SCO:
# Pass this on to the synchronous connection handler
forwarded_event = hci.HCI_Synchronous_Connection_Complete_Event(
status=event.status,
connection_handle=event.connection_handle,
bd_addr=event.bd_addr,
link_type=event.link_type,
transmission_interval=0,
retransmission_window=0,
rx_packet_length=0,
tx_packet_length=0,
air_mode=0,
)
self.on_hci_synchronous_connection_complete_event(forwarded_event)
return
if event.status == hci.HCI_SUCCESS:
# Create/update the connection
logger.debug(
f'### BR/EDR CONNECTION: [0x{event.connection_handle:04X}] '
f'{event.bd_addr}'
f'### BR/EDR ACL CONNECTION: [0x{event.connection_handle:04X}] '
f'{event.bd_addr} '
f'{event.link_type.name}'
)
connection = self.connections.get(event.connection_handle)
@@ -1581,6 +1600,9 @@ class Host(utils.EventEmitter):
event.bd_addr,
event.connection_handle,
event.link_type,
event.rx_packet_length,
event.tx_packet_length,
event.air_mode,
)
else:
logger.debug(f'### SCO CONNECTION FAILED: {event.status}')

View File

@@ -110,7 +110,7 @@ RFCOMM_DEFAULT_L2CAP_MTU = 2048
RFCOMM_DEFAULT_INITIAL_CREDITS = 7
RFCOMM_DEFAULT_MAX_CREDITS = 32
RFCOMM_DEFAULT_CREDIT_THRESHOLD = RFCOMM_DEFAULT_MAX_CREDITS // 2
RFCOMM_DEFAULT_MAX_FRAME_SIZE = 2000
RFCOMM_DEFAULT_MAX_FRAME_SIZE = 1000
RFCOMM_DYNAMIC_CHANNEL_NUMBER_START = 1
RFCOMM_DYNAMIC_CHANNEL_NUMBER_END = 30

File diff suppressed because it is too large Load Diff

View File

@@ -20,17 +20,110 @@ import contextlib
import functools
import json
import sys
import wave
import websockets.asyncio.server
import bumble.logging
from bumble import hci, hfp, rfcomm
from bumble.device import Connection, Device
from bumble.device import Connection, Device, ScoLink
from bumble.hfp import HfProtocol
from bumble.transport import open_transport
# -----------------------------------------------------------------------------
ws: websockets.asyncio.server.ServerConnection | None = None
hf_protocol: HfProtocol | None = None
input_wav: wave.Wave_read | None = None
output_wav: wave.Wave_write | None = None
# -----------------------------------------------------------------------------
def on_audio_packet(packet: hci.HCI_SynchronousDataPacket) -> None:
if (
packet.packet_status
== hci.HCI_SynchronousDataPacket.Status.CORRECTLY_RECEIVED_DATA
):
if output_wav:
# Save the PCM audio to the output
output_wav.writeframes(packet.data)
else:
print('!!! discarding packet with status ', packet.packet_status.name)
if input_wav and hf_protocol:
# Send PCM audio from the input
frame_count = len(packet.data) // 2
while frame_count:
# NOTE: we use a fixed number of frames here, this should likely be adjusted
# based on the transport parameters (like the USB max packet size)
chunk_size = min(frame_count, 16)
if not (pcm_data := input_wav.readframes(chunk_size)):
return
frame_count -= chunk_size
hf_protocol.dlc.multiplexer.l2cap_channel.connection.device.host.send_sco_sdu(
connection_handle=packet.connection_handle,
sdu=pcm_data,
)
# -----------------------------------------------------------------------------
def on_sco_connection(link: ScoLink) -> None:
print('### SCO connection established:', link)
if link.air_mode == hci.CodecID.TRANSPARENT:
print("@@@ The controller does not encode/decode voice")
return
link.sink = on_audio_packet
# -----------------------------------------------------------------------------
def on_sco_request(
link_type: int, connection: Connection, protocol: HfProtocol
) -> None:
if link_type == hci.HCI_Connection_Complete_Event.LinkType.SCO:
esco_parameters = hfp.ESCO_PARAMETERS[hfp.DefaultCodecParameters.SCO_CVSD_D1]
elif protocol.active_codec == hfp.AudioCodec.MSBC:
esco_parameters = hfp.ESCO_PARAMETERS[hfp.DefaultCodecParameters.ESCO_MSBC_T2]
elif protocol.active_codec == hfp.AudioCodec.CVSD:
esco_parameters = hfp.ESCO_PARAMETERS[hfp.DefaultCodecParameters.ESCO_CVSD_S4]
else:
raise RuntimeError("unknown active codec")
if connection.device.host.supports_command(
hci.HCI_ENHANCED_ACCEPT_SYNCHRONOUS_CONNECTION_REQUEST_COMMAND
):
connection.cancel_on_disconnection(
connection.device.send_async_command(
hci.HCI_Enhanced_Accept_Synchronous_Connection_Request_Command(
bd_addr=connection.peer_address, **esco_parameters.asdict()
)
)
)
elif connection.device.host.supports_command(
hci.HCI_ACCEPT_SYNCHRONOUS_CONNECTION_REQUEST_COMMAND
):
connection.cancel_on_disconnection(
connection.device.send_async_command(
hci.HCI_Accept_Synchronous_Connection_Request_Command(
bd_addr=connection.peer_address,
transmit_bandwidth=esco_parameters.transmit_bandwidth,
receive_bandwidth=esco_parameters.receive_bandwidth,
max_latency=esco_parameters.max_latency,
voice_setting=int(
hci.VoiceSetting(
input_sample_size=hci.VoiceSetting.InputSampleSize.SIZE_16_BITS,
input_data_format=hci.VoiceSetting.InputDataFormat.TWOS_COMPLEMENT,
)
),
retransmission_effort=esco_parameters.retransmission_effort,
packet_type=esco_parameters.packet_type,
)
)
)
else:
print('!!! no supported command for SCO connection request')
return
connection.on('sco_connection', on_sco_connection)
# -----------------------------------------------------------------------------
@@ -40,134 +133,172 @@ def on_dlc(dlc: rfcomm.DLC, configuration: hfp.HfConfiguration):
hf_protocol = HfProtocol(dlc, configuration)
asyncio.create_task(hf_protocol.run())
def on_sco_request(connection: Connection, link_type: int, protocol: HfProtocol):
if connection == protocol.dlc.multiplexer.l2cap_channel.connection:
if link_type == hci.HCI_Connection_Complete_Event.LinkType.SCO:
esco_parameters = hfp.ESCO_PARAMETERS[
hfp.DefaultCodecParameters.SCO_CVSD_D1
]
elif protocol.active_codec == hfp.AudioCodec.MSBC:
esco_parameters = hfp.ESCO_PARAMETERS[
hfp.DefaultCodecParameters.ESCO_MSBC_T2
]
elif protocol.active_codec == hfp.AudioCodec.CVSD:
esco_parameters = hfp.ESCO_PARAMETERS[
hfp.DefaultCodecParameters.ESCO_CVSD_S4
]
else:
raise RuntimeError("unknown active codec")
connection.cancel_on_disconnection(
connection.device.send_command(
hci.HCI_Enhanced_Accept_Synchronous_Connection_Request_Command(
bd_addr=connection.peer_address, **esco_parameters.asdict()
)
)
)
handler = functools.partial(on_sco_request, protocol=hf_protocol)
dlc.multiplexer.l2cap_channel.connection.device.on('sco_request', handler)
connection = dlc.multiplexer.l2cap_channel.connection
handler = functools.partial(
on_sco_request,
connection=connection,
protocol=hf_protocol,
)
connection.on('sco_request', handler)
dlc.multiplexer.l2cap_channel.once(
'close',
lambda: dlc.multiplexer.l2cap_channel.connection.device.remove_listener(
'sco_request', handler
),
lambda: connection.remove_listener('sco_request', handler),
)
def on_ag_indicator(indicator):
global ws
if ws:
asyncio.create_task(ws.send(str(indicator)))
hf_protocol.on('ag_indicator', on_ag_indicator)
hf_protocol.on('codec_negotiation', on_codec_negotiation)
# -----------------------------------------------------------------------------
def on_ag_indicator(indicator):
global ws
if ws:
asyncio.create_task(ws.send(str(indicator)))
# -----------------------------------------------------------------------------
def on_codec_negotiation(codec: hfp.AudioCodec):
print(f'### Negotiated codec: {codec.name}')
global output_wav
if output_wav:
output_wav.setnchannels(1)
output_wav.setsampwidth(2)
match codec:
case hfp.AudioCodec.CVSD:
output_wav.setframerate(8000)
case hfp.AudioCodec.MSBC:
output_wav.setframerate(16000)
# -----------------------------------------------------------------------------
async def run(device: Device, codec: str | None) -> None:
if codec is None:
supported_audio_codecs = [hfp.AudioCodec.CVSD, hfp.AudioCodec.MSBC]
else:
if codec == 'cvsd':
supported_audio_codecs = [hfp.AudioCodec.CVSD]
elif codec == 'msbc':
supported_audio_codecs = [hfp.AudioCodec.MSBC]
else:
print('Unknown codec: ', codec)
return
# Hands-Free profile configuration.
# TODO: load configuration from file.
configuration = hfp.HfConfiguration(
supported_hf_features=[
hfp.HfFeature.THREE_WAY_CALLING,
hfp.HfFeature.REMOTE_VOLUME_CONTROL,
hfp.HfFeature.ENHANCED_CALL_STATUS,
hfp.HfFeature.ENHANCED_CALL_CONTROL,
hfp.HfFeature.CODEC_NEGOTIATION,
hfp.HfFeature.HF_INDICATORS,
hfp.HfFeature.ESCO_S4_SETTINGS_SUPPORTED,
],
supported_hf_indicators=[
hfp.HfIndicator.BATTERY_LEVEL,
],
supported_audio_codecs=supported_audio_codecs,
)
# Create and register a server
rfcomm_server = rfcomm.Server(device)
# Listen for incoming DLC connections
channel_number = rfcomm_server.listen(lambda dlc: on_dlc(dlc, configuration))
print(f'### Listening for connection on channel {channel_number}')
# Advertise the HFP RFComm channel in the SDP
device.sdp_service_records = {
0x00010001: hfp.make_hf_sdp_records(0x00010001, channel_number, configuration)
}
# Let's go!
await device.power_on()
# Start being discoverable and connectable
await device.set_discoverable(True)
await device.set_connectable(True)
# Start the UI websocket server to offer a few buttons and input boxes
async def serve(websocket: websockets.asyncio.server.ServerConnection):
global ws
ws = websocket
async for message in websocket:
with contextlib.suppress(websockets.exceptions.ConnectionClosedOK):
print('Received: ', str(message))
parsed = json.loads(message)
message_type = parsed['type']
if message_type == 'at_command':
if hf_protocol is not None:
response = str(
await hf_protocol.execute_command(
parsed['command'],
response_type=hfp.AtResponseType.MULTIPLE,
)
)
await websocket.send(response)
elif message_type == 'query_call':
if hf_protocol:
response = str(await hf_protocol.query_current_calls())
await websocket.send(response)
await websockets.asyncio.server.serve(serve, 'localhost', 8989)
await asyncio.get_running_loop().create_future() # run forever
# -----------------------------------------------------------------------------
async def main() -> None:
if len(sys.argv) < 3:
print('Usage: run_classic_hfp.py <device-config> <transport-spec>')
print('example: run_classic_hfp.py classic2.json usb:04b4:f901')
print(
'Usage: run_hfp_handsfree.py <device-config> <transport-spec> '
'[codec] [input] [output]'
)
print('example: run_hfp_handsfree.py classic2.json usb:0')
return
print('<<< connecting to HCI...')
async with await open_transport(sys.argv[2]) as hci_transport:
print('<<< connected')
device_config = sys.argv[1]
transport_spec = sys.argv[2]
# Hands-Free profile configuration.
# TODO: load configuration from file.
configuration = hfp.HfConfiguration(
supported_hf_features=[
hfp.HfFeature.THREE_WAY_CALLING,
hfp.HfFeature.REMOTE_VOLUME_CONTROL,
hfp.HfFeature.ENHANCED_CALL_STATUS,
hfp.HfFeature.ENHANCED_CALL_CONTROL,
hfp.HfFeature.CODEC_NEGOTIATION,
hfp.HfFeature.HF_INDICATORS,
hfp.HfFeature.ESCO_S4_SETTINGS_SUPPORTED,
],
supported_hf_indicators=[
hfp.HfIndicator.BATTERY_LEVEL,
],
supported_audio_codecs=[
hfp.AudioCodec.CVSD,
hfp.AudioCodec.MSBC,
],
)
codec: str | None = None
if len(sys.argv) >= 4:
codec = sys.argv[3]
# Create a device
device = Device.from_config_file_with_hci(
sys.argv[1], hci_transport.source, hci_transport.sink
)
device.classic_enabled = True
input_file_name: str | None = None
if len(sys.argv) >= 5:
input_file_name = sys.argv[4]
# Create and register a server
rfcomm_server = rfcomm.Server(device)
output_file_name: str | None = None
if len(sys.argv) >= 6:
output_file_name = sys.argv[5]
# Listen for incoming DLC connections
channel_number = rfcomm_server.listen(lambda dlc: on_dlc(dlc, configuration))
print(f'### Listening for connection on channel {channel_number}')
global input_wav, output_wav
input_cm: contextlib.AbstractContextManager[wave.Wave_read | None] = (
wave.open(input_file_name, "rb")
if input_file_name
else contextlib.nullcontext(None)
)
output_cm: contextlib.AbstractContextManager[wave.Wave_write | None] = (
wave.open(output_file_name, "wb")
if output_file_name
else contextlib.nullcontext(None)
)
with input_cm as input_wav, output_cm as output_wav:
if input_wav and input_wav.getnchannels() != 1:
print("Mono input required")
return
if input_wav and input_wav.getsampwidth() != 2:
print("16-bit input required")
return
# Advertise the HFP RFComm channel in the SDP
device.sdp_service_records = {
0x00010001: hfp.make_hf_sdp_records(
0x00010001, channel_number, configuration
async with await open_transport(transport_spec) as transport:
device = Device.from_config_file_with_hci(
device_config, transport.source, transport.sink
)
}
# Let's go!
await device.power_on()
# Start being discoverable and connectable
await device.set_discoverable(True)
await device.set_connectable(True)
# Start the UI websocket server to offer a few buttons and input boxes
async def serve(websocket: websockets.asyncio.server.ServerConnection):
global ws
ws = websocket
async for message in websocket:
with contextlib.suppress(websockets.exceptions.ConnectionClosedOK):
print('Received: ', str(message))
parsed = json.loads(message)
message_type = parsed['type']
if message_type == 'at_command':
if hf_protocol is not None:
response = str(
await hf_protocol.execute_command(
parsed['command'],
response_type=hfp.AtResponseType.MULTIPLE,
)
)
await websocket.send(response)
elif message_type == 'query_call':
if hf_protocol:
response = str(await hf_protocol.query_current_calls())
await websocket.send(response)
await websockets.asyncio.server.serve(serve, 'localhost', 8989)
await hci_transport.source.terminated
device.classic_enabled = True
await run(device, codec)
# -----------------------------------------------------------------------------

View File

@@ -170,7 +170,9 @@ def format_code(ctx, check=False, diff=False):
@task
def check_types(ctx):
checklist = ["apps", "bumble", "examples", "tests", "tasks.py"]
print(">>> Running the type checker...")
try:
print("+++ Checking with mypy...")
ctx.run(f"mypy {' '.join(checklist)}")
except UnexpectedExit as exc:
print("Please check your code against the mypy messages.")