fix types

add workaround for some buggy controllers

apps/controller_info.py

@@ -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(

apps/controller_loopback.py

@@ -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())
 
 
 # -----------------------------------------------------------------------------

apps/scan.py

@@ -22,7 +22,7 @@ import click
 import bumble.logging
 from bumble import data_types
 from bumble.colors import color
-from bumble.device import Advertisement, Device
+from bumble.device import Advertisement, Device, DeviceConfiguration
 from bumble.hci import HCI_LE_1M_PHY, HCI_LE_CODED_PHY, Address, HCI_Constant
 from bumble.keys import JsonKeyStore
 from bumble.smp import AddressResolver
@@ -144,8 +144,14 @@ async def scan(
                 device_config, hci_source, hci_sink
             )
         else:
-            device = Device.with_hci(
-                'Bumble', 'F0:F1:F2:F3:F4:F5', hci_source, hci_sink
+            device = Device.from_config_with_hci(
+                DeviceConfiguration(
+                    name='Bumble',
+                    address=Address('F0:F1:F2:F3:F4:F5'),
+                    keystore='JsonKeyStore',
+                ),
+                hci_source,
+                hci_sink,
             )
 
         await device.power_on()

apps/usb_probe.py

@@ -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}'
                     )
 
 

bumble/avrcp.py

@@ -26,7 +26,7 @@ from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Sequen
 from dataclasses import dataclass, field
 from typing import ClassVar, SupportsBytes, TypeVar
 
-from bumble import avc, avctp, core, hci, l2cap, utils
+from bumble import avc, avctp, core, hci, l2cap, sdp, utils
 from bumble.colors import color
 from bumble.device import Connection, Device
 from bumble.sdp import (
@@ -194,82 +194,43 @@ class TargetFeatures(enum.IntFlag):
 
 
 # -----------------------------------------------------------------------------
-def make_controller_service_sdp_records(
-    service_record_handle: int,
-    avctp_version: tuple[int, int] = (1, 4),
-    avrcp_version: tuple[int, int] = (1, 6),
-    supported_features: int | ControllerFeatures = 1,
-) -> list[ServiceAttribute]:
-    avctp_version_int = avctp_version[0] << 8 | avctp_version[1]
-    avrcp_version_int = avrcp_version[0] << 8 | avrcp_version[1]
+@dataclass
+class ControllerServiceSdpRecord:
+    service_record_handle: int
+    avctp_version: tuple[int, int] = (1, 4)
+    avrcp_version: tuple[int, int] = (1, 6)
+    supported_features: int | ControllerFeatures = ControllerFeatures(1)
 
-    attributes = [
-        ServiceAttribute(
-            SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID,
-            DataElement.unsigned_integer_32(service_record_handle),
-        ),
-        ServiceAttribute(
-            SDP_BROWSE_GROUP_LIST_ATTRIBUTE_ID,
-            DataElement.sequence([DataElement.uuid(SDP_PUBLIC_BROWSE_ROOT)]),
-        ),
-        ServiceAttribute(
-            SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
-            DataElement.sequence(
-                [
-                    DataElement.uuid(core.BT_AV_REMOTE_CONTROL_SERVICE),
-                    DataElement.uuid(core.BT_AV_REMOTE_CONTROL_CONTROLLER_SERVICE),
-                ]
-            ),
-        ),
-        ServiceAttribute(
-            SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
-            DataElement.sequence(
-                [
-                    DataElement.sequence(
-                        [
-                            DataElement.uuid(core.BT_L2CAP_PROTOCOL_ID),
-                            DataElement.unsigned_integer_16(avctp.AVCTP_PSM),
-                        ]
-                    ),
-                    DataElement.sequence(
-                        [
-                            DataElement.uuid(core.BT_AVCTP_PROTOCOL_ID),
-                            DataElement.unsigned_integer_16(avctp_version_int),
-                        ]
-                    ),
-                ]
-            ),
-        ),
-        ServiceAttribute(
-            SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
-            DataElement.sequence(
-                [
-                    DataElement.sequence(
-                        [
-                            DataElement.uuid(core.BT_AV_REMOTE_CONTROL_SERVICE),
-                            DataElement.unsigned_integer_16(avrcp_version_int),
-                        ]
-                    ),
-                ]
-            ),
-        ),
-        ServiceAttribute(
-            SDP_SUPPORTED_FEATURES_ATTRIBUTE_ID,
-            DataElement.unsigned_integer_16(supported_features),
-        ),
-    ]
-    if supported_features & ControllerFeatures.SUPPORTS_BROWSING:
-        attributes.append(
+    def to_service_attributes(self) -> list[ServiceAttribute]:
+        avctp_version_int = self.avctp_version[0] << 8 | self.avctp_version[1]
+        avrcp_version_int = self.avrcp_version[0] << 8 | self.avrcp_version[1]
+
+        attributes = [
             ServiceAttribute(
-                SDP_ADDITIONAL_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
+                SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID,
+                DataElement.unsigned_integer_32(self.service_record_handle),
+            ),
+            ServiceAttribute(
+                SDP_BROWSE_GROUP_LIST_ATTRIBUTE_ID,
+                DataElement.sequence([DataElement.uuid(SDP_PUBLIC_BROWSE_ROOT)]),
+            ),
+            ServiceAttribute(
+                SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
+                DataElement.sequence(
+                    [
+                        DataElement.uuid(core.BT_AV_REMOTE_CONTROL_SERVICE),
+                        DataElement.uuid(core.BT_AV_REMOTE_CONTROL_CONTROLLER_SERVICE),
+                    ]
+                ),
+            ),
+            ServiceAttribute(
+                SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
                 DataElement.sequence(
                     [
                         DataElement.sequence(
                             [
                                 DataElement.uuid(core.BT_L2CAP_PROTOCOL_ID),
-                                DataElement.unsigned_integer_16(
-                                    avctp.AVCTP_BROWSING_PSM
-                                ),
+                                DataElement.unsigned_integer_16(avctp.AVCTP_PSM),
                             ]
                         ),
                         DataElement.sequence(
@@ -281,87 +242,130 @@ def make_controller_service_sdp_records(
                     ]
                 ),
             ),
-        )
-    return attributes
+            ServiceAttribute(
+                SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
+                DataElement.sequence(
+                    [
+                        DataElement.sequence(
+                            [
+                                DataElement.uuid(core.BT_AV_REMOTE_CONTROL_SERVICE),
+                                DataElement.unsigned_integer_16(avrcp_version_int),
+                            ]
+                        ),
+                    ]
+                ),
+            ),
+            ServiceAttribute(
+                SDP_SUPPORTED_FEATURES_ATTRIBUTE_ID,
+                DataElement.unsigned_integer_16(self.supported_features),
+            ),
+        ]
+        if self.supported_features & ControllerFeatures.SUPPORTS_BROWSING:
+            attributes.append(
+                ServiceAttribute(
+                    SDP_ADDITIONAL_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
+                    DataElement.sequence(
+                        [
+                            DataElement.sequence(
+                                [
+                                    DataElement.uuid(core.BT_L2CAP_PROTOCOL_ID),
+                                    DataElement.unsigned_integer_16(
+                                        avctp.AVCTP_BROWSING_PSM
+                                    ),
+                                ]
+                            ),
+                            DataElement.sequence(
+                                [
+                                    DataElement.uuid(core.BT_AVCTP_PROTOCOL_ID),
+                                    DataElement.unsigned_integer_16(avctp_version_int),
+                                ]
+                            ),
+                        ]
+                    ),
+                ),
+            )
+        return attributes
+
+    @classmethod
+    async def find(cls, connection: Connection) -> list[ControllerServiceSdpRecord]:
+        async with sdp.Client(connection) as sdp_client:
+            search_result = await sdp_client.search_attributes(
+                uuids=[core.BT_AV_REMOTE_CONTROL_CONTROLLER_SERVICE],
+                attribute_ids=[
+                    SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID,
+                    SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
+                    SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
+                    SDP_SUPPORTED_FEATURES_ATTRIBUTE_ID,
+                ],
+            )
+
+            records: list[ControllerServiceSdpRecord] = []
+            for attribute_lists in search_result:
+                record = cls(0)
+                for attribute in attribute_lists:
+                    if attribute.id == SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID:
+                        record.service_record_handle = attribute.value.value
+                    elif attribute.id == SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID:
+                        # [[L2CAP, PSM], [AVCTP, version]]
+                        record.avctp_version = (
+                            attribute.value.value[1].value[1].value >> 8,
+                            attribute.value.value[1].value[1].value & 0xFF,
+                        )
+                    elif (
+                        attribute.id
+                        == SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID
+                    ):
+                        # [[AV_REMOTE_CONTROL, version]]
+                        record.avrcp_version = (
+                            attribute.value.value[0].value[1].value >> 8,
+                            attribute.value.value[0].value[1].value & 0xFF,
+                        )
+                    elif attribute.id == SDP_SUPPORTED_FEATURES_ATTRIBUTE_ID:
+                        record.supported_features = ControllerFeatures(
+                            attribute.value.value
+                        )
+                records.append(record)
+            return records
 
 
 # -----------------------------------------------------------------------------
-def make_target_service_sdp_records(
-    service_record_handle: int,
-    avctp_version: tuple[int, int] = (1, 4),
-    avrcp_version: tuple[int, int] = (1, 6),
-    supported_features: int | TargetFeatures = 0x23,
-) -> list[ServiceAttribute]:
-    # TODO: support a way to compute the supported features from a feature list
-    avctp_version_int = avctp_version[0] << 8 | avctp_version[1]
-    avrcp_version_int = avrcp_version[0] << 8 | avrcp_version[1]
+@dataclass
+class TargetServiceSdpRecord:
+    service_record_handle: int
+    avctp_version: tuple[int, int] = (1, 4)
+    avrcp_version: tuple[int, int] = (1, 6)
+    supported_features: int | TargetFeatures = TargetFeatures(0x23)
 
-    attributes = [
-        ServiceAttribute(
-            SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID,
-            DataElement.unsigned_integer_32(service_record_handle),
-        ),
-        ServiceAttribute(
-            SDP_BROWSE_GROUP_LIST_ATTRIBUTE_ID,
-            DataElement.sequence([DataElement.uuid(SDP_PUBLIC_BROWSE_ROOT)]),
-        ),
-        ServiceAttribute(
-            SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
-            DataElement.sequence(
-                [
-                    DataElement.uuid(core.BT_AV_REMOTE_CONTROL_TARGET_SERVICE),
-                ]
-            ),
-        ),
-        ServiceAttribute(
-            SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
-            DataElement.sequence(
-                [
-                    DataElement.sequence(
-                        [
-                            DataElement.uuid(core.BT_L2CAP_PROTOCOL_ID),
-                            DataElement.unsigned_integer_16(avctp.AVCTP_PSM),
-                        ]
-                    ),
-                    DataElement.sequence(
-                        [
-                            DataElement.uuid(core.BT_AVCTP_PROTOCOL_ID),
-                            DataElement.unsigned_integer_16(avctp_version_int),
-                        ]
-                    ),
-                ]
-            ),
-        ),
-        ServiceAttribute(
-            SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
-            DataElement.sequence(
-                [
-                    DataElement.sequence(
-                        [
-                            DataElement.uuid(core.BT_AV_REMOTE_CONTROL_SERVICE),
-                            DataElement.unsigned_integer_16(avrcp_version_int),
-                        ]
-                    ),
-                ]
-            ),
-        ),
-        ServiceAttribute(
-            SDP_SUPPORTED_FEATURES_ATTRIBUTE_ID,
-            DataElement.unsigned_integer_16(supported_features),
-        ),
-    ]
-    if supported_features & TargetFeatures.SUPPORTS_BROWSING:
-        attributes.append(
+    def to_service_attributes(self) -> list[ServiceAttribute]:
+        # TODO: support a way to compute the supported features from a feature list
+        avctp_version_int = self.avctp_version[0] << 8 | self.avctp_version[1]
+        avrcp_version_int = self.avrcp_version[0] << 8 | self.avrcp_version[1]
+
+        attributes = [
             ServiceAttribute(
-                SDP_ADDITIONAL_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
+                SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID,
+                DataElement.unsigned_integer_32(self.service_record_handle),
+            ),
+            ServiceAttribute(
+                SDP_BROWSE_GROUP_LIST_ATTRIBUTE_ID,
+                DataElement.sequence([DataElement.uuid(SDP_PUBLIC_BROWSE_ROOT)]),
+            ),
+            ServiceAttribute(
+                SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
+                DataElement.sequence(
+                    [
+                        DataElement.uuid(core.BT_AV_REMOTE_CONTROL_TARGET_SERVICE),
+                    ]
+                ),
+            ),
+            ServiceAttribute(
+                SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
                 DataElement.sequence(
                     [
                         DataElement.sequence(
                             [
                                 DataElement.uuid(core.BT_L2CAP_PROTOCOL_ID),
-                                DataElement.unsigned_integer_16(
-                                    avctp.AVCTP_BROWSING_PSM
-                                ),
+                                DataElement.unsigned_integer_16(avctp.AVCTP_PSM),
                             ]
                         ),
                         DataElement.sequence(
@@ -373,8 +377,90 @@ def make_target_service_sdp_records(
                     ]
                 ),
             ),
-        )
-    return attributes
+            ServiceAttribute(
+                SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
+                DataElement.sequence(
+                    [
+                        DataElement.sequence(
+                            [
+                                DataElement.uuid(core.BT_AV_REMOTE_CONTROL_SERVICE),
+                                DataElement.unsigned_integer_16(avrcp_version_int),
+                            ]
+                        ),
+                    ]
+                ),
+            ),
+            ServiceAttribute(
+                SDP_SUPPORTED_FEATURES_ATTRIBUTE_ID,
+                DataElement.unsigned_integer_16(self.supported_features),
+            ),
+        ]
+        if self.supported_features & TargetFeatures.SUPPORTS_BROWSING:
+            attributes.append(
+                ServiceAttribute(
+                    SDP_ADDITIONAL_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
+                    DataElement.sequence(
+                        [
+                            DataElement.sequence(
+                                [
+                                    DataElement.uuid(core.BT_L2CAP_PROTOCOL_ID),
+                                    DataElement.unsigned_integer_16(
+                                        avctp.AVCTP_BROWSING_PSM
+                                    ),
+                                ]
+                            ),
+                            DataElement.sequence(
+                                [
+                                    DataElement.uuid(core.BT_AVCTP_PROTOCOL_ID),
+                                    DataElement.unsigned_integer_16(avctp_version_int),
+                                ]
+                            ),
+                        ]
+                    ),
+                ),
+            )
+        return attributes
+
+    @classmethod
+    async def find(cls, connection: Connection) -> list[TargetServiceSdpRecord]:
+        async with sdp.Client(connection) as sdp_client:
+            search_result = await sdp_client.search_attributes(
+                uuids=[core.BT_AV_REMOTE_CONTROL_TARGET_SERVICE],
+                attribute_ids=[
+                    SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID,
+                    SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
+                    SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
+                    SDP_SUPPORTED_FEATURES_ATTRIBUTE_ID,
+                ],
+            )
+
+            records: list[TargetServiceSdpRecord] = []
+            for attribute_lists in search_result:
+                record = cls(0)
+                for attribute in attribute_lists:
+                    if attribute.id == SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID:
+                        record.service_record_handle = attribute.value.value
+                    elif attribute.id == SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID:
+                        # [[L2CAP, PSM], [AVCTP, version]]
+                        record.avctp_version = (
+                            attribute.value.value[1].value[1].value >> 8,
+                            attribute.value.value[1].value[1].value & 0xFF,
+                        )
+                    elif (
+                        attribute.id
+                        == SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID
+                    ):
+                        # [[AV_REMOTE_CONTROL, version]]
+                        record.avrcp_version = (
+                            attribute.value.value[0].value[1].value >> 8,
+                            attribute.value.value[0].value[1].value & 0xFF,
+                        )
+                    elif attribute.id == SDP_SUPPORTED_FEATURES_ATTRIBUTE_ID:
+                        record.supported_features = TargetFeatures(
+                            attribute.value.value
+                        )
+                records.append(record)
+            return records
 
 
 # -----------------------------------------------------------------------------
@@ -1204,6 +1290,10 @@ class InformBatteryStatusOfCtResponse(Response):
 @dataclass
 class GetPlayStatusResponse(Response):
     pdu_id = PduId.GET_PLAY_STATUS
+
+    # TG doesn't support Song Length or Position.
+    UNAVAILABLE = 0xFFFFFFFF
+
     song_length: int = field(metadata=hci.metadata(">4"))
     song_position: int = field(metadata=hci.metadata(">4"))
     play_status: PlayStatus = field(metadata=PlayStatus.type_metadata(1))
@@ -1521,16 +1611,33 @@ class Delegate:
         def __init__(self, status_code: StatusCode) -> None:
             self.status_code = status_code
 
-    supported_events: list[EventId]
-    volume: int
+    class AvcError(Exception):
+        """The delegate AVC method failed, with a specified status code."""
 
-    def __init__(self, supported_events: Iterable[EventId] = ()) -> None:
+        def __init__(self, status_code: avc.ResponseFrame.ResponseCode) -> None:
+            self.status_code = status_code
+
+    supported_events: list[EventId]
+    supported_company_ids: list[int]
+    volume: int
+    playback_status: PlayStatus
+
+    def __init__(
+        self,
+        supported_events: Iterable[EventId] = (),
+        supported_company_ids: Iterable[int] = (AVRCP_BLUETOOTH_SIG_COMPANY_ID,),
+    ) -> None:
+        self.supported_company_ids = list(supported_company_ids)
         self.supported_events = list(supported_events)
         self.volume = 0
+        self.playback_status = PlayStatus.STOPPED
 
     async def get_supported_events(self) -> list[EventId]:
         return self.supported_events
 
+    async def get_supported_company_ids(self) -> list[int]:
+        return self.supported_company_ids
+
     async def set_absolute_volume(self, volume: int) -> None:
         """
         Set the absolute volume.
@@ -1543,6 +1650,19 @@ class Delegate:
     async def get_absolute_volume(self) -> int:
         return self.volume
 
+    async def on_key_event(
+        self,
+        key: avc.PassThroughFrame.OperationId,
+        pressed: bool,
+        data: bytes,
+    ) -> None:
+        logger.debug(
+            "@@@ on_key_event: key=%s, pressed=%s, data=%s", key, pressed, data.hex()
+        )
+
+    async def get_playback_status(self) -> PlayStatus:
+        return self.playback_status
+
     # TODO add other delegate methods
 
 
@@ -1756,6 +1876,19 @@ class Protocol(utils.EventEmitter):
             if isinstance(capability, EventId)
         )
 
+    async def get_supported_company_ids(self) -> list[int]:
+        """Get the list of events supported by the connected peer."""
+        response_context = await self.send_avrcp_command(
+            avc.CommandFrame.CommandType.STATUS,
+            GetCapabilitiesCommand(GetCapabilitiesCommand.CapabilityId.COMPANY_ID),
+        )
+        response = self._check_response(response_context, GetCapabilitiesResponse)
+        return list(
+            int.from_bytes(capability, 'big')
+            for capability in response.capabilities
+            if isinstance(capability, bytes)
+        )
+
     async def get_play_status(self) -> SongAndPlayStatus:
         """Get the play status of the connected peer."""
         response_context = await self.send_avrcp_command(
@@ -2052,16 +2185,28 @@ class Protocol(utils.EventEmitter):
             return
 
         if isinstance(command, avc.PassThroughCommandFrame):
-            # TODO: delegate
-            response = avc.PassThroughResponseFrame(
-                avc.ResponseFrame.ResponseCode.ACCEPTED,
-                command.subunit_type,
-                command.subunit_id,
-                command.state_flag,
-                command.operation_id,
-                command.operation_data,
-            )
-            self.send_response(transaction_label, response)
+
+            async def dispatch_key_event() -> None:
+                try:
+                    await self.delegate.on_key_event(
+                        command.operation_id,
+                        command.state_flag == avc.PassThroughFrame.StateFlag.PRESSED,
+                        command.operation_data,
+                    )
+                    response_code = avc.ResponseFrame.ResponseCode.ACCEPTED
+                except Delegate.AvcError as error:
+                    logger.exception("delegate method raised exception")
+                    response_code = error.status_code
+                except Exception:
+                    logger.exception("delegate method raised exception")
+                    response_code = avc.ResponseFrame.ResponseCode.REJECTED
+                self.send_passthrough_response(
+                    transaction_label=transaction_label,
+                    command=command,
+                    response_code=response_code,
+                )
+
+            utils.AsyncRunner.spawn(dispatch_key_event())
             return
 
         # TODO handle other types
@@ -2141,6 +2286,8 @@ class Protocol(utils.EventEmitter):
                 self._on_set_absolute_volume_command(transaction_label, command)
             elif isinstance(command, RegisterNotificationCommand):
                 self._on_register_notification_command(transaction_label, command)
+            elif isinstance(command, GetPlayStatusCommand):
+                self._on_get_play_status_command(transaction_label, command)
             else:
                 # Not supported.
                 # TODO: check that this is the right way to respond in this case.
@@ -2364,17 +2511,27 @@ class Protocol(utils.EventEmitter):
         logger.debug(f"<<< AVRCP command PDU: {command}")
 
         async def get_supported_events() -> None:
+            capabilities: Sequence[bytes | SupportsBytes]
             if (
                 command.capability_id
-                != GetCapabilitiesCommand.CapabilityId.EVENTS_SUPPORTED
+                == GetCapabilitiesCommand.CapabilityId.EVENTS_SUPPORTED
             ):
-                raise core.InvalidArgumentError()
-
-            supported_events = await self.delegate.get_supported_events()
+                capabilities = await self.delegate.get_supported_events()
+            elif (
+                command.capability_id == GetCapabilitiesCommand.CapabilityId.COMPANY_ID
+            ):
+                company_ids = await self.delegate.get_supported_company_ids()
+                capabilities = [
+                    company_id.to_bytes(3, 'big') for company_id in company_ids
+                ]
+            else:
+                raise core.InvalidArgumentError(
+                    f"Unsupported capability: {command.capability_id}"
+                )
             self.send_avrcp_response(
                 transaction_label,
                 avc.ResponseFrame.ResponseCode.IMPLEMENTED_OR_STABLE,
-                GetCapabilitiesResponse(command.capability_id, supported_events),
+                GetCapabilitiesResponse(command.capability_id, capabilities),
             )
 
         self._delegate_command(transaction_label, command, get_supported_events())
@@ -2395,6 +2552,26 @@ class Protocol(utils.EventEmitter):
 
         self._delegate_command(transaction_label, command, set_absolute_volume())
 
+    def _on_get_play_status_command(
+        self, transaction_label: int, command: GetPlayStatusCommand
+    ) -> None:
+        logger.debug("<<< AVRCP command PDU: %s", command)
+
+        async def get_playback_status() -> None:
+            play_status: PlayStatus = await self.delegate.get_playback_status()
+            self.send_avrcp_response(
+                transaction_label,
+                avc.ResponseFrame.ResponseCode.IMPLEMENTED_OR_STABLE,
+                GetPlayStatusResponse(
+                    # TODO: Delegate this.
+                    song_length=GetPlayStatusResponse.UNAVAILABLE,
+                    song_position=GetPlayStatusResponse.UNAVAILABLE,
+                    play_status=play_status,
+                ),
+            )
+
+        self._delegate_command(transaction_label, command, get_playback_status())
+
     def _on_register_notification_command(
         self, transaction_label: int, command: RegisterNotificationCommand
     ) -> None:
@@ -2410,28 +2587,27 @@ class Protocol(utils.EventEmitter):
                 )
                 return
 
+            response: Response
             if command.event_id == EventId.VOLUME_CHANGED:
                 volume = await self.delegate.get_absolute_volume()
                 response = RegisterNotificationResponse(VolumeChangedEvent(volume))
-                self.send_avrcp_response(
-                    transaction_label,
-                    avc.ResponseFrame.ResponseCode.INTERIM,
-                    response,
+            elif command.event_id == EventId.PLAYBACK_STATUS_CHANGED:
+                playback_status = await self.delegate.get_playback_status()
+                response = RegisterNotificationResponse(
+                    PlaybackStatusChangedEvent(play_status=playback_status)
                 )
-                self._register_notification_listener(transaction_label, command)
+            elif command.event_id == EventId.NOW_PLAYING_CONTENT_CHANGED:
+                playback_status = await self.delegate.get_playback_status()
+                response = RegisterNotificationResponse(NowPlayingContentChangedEvent())
+            else:
+                logger.warning("Event supported but not handled %s", command.event_id)
                 return
 
-            if command.event_id == EventId.PLAYBACK_STATUS_CHANGED:
-                # TODO: testing only, use delegate
-                response = RegisterNotificationResponse(
-                    PlaybackStatusChangedEvent(play_status=PlayStatus.PLAYING)
-                )
-                self.send_avrcp_response(
-                    transaction_label,
-                    avc.ResponseFrame.ResponseCode.INTERIM,
-                    response,
-                )
-                self._register_notification_listener(transaction_label, command)
-                return
+            self.send_avrcp_response(
+                transaction_label,
+                avc.ResponseFrame.ResponseCode.INTERIM,
+                response,
+            )
+            self._register_notification_listener(transaction_label, command)
 
         self._delegate_command(transaction_label, command, register_notification())

bumble/device.py

@@ -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"
@@ -3748,6 +3751,292 @@ class Device(utils.CompositeEventEmitter):
                 page_scan_enabled=self.connectable,
             )
 
+    async def connect_le(
+        self,
+        peer_address: hci.Address | str,
+        connection_parameters_preferences: (
+            dict[hci.Phy, ConnectionParametersPreferences] | None
+        ) = None,
+        own_address_type: hci.OwnAddressType = hci.OwnAddressType.RANDOM,
+        timeout: float | None = DEVICE_DEFAULT_CONNECT_TIMEOUT,
+    ) -> Connection:
+        # Check that there isn't already a pending connection
+        if self.is_le_connecting:
+            raise InvalidStateError('connection already pending')
+
+        try_resolve = not self.address_resolution_offload
+        if isinstance(peer_address, str):
+            try:
+                peer_address = hci.Address.from_string_for_transport(
+                    peer_address, PhysicalTransport.LE
+                )
+            except (InvalidArgumentError, ValueError):
+                # If the address is not parsable, assume it is a name instead
+                logger.debug('looking for peer by name')
+                assert isinstance(peer_address, str)
+                peer_address = await self.find_peer_by_name(
+                    peer_address, PhysicalTransport.LE
+                )  # TODO: timeout
+                try_resolve = False
+
+        assert isinstance(peer_address, hci.Address)
+
+        if (
+            try_resolve
+            and self.address_resolver is not None
+            and self.address_resolver.can_resolve_to(peer_address)
+        ):
+            # If we have an IRK for this address, we should resolve.
+            logger.debug('have IRK for address, resolving...')
+            peer_address = await self.find_peer_by_identity_address(
+                peer_address
+            )  # TODO: timeout
+
+        def on_connection(connection):
+            pending_connection.set_result(connection)
+
+        def on_connection_failure(error: core.ConnectionError):
+            pending_connection.set_exception(error)
+
+        # Create a future so that we can wait for the connection result
+        pending_connection = asyncio.get_running_loop().create_future()
+        self.on(self.EVENT_CONNECTION, on_connection)
+        self.on(self.EVENT_CONNECTION_FAILURE, on_connection_failure)
+
+        try:
+            # Tell the controller to connect
+            if connection_parameters_preferences is None:
+                connection_parameters_preferences = {
+                    hci.HCI_LE_1M_PHY: ConnectionParametersPreferences.default
+                }
+
+            self.connect_own_address_type = own_address_type
+
+            if self.host.supports_command(
+                hci.HCI_LE_EXTENDED_CREATE_CONNECTION_COMMAND
+            ):
+                # Only keep supported PHYs
+                phys = sorted(
+                    list(
+                        set(
+                            filter(
+                                self.supports_le_phy,
+                                connection_parameters_preferences.keys(),
+                            )
+                        )
+                    )
+                )
+                if not phys:
+                    raise InvalidArgumentError('at least one supported PHY needed')
+
+                phy_count = len(phys)
+                initiating_phys = hci.phy_list_to_bits(phys)
+
+                connection_interval_mins = [
+                    int(
+                        connection_parameters_preferences[phy].connection_interval_min
+                        / 1.25
+                    )
+                    for phy in phys
+                ]
+                connection_interval_maxs = [
+                    int(
+                        connection_parameters_preferences[phy].connection_interval_max
+                        / 1.25
+                    )
+                    for phy in phys
+                ]
+                max_latencies = [
+                    connection_parameters_preferences[phy].max_latency for phy in phys
+                ]
+                supervision_timeouts = [
+                    int(connection_parameters_preferences[phy].supervision_timeout / 10)
+                    for phy in phys
+                ]
+                min_ce_lengths = [
+                    int(connection_parameters_preferences[phy].min_ce_length / 0.625)
+                    for phy in phys
+                ]
+                max_ce_lengths = [
+                    int(connection_parameters_preferences[phy].max_ce_length / 0.625)
+                    for phy in phys
+                ]
+
+                await self.send_async_command(
+                    hci.HCI_LE_Extended_Create_Connection_Command(
+                        initiator_filter_policy=0,
+                        own_address_type=own_address_type,
+                        peer_address_type=peer_address.address_type,
+                        peer_address=peer_address,
+                        initiating_phys=initiating_phys,
+                        scan_intervals=(
+                            int(DEVICE_DEFAULT_CONNECT_SCAN_INTERVAL / 0.625),
+                        )
+                        * phy_count,
+                        scan_windows=(int(DEVICE_DEFAULT_CONNECT_SCAN_WINDOW / 0.625),)
+                        * phy_count,
+                        connection_interval_mins=connection_interval_mins,
+                        connection_interval_maxs=connection_interval_maxs,
+                        max_latencies=max_latencies,
+                        supervision_timeouts=supervision_timeouts,
+                        min_ce_lengths=min_ce_lengths,
+                        max_ce_lengths=max_ce_lengths,
+                    )
+                )
+            else:
+                if hci.HCI_LE_1M_PHY not in connection_parameters_preferences:
+                    raise InvalidArgumentError('1M PHY preferences required')
+
+                prefs = connection_parameters_preferences[hci.HCI_LE_1M_PHY]
+                await self.send_async_command(
+                    hci.HCI_LE_Create_Connection_Command(
+                        le_scan_interval=int(
+                            DEVICE_DEFAULT_CONNECT_SCAN_INTERVAL / 0.625
+                        ),
+                        le_scan_window=int(DEVICE_DEFAULT_CONNECT_SCAN_WINDOW / 0.625),
+                        initiator_filter_policy=0,
+                        peer_address_type=peer_address.address_type,
+                        peer_address=peer_address,
+                        own_address_type=own_address_type,
+                        connection_interval_min=int(
+                            prefs.connection_interval_min / 1.25
+                        ),
+                        connection_interval_max=int(
+                            prefs.connection_interval_max / 1.25
+                        ),
+                        max_latency=prefs.max_latency,
+                        supervision_timeout=int(prefs.supervision_timeout / 10),
+                        min_ce_length=int(prefs.min_ce_length / 0.625),
+                        max_ce_length=int(prefs.max_ce_length / 0.625),
+                    )
+                )
+
+            # Wait for the connection process to complete
+            self.le_connecting = True
+
+            if timeout is None:
+                return await utils.cancel_on_event(
+                    self, Device.EVENT_FLUSH, pending_connection
+                )
+
+            try:
+                return await asyncio.wait_for(
+                    asyncio.shield(pending_connection), timeout
+                )
+            except asyncio.TimeoutError:
+                await self.send_sync_command(
+                    hci.HCI_LE_Create_Connection_Cancel_Command()
+                )
+
+                try:
+                    return await utils.cancel_on_event(
+                        self, Device.EVENT_FLUSH, pending_connection
+                    )
+                except core.ConnectionError as error:
+                    raise core.TimeoutError() from error
+        finally:
+            self.remove_listener(self.EVENT_CONNECTION, on_connection)
+            self.remove_listener(self.EVENT_CONNECTION_FAILURE, on_connection_failure)
+            self.le_connecting = False
+            self.connect_own_address_type = None
+
+    async def connect_classic(
+        self,
+        peer_address: hci.Address | str,
+        timeout: float | None = DEVICE_DEFAULT_CONNECT_TIMEOUT,
+    ) -> Connection:
+        if isinstance(peer_address, str):
+            try:
+                peer_address = hci.Address.from_string_for_transport(
+                    peer_address, PhysicalTransport.BR_EDR
+                )
+            except (InvalidArgumentError, ValueError):
+                # If the address is not parsable, assume it is a name instead
+                logger.debug('looking for peer by name')
+                assert isinstance(peer_address, str)
+                peer_address = await self.find_peer_by_name(
+                    peer_address, PhysicalTransport.BR_EDR
+                )  # TODO: timeout
+        else:
+            # All BR/EDR addresses should be public addresses
+            if peer_address.address_type != hci.Address.PUBLIC_DEVICE_ADDRESS:
+                raise InvalidArgumentError('BR/EDR addresses must be PUBLIC')
+
+        assert isinstance(peer_address, hci.Address)
+
+        def on_connection(connection):
+            if (
+                # match BR/EDR connection event against peer address
+                connection.transport == PhysicalTransport.BR_EDR
+                and connection.peer_address == peer_address
+            ):
+                pending_connection.set_result(connection)
+
+        def on_connection_failure(error: core.ConnectionError):
+            if (
+                # match BR/EDR connection failure event against peer address
+                error.transport == PhysicalTransport.BR_EDR
+                and error.peer_address == peer_address
+            ):
+                pending_connection.set_exception(error)
+
+        # Create a future so that we can wait for the connection result
+        pending_connection = asyncio.get_running_loop().create_future()
+        self.on(self.EVENT_CONNECTION, on_connection)
+        self.on(self.EVENT_CONNECTION_FAILURE, on_connection_failure)
+
+        try:
+            # Save pending connection
+            self.pending_connections[peer_address] = Connection(
+                device=self,
+                handle=0,
+                transport=core.PhysicalTransport.BR_EDR,
+                self_address=self.public_address,
+                self_resolvable_address=None,
+                peer_address=peer_address,
+                peer_resolvable_address=None,
+                role=hci.Role.CENTRAL,
+                parameters=Connection.Parameters(0, 0, 0),
+            )
+
+            # TODO: allow passing other settings
+            await self.send_async_command(
+                hci.HCI_Create_Connection_Command(
+                    bd_addr=peer_address,
+                    packet_type=0xCC18,  # FIXME: change
+                    page_scan_repetition_mode=hci.HCI_R2_PAGE_SCAN_REPETITION_MODE,
+                    clock_offset=0x0000,
+                    allow_role_switch=0x01,
+                    reserved=0,
+                )
+            )
+
+            # Wait for the connection process to complete
+            if timeout is None:
+                return await utils.cancel_on_event(
+                    self, Device.EVENT_FLUSH, pending_connection
+                )
+
+            try:
+                return await asyncio.wait_for(
+                    asyncio.shield(pending_connection), timeout
+                )
+            except asyncio.TimeoutError:
+                await self.send_sync_command(
+                    hci.HCI_Create_Connection_Cancel_Command(bd_addr=peer_address)
+                )
+
+                try:
+                    return await utils.cancel_on_event(
+                        self, Device.EVENT_FLUSH, pending_connection
+                    )
+                except core.ConnectionError as error:
+                    raise core.TimeoutError() from error
+        finally:
+            self.remove_listener(self.EVENT_CONNECTION, on_connection)
+            self.remove_listener(self.EVENT_CONNECTION_FAILURE, on_connection_failure)
+            self.pending_connections.pop(peer_address, None)
+
     async def connect(
         self,
         peer_address: hci.Address | str,
@@ -3769,9 +4058,9 @@ class Device(utils.CompositeEventEmitter):
           peer_address:
             hci.Address or name of the device to connect to.
             If a string is passed:
-              If the string is an address followed by a `@` suffix, the `always_resolve`
-              argument is implicitly set to True, so the connection is made to the
-              address after resolution.
+              [deprecated] If the string is an address followed by a `@` suffix, the
+              `always_resolve`argument is implicitly set to True, so the connection is
+              made to the address after resolution.
               If the string is any other address, the connection is made to that
               address (with or without address resolution, depending on the
               `always_resolve` argument).
@@ -3795,271 +4084,29 @@ class Device(utils.CompositeEventEmitter):
             Pass None for an unlimited time.
 
           always_resolve:
-            (BLE only, ignored for BR/EDR)
-            If True, always initiate a scan, resolving addresses, and connect to the
-            address that resolves to `peer_address`.
+            [deprecated] (ignore)
         '''
 
-        # Check parameters
-        if transport not in (PhysicalTransport.LE, PhysicalTransport.BR_EDR):
-            raise InvalidArgumentError('invalid transport')
-        transport = core.PhysicalTransport(transport)
+        # Connect using the appropriate transport
+        # (auto-correct the transport based on declared capabilities)
+        if transport == PhysicalTransport.LE or (
+            self.le_enabled and not self.classic_enabled
+        ):
+            return await self.connect_le(
+                peer_address=peer_address,
+                connection_parameters_preferences=connection_parameters_preferences,
+                own_address_type=own_address_type,
+                timeout=timeout,
+            )
 
-        # Adjust the transport automatically if we need to
-        if transport == PhysicalTransport.LE and not self.le_enabled:
-            transport = PhysicalTransport.BR_EDR
-        elif transport == PhysicalTransport.BR_EDR and not self.classic_enabled:
-            transport = PhysicalTransport.LE
+        if transport == PhysicalTransport.BR_EDR or (
+            self.classic_enabled and not self.le_enabled
+        ):
+            return await self.connect_classic(
+                peer_address=peer_address, timeout=timeout
+            )
 
-        # Check that there isn't already a pending connection
-        if transport == PhysicalTransport.LE and self.is_le_connecting:
-            raise InvalidStateError('connection already pending')
-
-        if isinstance(peer_address, str):
-            try:
-                if transport == PhysicalTransport.LE and peer_address.endswith('@'):
-                    peer_address = hci.Address.from_string_for_transport(
-                        peer_address[:-1], transport
-                    )
-                    always_resolve = True
-                    logger.debug('forcing address resolution')
-                else:
-                    peer_address = hci.Address.from_string_for_transport(
-                        peer_address, transport
-                    )
-            except (InvalidArgumentError, ValueError):
-                # If the address is not parsable, assume it is a name instead
-                always_resolve = False
-                logger.debug('looking for peer by name')
-                assert isinstance(peer_address, str)
-                peer_address = await self.find_peer_by_name(
-                    peer_address, transport
-                )  # TODO: timeout
-        else:
-            # All BR/EDR addresses should be public addresses
-            if (
-                transport == PhysicalTransport.BR_EDR
-                and peer_address.address_type != hci.Address.PUBLIC_DEVICE_ADDRESS
-            ):
-                raise InvalidArgumentError('BR/EDR addresses must be PUBLIC')
-
-        assert isinstance(peer_address, hci.Address)
-
-        if transport == PhysicalTransport.LE and always_resolve:
-            logger.debug('resolving address')
-            peer_address = await self.find_peer_by_identity_address(
-                peer_address
-            )  # TODO: timeout
-
-        def on_connection(connection):
-            if transport == PhysicalTransport.LE or (
-                # match BR/EDR connection event against peer address
-                connection.transport == transport
-                and connection.peer_address == peer_address
-            ):
-                pending_connection.set_result(connection)
-
-        def on_connection_failure(error: core.ConnectionError):
-            if transport == PhysicalTransport.LE or (
-                # match BR/EDR connection failure event against peer address
-                error.transport == transport
-                and error.peer_address == peer_address
-            ):
-                pending_connection.set_exception(error)
-
-        # Create a future so that we can wait for the connection's result
-        pending_connection = asyncio.get_running_loop().create_future()
-        self.on(self.EVENT_CONNECTION, on_connection)
-        self.on(self.EVENT_CONNECTION_FAILURE, on_connection_failure)
-
-        try:
-            # Tell the controller to connect
-            if transport == PhysicalTransport.LE:
-                if connection_parameters_preferences is None:
-                    if connection_parameters_preferences is None:
-                        connection_parameters_preferences = {
-                            hci.HCI_LE_1M_PHY: ConnectionParametersPreferences.default
-                        }
-
-                self.connect_own_address_type = own_address_type
-
-                if self.host.supports_command(
-                    hci.HCI_LE_EXTENDED_CREATE_CONNECTION_COMMAND
-                ):
-                    # Only keep supported PHYs
-                    phys = sorted(
-                        list(
-                            set(
-                                filter(
-                                    self.supports_le_phy,
-                                    connection_parameters_preferences.keys(),
-                                )
-                            )
-                        )
-                    )
-                    if not phys:
-                        raise InvalidArgumentError('at least one supported PHY needed')
-
-                    phy_count = len(phys)
-                    initiating_phys = hci.phy_list_to_bits(phys)
-
-                    connection_interval_mins = [
-                        int(
-                            connection_parameters_preferences[
-                                phy
-                            ].connection_interval_min
-                            / 1.25
-                        )
-                        for phy in phys
-                    ]
-                    connection_interval_maxs = [
-                        int(
-                            connection_parameters_preferences[
-                                phy
-                            ].connection_interval_max
-                            / 1.25
-                        )
-                        for phy in phys
-                    ]
-                    max_latencies = [
-                        connection_parameters_preferences[phy].max_latency
-                        for phy in phys
-                    ]
-                    supervision_timeouts = [
-                        int(
-                            connection_parameters_preferences[phy].supervision_timeout
-                            / 10
-                        )
-                        for phy in phys
-                    ]
-                    min_ce_lengths = [
-                        int(
-                            connection_parameters_preferences[phy].min_ce_length / 0.625
-                        )
-                        for phy in phys
-                    ]
-                    max_ce_lengths = [
-                        int(
-                            connection_parameters_preferences[phy].max_ce_length / 0.625
-                        )
-                        for phy in phys
-                    ]
-
-                    await self.send_async_command(
-                        hci.HCI_LE_Extended_Create_Connection_Command(
-                            initiator_filter_policy=0,
-                            own_address_type=own_address_type,
-                            peer_address_type=peer_address.address_type,
-                            peer_address=peer_address,
-                            initiating_phys=initiating_phys,
-                            scan_intervals=(
-                                int(DEVICE_DEFAULT_CONNECT_SCAN_INTERVAL / 0.625),
-                            )
-                            * phy_count,
-                            scan_windows=(
-                                int(DEVICE_DEFAULT_CONNECT_SCAN_WINDOW / 0.625),
-                            )
-                            * phy_count,
-                            connection_interval_mins=connection_interval_mins,
-                            connection_interval_maxs=connection_interval_maxs,
-                            max_latencies=max_latencies,
-                            supervision_timeouts=supervision_timeouts,
-                            min_ce_lengths=min_ce_lengths,
-                            max_ce_lengths=max_ce_lengths,
-                        )
-                    )
-                else:
-                    if hci.HCI_LE_1M_PHY not in connection_parameters_preferences:
-                        raise InvalidArgumentError('1M PHY preferences required')
-
-                    prefs = connection_parameters_preferences[hci.HCI_LE_1M_PHY]
-                    await self.send_async_command(
-                        hci.HCI_LE_Create_Connection_Command(
-                            le_scan_interval=int(
-                                DEVICE_DEFAULT_CONNECT_SCAN_INTERVAL / 0.625
-                            ),
-                            le_scan_window=int(
-                                DEVICE_DEFAULT_CONNECT_SCAN_WINDOW / 0.625
-                            ),
-                            initiator_filter_policy=0,
-                            peer_address_type=peer_address.address_type,
-                            peer_address=peer_address,
-                            own_address_type=own_address_type,
-                            connection_interval_min=int(
-                                prefs.connection_interval_min / 1.25
-                            ),
-                            connection_interval_max=int(
-                                prefs.connection_interval_max / 1.25
-                            ),
-                            max_latency=prefs.max_latency,
-                            supervision_timeout=int(prefs.supervision_timeout / 10),
-                            min_ce_length=int(prefs.min_ce_length / 0.625),
-                            max_ce_length=int(prefs.max_ce_length / 0.625),
-                        )
-                    )
-            else:
-                # Save pending connection
-                self.pending_connections[peer_address] = Connection(
-                    device=self,
-                    handle=0,
-                    transport=core.PhysicalTransport.BR_EDR,
-                    self_address=self.public_address,
-                    self_resolvable_address=None,
-                    peer_address=peer_address,
-                    peer_resolvable_address=None,
-                    role=hci.Role.CENTRAL,
-                    parameters=Connection.Parameters(0, 0, 0),
-                )
-
-                # TODO: allow passing other settings
-                await self.send_async_command(
-                    hci.HCI_Create_Connection_Command(
-                        bd_addr=peer_address,
-                        packet_type=0xCC18,  # FIXME: change
-                        page_scan_repetition_mode=hci.HCI_R2_PAGE_SCAN_REPETITION_MODE,
-                        clock_offset=0x0000,
-                        allow_role_switch=0x01,
-                        reserved=0,
-                    )
-                )
-
-            # Wait for the connection process to complete
-            if transport == PhysicalTransport.LE:
-                self.le_connecting = True
-
-            if timeout is None:
-                return await utils.cancel_on_event(
-                    self, Device.EVENT_FLUSH, pending_connection
-                )
-
-            try:
-                return await asyncio.wait_for(
-                    asyncio.shield(pending_connection), timeout
-                )
-            except asyncio.TimeoutError:
-                if transport == PhysicalTransport.LE:
-                    await self.send_sync_command(
-                        hci.HCI_LE_Create_Connection_Cancel_Command()
-                    )
-                else:
-                    await self.send_sync_command(
-                        hci.HCI_Create_Connection_Cancel_Command(bd_addr=peer_address)
-                    )
-
-                try:
-                    return await utils.cancel_on_event(
-                        self, Device.EVENT_FLUSH, pending_connection
-                    )
-                except core.ConnectionError as error:
-                    raise core.TimeoutError() from error
-        finally:
-            self.remove_listener(self.EVENT_CONNECTION, on_connection)
-            self.remove_listener(self.EVENT_CONNECTION_FAILURE, on_connection_failure)
-            if transport == PhysicalTransport.LE:
-                self.le_connecting = False
-                self.connect_own_address_type = None
-            else:
-                self.pending_connections.pop(peer_address, None)
+        raise ValueError('invalid transport')
 
     async def accept(
         self,
@@ -4706,6 +4753,8 @@ class Device(utils.CompositeEventEmitter):
         Scan for a peer with a resolvable address that can be resolved to a given
         identity address.
         """
+        if self.address_resolver is None:
+            raise InvalidStateError('no resolver')
 
         # Create a future to wait for an address to be found
         peer_address = asyncio.get_running_loop().create_future()
@@ -5922,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 (
@@ -5932,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}')
@@ -6291,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
@@ -6309,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}, '
@@ -6321,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]
@@ -6332,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
@@ -6795,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,

bumble/hci.py

@@ -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,

bumble/hfp.py

@@ -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

bumble/host.py

@@ -616,22 +616,28 @@ class Host(utils.EventEmitter):
         if self.supports_command(
             hci.HCI_LE_READ_NUMBER_OF_SUPPORTED_ADVERTISING_SETS_COMMAND
         ):
-            response10 = await self.send_sync_command(
-                hci.HCI_LE_Read_Number_Of_Supported_Advertising_Sets_Command()
-            )
-            self.number_of_supported_advertising_sets = (
-                response10.num_supported_advertising_sets
-            )
+            try:
+                response10 = await self.send_sync_command(
+                    hci.HCI_LE_Read_Number_Of_Supported_Advertising_Sets_Command()
+                )
+                self.number_of_supported_advertising_sets = (
+                    response10.num_supported_advertising_sets
+                )
+            except hci.HCI_Error:
+                logger.warning('Failed to read number of supported advertising sets')
 
         if self.supports_command(
             hci.HCI_LE_READ_MAXIMUM_ADVERTISING_DATA_LENGTH_COMMAND
         ):
-            response11 = await self.send_sync_command(
-                hci.HCI_LE_Read_Maximum_Advertising_Data_Length_Command()
-            )
-            self.maximum_advertising_data_length = (
-                response11.max_advertising_data_length
-            )
+            try:
+                response11 = await self.send_sync_command(
+                    hci.HCI_LE_Read_Maximum_Advertising_Data_Length_Command()
+                )
+                self.maximum_advertising_data_length = (
+                    response11.max_advertising_data_length
+                )
+            except hci.HCI_Error:
+                logger.warning('Failed to read maximum advertising data length')
 
     @property
     def controller(self) -> TransportSink | None:
@@ -680,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
@@ -776,6 +784,20 @@ class Host(utils.EventEmitter):
     ) -> hci.HCI_Command_Complete_Event[_RP]:
         response = await self._send_command(command, response_timeout)
 
+        # For unknown HCI commands, some controllers return Command Status instead of
+        # Command Complete.
+        if (
+            isinstance(response, hci.HCI_Command_Status_Event)
+            and response.status == hci.HCI_ErrorCode.UNKNOWN_HCI_COMMAND_ERROR
+        ):
+            return hci.HCI_Command_Complete_Event(
+                num_hci_command_packets=response.num_hci_command_packets,
+                command_opcode=command.op_code,
+                return_parameters=hci.HCI_StatusReturnParameters(
+                    status=hci.HCI_ErrorCode(response.status)
+                ),  # type: ignore
+            )
+
         # Check that the response is of the expected type
         assert isinstance(response, hci.HCI_Command_Complete_Event)
 
@@ -789,19 +811,25 @@ class Host(utils.EventEmitter):
     ) -> hci.HCI_ErrorCode:
         response = await self._send_command(command, response_timeout)
 
-        # Check that the response is of the expected type
-        assert isinstance(response, hci.HCI_Command_Status_Event)
+        # For unknown HCI commands, some controllers return Command Complete instead of
+        # Command Status.
+        if isinstance(response, hci.HCI_Command_Complete_Event):
+            # Assume the first byte of the return parameters is the status
+            if (
+                status := hci.HCI_ErrorCode(response.parameters[3])
+            ) != hci.HCI_ErrorCode.UNKNOWN_HCI_COMMAND_ERROR:
+                logger.warning(f'unexpected return paramerers status {status}')
+        else:
+            assert isinstance(response, hci.HCI_Command_Status_Event)
+            status = hci.HCI_ErrorCode(response.status)
 
-        # Check the return parameters if required
-        status = response.status
+        # Check the status if required
         if check_status:
             if status != hci.HCI_CommandStatus.PENDING:
-                logger.warning(
-                    f'{command.name} failed ' f'({hci.HCI_Constant.error_name(status)})'
-                )
+                logger.warning(f'{command.name} failed ' f'({status.name})')
                 raise hci.HCI_Error(status)
 
-        return hci.HCI_ErrorCode(status)
+        return status
 
     @utils.deprecated("Use utils.AsyncRunner.spawn() instead.")
     def send_command_sync(self, command: hci.HCI_AsyncCommand) -> None:
@@ -830,7 +858,9 @@ class Host(utils.EventEmitter):
                 data=pdu,
             )
             logger.debug(
-                '>>> ACL packet enqueue: (Handle=0x%04X) %s', connection_handle, pdu
+                '>>> ACL packet enqueue: (handle=0x%04X) %s',
+                connection_handle,
+                pdu.hex(),
             )
             packet_queue.enqueue(acl_packet, connection_handle)
 
@@ -838,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,
             )
@@ -1149,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)
@@ -1553,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}')

bumble/rfcomm.py

@@ -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

bumble/smp.py

@@ -27,7 +27,7 @@ from __future__ import annotations
 import asyncio
 import enum
 import logging
-from collections.abc import Awaitable, Callable
+from collections.abc import Awaitable, Callable, Sequence
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, ClassVar, TypeVar, cast
 
@@ -507,10 +507,15 @@ def smp_auth_req(bonding: bool, mitm: bool, sc: bool, keypress: bool, ct2: bool)
 
 # -----------------------------------------------------------------------------
 class AddressResolver:
-    def __init__(self, resolving_keys):
+    def __init__(self, resolving_keys: Sequence[tuple[bytes, Address]]) -> None:
         self.resolving_keys = resolving_keys
 
-    def resolve(self, address):
+    def can_resolve_to(self, address: Address) -> bool:
+        return any(
+            resolved_address == address for _, resolved_address in self.resolving_keys
+        )
+
+    def resolve(self, address: Address) -> Address | None:
         address_bytes = bytes(address)
         hash_part = address_bytes[0:3]
         prand = address_bytes[3:6]

examples/run_avrcp.py

@@ -25,7 +25,7 @@ import sys
 import websockets.asyncio.server
 
 import bumble.logging
-from bumble import a2dp, avc, avdtp, avrcp, utils
+from bumble import a2dp, avc, avdtp, avrcp, sdp, utils
 from bumble.core import PhysicalTransport
 from bumble.device import Device
 from bumble.transport import open_transport
@@ -34,7 +34,7 @@ logger = logging.getLogger(__name__)
 
 
 # -----------------------------------------------------------------------------
-def sdp_records():
+def sdp_records() -> dict[int, list[sdp.ServiceAttribute]]:
     a2dp_sink_service_record_handle = 0x00010001
     avrcp_controller_service_record_handle = 0x00010002
     avrcp_target_service_record_handle = 0x00010003
@@ -43,17 +43,17 @@ def sdp_records():
         a2dp_sink_service_record_handle: a2dp.make_audio_sink_service_sdp_records(
             a2dp_sink_service_record_handle
         ),
-        avrcp_controller_service_record_handle: avrcp.make_controller_service_sdp_records(
+        avrcp_controller_service_record_handle: avrcp.ControllerServiceSdpRecord(
             avrcp_controller_service_record_handle
-        ),
-        avrcp_target_service_record_handle: avrcp.make_target_service_sdp_records(
-            avrcp_controller_service_record_handle
-        ),
+        ).to_service_attributes(),
+        avrcp_target_service_record_handle: avrcp.TargetServiceSdpRecord(
+            avrcp_target_service_record_handle
+        ).to_service_attributes(),
     }
 
 
 # -----------------------------------------------------------------------------
-def codec_capabilities():
+def codec_capabilities() -> avdtp.MediaCodecCapabilities:
     return avdtp.MediaCodecCapabilities(
         media_type=avdtp.AVDTP_AUDIO_MEDIA_TYPE,
         media_codec_type=a2dp.A2DP_SBC_CODEC_TYPE,
@@ -81,20 +81,22 @@ def codec_capabilities():
 
 
 # -----------------------------------------------------------------------------
-def on_avdtp_connection(server):
+def on_avdtp_connection(server: avdtp.Protocol) -> None:
     # Add a sink endpoint to the server
     sink = server.add_sink(codec_capabilities())
-    sink.on('rtp_packet', on_rtp_packet)
+    sink.on(sink.EVENT_RTP_PACKET, on_rtp_packet)
 
 
 # -----------------------------------------------------------------------------
-def on_rtp_packet(packet):
+def on_rtp_packet(packet: avdtp.MediaPacket) -> None:
     print(f'RTP: {packet}')
 
 
 # -----------------------------------------------------------------------------
-def on_avrcp_start(avrcp_protocol: avrcp.Protocol, websocket_server: WebSocketServer):
-    async def get_supported_events():
+def on_avrcp_start(
+    avrcp_protocol: avrcp.Protocol, websocket_server: WebSocketServer
+) -> None:
+    async def get_supported_events() -> None:
         events = await avrcp_protocol.get_supported_events()
         print("SUPPORTED EVENTS:", events)
         websocket_server.send_message(
@@ -130,14 +132,14 @@ def on_avrcp_start(avrcp_protocol: avrcp.Protocol, websocket_server: WebSocketSe
 
     utils.AsyncRunner.spawn(get_supported_events())
 
-    async def monitor_track_changed():
+    async def monitor_track_changed() -> None:
         async for identifier in avrcp_protocol.monitor_track_changed():
             print("TRACK CHANGED:", identifier.hex())
             websocket_server.send_message(
                 {"type": "track-changed", "params": {"identifier": identifier.hex()}}
             )
 
-    async def monitor_playback_status():
+    async def monitor_playback_status() -> None:
         async for playback_status in avrcp_protocol.monitor_playback_status():
             print("PLAYBACK STATUS CHANGED:", playback_status.name)
             websocket_server.send_message(
@@ -147,7 +149,7 @@ def on_avrcp_start(avrcp_protocol: avrcp.Protocol, websocket_server: WebSocketSe
                 }
             )
 
-    async def monitor_playback_position():
+    async def monitor_playback_position() -> None:
         async for playback_position in avrcp_protocol.monitor_playback_position(
             playback_interval=1
         ):
@@ -159,7 +161,7 @@ def on_avrcp_start(avrcp_protocol: avrcp.Protocol, websocket_server: WebSocketSe
                 }
             )
 
-    async def monitor_player_application_settings():
+    async def monitor_player_application_settings() -> None:
         async for settings in avrcp_protocol.monitor_player_application_settings():
             print("PLAYER APPLICATION SETTINGS:", settings)
             settings_as_dict = [
@@ -173,14 +175,14 @@ def on_avrcp_start(avrcp_protocol: avrcp.Protocol, websocket_server: WebSocketSe
                 }
             )
 
-    async def monitor_available_players():
+    async def monitor_available_players() -> None:
         async for _ in avrcp_protocol.monitor_available_players():
             print("AVAILABLE PLAYERS CHANGED")
             websocket_server.send_message(
                 {"type": "available-players-changed", "params": {}}
             )
 
-    async def monitor_addressed_player():
+    async def monitor_addressed_player() -> None:
         async for player in avrcp_protocol.monitor_addressed_player():
             print("ADDRESSED PLAYER CHANGED")
             websocket_server.send_message(
@@ -195,7 +197,7 @@ def on_avrcp_start(avrcp_protocol: avrcp.Protocol, websocket_server: WebSocketSe
                 }
             )
 
-    async def monitor_uids():
+    async def monitor_uids() -> None:
         async for uid_counter in avrcp_protocol.monitor_uids():
             print("UIDS CHANGED")
             websocket_server.send_message(
@@ -207,7 +209,7 @@ def on_avrcp_start(avrcp_protocol: avrcp.Protocol, websocket_server: WebSocketSe
                 }
             )
 
-    async def monitor_volume():
+    async def monitor_volume() -> None:
         async for volume in avrcp_protocol.monitor_volume():
             print("VOLUME CHANGED:", volume)
             websocket_server.send_message(
@@ -360,7 +362,7 @@ async def main() -> None:
 
         # Create a listener to wait for AVDTP connections
         listener = avdtp.Listener(avdtp.Listener.create_registrar(device))
-        listener.on('connection', on_avdtp_connection)
+        listener.on(listener.EVENT_CONNECTION, on_avdtp_connection)
 
         avrcp_delegate = Delegate()
         avrcp_protocol = avrcp.Protocol(avrcp_delegate)

examples/run_hfp_handsfree.py

@@ -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)
 
 
 # -----------------------------------------------------------------------------

tasks.py

@@ -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.")

tests/avrcp_test.py

@@ -17,6 +17,7 @@
 # -----------------------------------------------------------------------------
 from __future__ import annotations
 
+import asyncio
 import struct
 from collections.abc import Sequence
 
@@ -422,6 +423,47 @@ def test_passthrough_commands():
     assert bytes(parsed) == play_pressed_bytes
 
 
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_find_sdp_records():
+    two_devices = await TwoDevices.create_with_avdtp()
+
+    # Add SDP records to device 1
+    controller_record = avrcp.ControllerServiceSdpRecord(
+        service_record_handle=0x10001,
+        avctp_version=(1, 4),
+        avrcp_version=(1, 6),
+        supported_features=(
+            avrcp.ControllerFeatures.CATEGORY_1
+            | avrcp.ControllerFeatures.SUPPORTS_BROWSING
+        ),
+    )
+    target_record = avrcp.TargetServiceSdpRecord(
+        service_record_handle=0x10002,
+        avctp_version=(1, 4),
+        avrcp_version=(1, 6),
+        supported_features=(
+            avrcp.TargetFeatures.CATEGORY_1 | avrcp.TargetFeatures.SUPPORTS_BROWSING
+        ),
+    )
+
+    two_devices.devices[1].sdp_service_records = {
+        0x10001: controller_record.to_service_attributes(),
+        0x10002: target_record.to_service_attributes(),
+    }
+
+    # Find records from device 0
+    controller_records = await avrcp.ControllerServiceSdpRecord.find(
+        two_devices.connections[0]
+    )
+    assert len(controller_records) == 1
+    assert controller_records[0] == controller_record
+
+    target_records = await avrcp.TargetServiceSdpRecord.find(two_devices.connections[0])
+    assert len(target_records) == 1
+    assert target_records[0] == target_record
+
+
 # -----------------------------------------------------------------------------
 @pytest.mark.asyncio
 async def test_get_supported_events():
@@ -436,6 +478,163 @@ async def test_get_supported_events():
     assert supported_events == [avrcp.EventId.VOLUME_CHANGED]
 
 
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_passthrough_key_event():
+    two_devices = await TwoDevices.create_with_avdtp()
+
+    q = asyncio.Queue[tuple[avc.PassThroughFrame.OperationId, bool, bytes]]()
+
+    class Delegate(avrcp.Delegate):
+        async def on_key_event(
+            self, key: avc.PassThroughFrame.OperationId, pressed: bool, data: bytes
+        ) -> None:
+            q.put_nowait((key, pressed, data))
+
+    two_devices.protocols[1].delegate = Delegate()
+
+    for key, pressed in [
+        (avc.PassThroughFrame.OperationId.PLAY, True),
+        (avc.PassThroughFrame.OperationId.PLAY, False),
+        (avc.PassThroughFrame.OperationId.PAUSE, True),
+        (avc.PassThroughFrame.OperationId.PAUSE, False),
+    ]:
+        await two_devices.protocols[0].send_key_event(key, pressed)
+        assert (await q.get()) == (key, pressed, b'')
+
+
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_passthrough_key_event_rejected():
+    two_devices = await TwoDevices.create_with_avdtp()
+
+    class Delegate(avrcp.Delegate):
+        async def on_key_event(
+            self, key: avc.PassThroughFrame.OperationId, pressed: bool, data: bytes
+        ) -> None:
+            raise avrcp.Delegate.AvcError(avc.ResponseFrame.ResponseCode.REJECTED)
+
+    two_devices.protocols[1].delegate = Delegate()
+
+    response = await two_devices.protocols[0].send_key_event(
+        avc.PassThroughFrame.OperationId.PLAY, True
+    )
+    assert response.response == avc.ResponseFrame.ResponseCode.REJECTED
+
+
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_passthrough_key_event_exception():
+    two_devices = await TwoDevices.create_with_avdtp()
+
+    class Delegate(avrcp.Delegate):
+        async def on_key_event(
+            self, key: avc.PassThroughFrame.OperationId, pressed: bool, data: bytes
+        ) -> None:
+            raise Exception()
+
+    two_devices.protocols[1].delegate = Delegate()
+
+    response = await two_devices.protocols[0].send_key_event(
+        avc.PassThroughFrame.OperationId.PLAY, True
+    )
+    assert response.response == avc.ResponseFrame.ResponseCode.REJECTED
+
+
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_set_volume():
+    two_devices = await TwoDevices.create_with_avdtp()
+
+    for volume in range(avrcp.SetAbsoluteVolumeCommand.MAXIMUM_VOLUME + 1):
+        response = await two_devices.protocols[1].send_avrcp_command(
+            avc.CommandFrame.CommandType.CONTROL, avrcp.SetAbsoluteVolumeCommand(volume)
+        )
+        assert isinstance(response.response, avrcp.SetAbsoluteVolumeResponse)
+        assert response.response.volume == volume
+        assert two_devices.protocols[0].delegate.volume == volume
+
+
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_get_playback_status():
+    two_devices = await TwoDevices.create_with_avdtp()
+
+    for status in avrcp.PlayStatus:
+        two_devices.protocols[0].delegate.playback_status = status
+        response = await two_devices.protocols[1].get_play_status()
+        assert response.play_status == status
+
+
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_get_supported_company_ids():
+    two_devices = await TwoDevices.create_with_avdtp()
+
+    for status in avrcp.PlayStatus:
+        two_devices.protocols[0].delegate = avrcp.Delegate(
+            supported_company_ids=[avrcp.AVRCP_BLUETOOTH_SIG_COMPANY_ID]
+        )
+        supported_company_ids = await two_devices.protocols[
+            1
+        ].get_supported_company_ids()
+        assert supported_company_ids == [avrcp.AVRCP_BLUETOOTH_SIG_COMPANY_ID]
+
+
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_monitor_volume():
+    two_devices = await TwoDevices.create_with_avdtp()
+
+    two_devices.protocols[1].delegate = avrcp.Delegate([avrcp.EventId.VOLUME_CHANGED])
+    volume_iter = two_devices.protocols[0].monitor_volume()
+
+    for volume in range(avrcp.SetAbsoluteVolumeCommand.MAXIMUM_VOLUME + 1):
+        # Interim
+        two_devices.protocols[1].delegate.volume = 0
+        assert (await anext(volume_iter)) == 0
+        # Changed
+        two_devices.protocols[1].notify_volume_changed(volume)
+        assert (await anext(volume_iter)) == volume
+
+
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_monitor_playback_status():
+    two_devices = await TwoDevices.create_with_avdtp()
+
+    two_devices.protocols[1].delegate = avrcp.Delegate(
+        [avrcp.EventId.PLAYBACK_STATUS_CHANGED]
+    )
+    playback_status_iter = two_devices.protocols[0].monitor_playback_status()
+
+    for playback_status in avrcp.PlayStatus:
+        # Interim
+        two_devices.protocols[1].delegate.playback_status = avrcp.PlayStatus.STOPPED
+        assert (await anext(playback_status_iter)) == avrcp.PlayStatus.STOPPED
+        # Changed
+        two_devices.protocols[1].notify_playback_status_changed(playback_status)
+        assert (await anext(playback_status_iter)) == playback_status
+
+
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_monitor_now_playing_content():
+    two_devices = await TwoDevices.create_with_avdtp()
+
+    two_devices.protocols[1].delegate = avrcp.Delegate(
+        [avrcp.EventId.NOW_PLAYING_CONTENT_CHANGED]
+    )
+    now_playing_iter = two_devices.protocols[0].monitor_now_playing_content()
+
+    for _ in range(2):
+        # Interim
+        await anext(now_playing_iter)
+        # Changed
+        two_devices.protocols[1].notify_now_playing_content_changed()
+        await anext(now_playing_iter)
+
+
 # -----------------------------------------------------------------------------
 if __name__ == '__main__':
     test_frame_parser()

tests/hci_test.py

@@ -232,6 +232,14 @@ def test_return_parameters() -> None:
     assert len(params.local_name) == 248
     assert hci.map_null_terminated_utf8_string(params.local_name) == 'hello'
 
+    # Some return parameters may be shorter than the full length
+    # (for Command Complete events with errors)
+    params = hci.HCI_Read_BD_ADDR_Command.parse_return_parameters(
+        bytes.fromhex('010011223344')
+    )
+    assert isinstance(params, hci.HCI_StatusReturnParameters)
+    assert params.status == hci.HCI_ErrorCode.UNKNOWN_HCI_COMMAND_ERROR
+
 
 # -----------------------------------------------------------------------------
 def test_HCI_Command():

tests/host_test.py

@@ -26,11 +26,14 @@ from bumble.controller import Controller
 from bumble.hci import (
     HCI_AclDataPacket,
     HCI_Command_Complete_Event,
+    HCI_Command_Status_Event,
+    HCI_CommandStatus,
     HCI_Disconnect_Command,
     HCI_Error,
     HCI_ErrorCode,
     HCI_Event,
     HCI_GenericReturnParameters,
+    HCI_LE_Terminate_BIG_Command,
     HCI_Reset_Command,
     HCI_StatusReturnParameters,
 )
@@ -229,3 +232,47 @@ async def test_send_sync_command() -> None:
     )
     response3 = await host.send_sync_command_raw(command)  # type: ignore
     assert isinstance(response3.return_parameters, HCI_GenericReturnParameters)
+
+
+@pytest.mark.asyncio
+async def test_send_async_command() -> None:
+    source = Source()
+    sink = Sink(
+        source,
+        HCI_Command_Status_Event(
+            HCI_CommandStatus.PENDING,
+            1,
+            HCI_Reset_Command.op_code,
+        ),
+    )
+
+    host = Host(source, sink)
+    host.ready = True
+
+    # Normal pending status
+    response = await host.send_async_command(
+        HCI_LE_Terminate_BIG_Command(big_handle=0, reason=0)
+    )
+    assert response == HCI_CommandStatus.PENDING
+
+    # Unknown HCI command result returned as a Command Status
+    sink.response = HCI_Command_Status_Event(
+        HCI_ErrorCode.UNKNOWN_HCI_COMMAND_ERROR,
+        1,
+        HCI_LE_Terminate_BIG_Command.op_code,
+    )
+    response = await host.send_async_command(
+        HCI_LE_Terminate_BIG_Command(big_handle=0, reason=0), check_status=False
+    )
+    assert response == HCI_ErrorCode.UNKNOWN_HCI_COMMAND_ERROR
+
+    # Unknown HCI command result returned as a Command Complete
+    sink.response = HCI_Command_Complete_Event(
+        1,
+        HCI_LE_Terminate_BIG_Command.op_code,
+        HCI_StatusReturnParameters(HCI_ErrorCode.UNKNOWN_HCI_COMMAND_ERROR),
+    )
+    response = await host.send_async_command(
+        HCI_LE_Terminate_BIG_Command(big_handle=0, reason=0), check_status=False
+    )
+    assert response == HCI_ErrorCode.UNKNOWN_HCI_COMMAND_ERROR