Merge pull request #932 from zxzxwu/usb-iso-bulk-workaround

fix(usb): support LE ISO data over Bulk endpoints

bumble/host.py

@@ -247,6 +247,7 @@ class Host(utils.EventEmitter):
     bis_links: dict[int, IsoLink]
     sco_links: dict[int, ScoLink]
     bigs: dict[int, set[int]]
+    link_ts_flags: dict[int, int]
     acl_packet_queue: DataPacketQueue | None = None
     le_acl_packet_queue: DataPacketQueue | None = None
     iso_packet_queue: DataPacketQueue | None = None
@@ -269,6 +270,7 @@ class Host(utils.EventEmitter):
         self.bis_links = {}  # BIS links, by connection handle
         self.sco_links = {}  # SCO links, by connection handle
         self.bigs = {}  # BIG Handle to BIS Handles
+        self.link_ts_flags = {}  # TS_Flag for ISO links, by handle
         self.pending_command: hci.HCI_SyncCommand | hci.HCI_AsyncCommand | None = None
         self.pending_response: (
             asyncio.Future[
@@ -486,6 +488,7 @@ class Host(utils.EventEmitter):
                     hci.HCI_LE_PHY_UPDATE_COMPLETE_EVENT,
                     hci.HCI_LE_EXTENDED_ADVERTISING_REPORT_EVENT,
                     hci.HCI_LE_PERIODIC_ADVERTISING_SYNC_ESTABLISHED_EVENT,
+                    hci.HCI_LE_PERIODIC_ADVERTISING_SYNC_ESTABLISHED_V2_EVENT,
                     hci.HCI_LE_PERIODIC_ADVERTISING_REPORT_EVENT,
                     hci.HCI_LE_PERIODIC_ADVERTISING_SYNC_LOST_EVENT,
                     hci.HCI_LE_SCAN_TIMEOUT_EVENT,
@@ -1028,6 +1031,82 @@ class Host(utils.EventEmitter):
         # Look for the connection to which this data belongs
         if connection := self.connections.get(packet.connection_handle):
             connection.on_hci_acl_data_packet(packet)
+            return
+
+        # WORKAROUND: Some controllers (e.g. Intel BE200) send ISO data wrapped in ACL packets
+        # using the CIS handle.
+        is_cis = packet.connection_handle in self.cis_links
+        is_bis = packet.connection_handle in self.bis_links
+
+        if is_cis or is_bis:
+            logger.debug(
+                f"Received ISO data wrapped in ACL packet for handle 0x{packet.connection_handle:04X}"
+            )
+            payload = packet.data
+
+            ts_flag = self.link_ts_flags.get(packet.connection_handle)
+            if ts_flag is None:
+                # Learn TS flag from the first packet on this link
+                if is_bis:
+                    # BIS packets always have Timestamp according to spec
+                    ts_flag = 1
+                elif len(payload) < 8:
+                    # Too short to have 8-byte header (TS), must be No TS
+                    ts_flag = 0
+                else:
+                    psn_no_ts = int.from_bytes(payload[0:2], 'little')
+                    psn_has_ts = int.from_bytes(payload[4:6], 'little')
+                    if psn_has_ts == 0:
+                        ts_flag = 1
+                    elif psn_no_ts == 0:
+                        ts_flag = 0
+                    else:
+                        # Fallback heuristic
+                        ts_flag = 1 if psn_has_ts < psn_no_ts else 0
+                self.link_ts_flags[packet.connection_handle] = ts_flag
+                logger.info(
+                    f"Learned TS_Flag = {ts_flag} for handle 0x{packet.connection_handle:04X}"
+                )
+
+            if ts_flag:
+                header_size = 8
+                sdu_length_offset = 6
+            else:
+                header_size = 4
+                sdu_length_offset = 2
+
+            pb_flag = 0b10
+            if len(payload) >= header_size:
+                sdu_length = int.from_bytes(
+                    payload[sdu_length_offset : sdu_length_offset + 2], 'little'
+                )
+                if sdu_length == len(payload) - header_size:
+                    pb_flag = 0b10  # Complete SDU
+                else:
+                    pb_flag = 0b00  # First fragment
+            else:
+                pb_flag = 0b01  # Continuation
+                ts_flag = 0
+
+            # Reconstruct the raw ISO packet (excluding packet indicator 0x05)
+            pdu_info = packet.connection_handle | (pb_flag << 12) | (ts_flag << 14)
+            header = bytes(
+                [
+                    pdu_info & 0xFF,
+                    (pdu_info >> 8) & 0xFF,
+                    len(payload) & 0xFF,
+                    (len(payload) >> 8) & 0xFF,
+                ]
+            )
+            raw_iso_packet = header + payload
+
+            try:
+                iso_packet = hci.HCI_IsoDataPacket.from_bytes(
+                    bytes([hci.HCI_ISO_DATA_PACKET]) + raw_iso_packet
+                )
+                self.on_hci_iso_data_packet(iso_packet)
+            except Exception as e:
+                logger.warning(f"Failed to reconstruct ISO packet from ACL: {e}")
 
     def on_hci_sco_data_packet(self, packet: hci.HCI_SynchronousDataPacket) -> None:
         # Experimental
@@ -1251,6 +1330,7 @@ class Host(utils.EventEmitter):
             self.emit('disconnection', handle, event.reason)
 
             # Remove the handle reference
+            self.link_ts_flags.pop(handle, None)
             _ = (
                 self.connections.pop(handle, 0)
                 or self.cis_links.pop(handle, 0)
@@ -1371,6 +1451,20 @@ class Host(utils.EventEmitter):
             event.advertiser_clock_accuracy,
         )
 
+    def on_hci_le_periodic_advertising_sync_established_v2_event(
+        self, event: hci.HCI_LE_Periodic_Advertising_Sync_Established_V2_Event
+    ):
+        self.emit(
+            'periodic_advertising_sync_establishment',
+            event.status,
+            event.sync_handle,
+            event.advertising_sid,
+            event.advertiser_address,
+            event.advertiser_phy,
+            event.periodic_advertising_interval,
+            event.advertiser_clock_accuracy,
+        )
+
     def on_hci_le_periodic_advertising_sync_lost_event(
         self, event: hci.HCI_LE_Periodic_Advertising_Sync_Lost_Event
     ):

bumble/transport/pyusb.py

@@ -104,6 +104,9 @@ async def open_pyusb_transport(spec: str) -> Transport:
                         0,
                         packet[1:],
                     )
+                elif packet_type == hci.HCI_ISO_DATA_PACKET:
+                    # Workaround: Send ISO packets over Bulk Out
+                    self.device.write(USB_ENDPOINT_ACL_OUT, packet[1:])
                 else:
                     logger.warning(
                         color(f'unsupported packet type {packet_type}', 'red')

bumble/transport/usb.py

@@ -336,6 +336,25 @@ class UsbPacketSink:
                     )
                     self.isochronous_out_transfer.submit()
                     submitted = True
+                elif packet_type == hci.HCI_ISO_DATA_PACKET:
+                    if self.isochronous_out_transfer is None:
+                        # Workaround: Send ISO packets over Bulk Out when Isochronous endpoints are not enabled
+                        self.bulk_or_control_out_transfer.setBulk(
+                            self.bulk_out.getAddress(),
+                            packet_payload,
+                            callback=self.transfer_callback,
+                        )
+                        self.bulk_or_control_out_transfer.submit()
+                        submitted = True
+                    else:
+                        logger.warning(
+                            color(
+                                'ISO packets over Isochronous endpoints not supported yet',
+                                'red',
+                            )
+                        )
+                        self.out_transfer_ready.release()
+                        continue
                 else:
                     logger.warning(
                         color(f'unsupported packet type {packet_type}', 'red')

tests/usb_test.py

@@ -0,0 +1,95 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import asyncio
+from unittest import mock
+
+import pytest
+
+from bumble import hci
+from bumble.transport import usb
+
+
+@pytest.mark.asyncio
+async def test_usb_packet_sink_iso_routing():
+    # Mock usb1 device and endpoints
+    mock_device = mock.Mock()
+    mock_bulk_out = mock.Mock()
+    mock_bulk_out.getAddress.return_value = 0x02
+
+    # Scenario 1: Isochronous endpoints are not enabled (isochronous_out is None)
+    mock_transfer = mock.Mock()
+    mock_device.getTransfer.return_value = mock_transfer
+
+    sink = usb.UsbPacketSink(mock_device, mock_bulk_out, isochronous_out=None)
+    sink.start()
+
+    # Send HCI_ISO_DATA_PACKET
+    iso_packet = bytes([hci.HCI_ISO_DATA_PACKET, 0x01, 0x02, 0x03])
+    sink.on_packet(iso_packet)
+
+    # Yield control to let the queue processor run
+    await asyncio.sleep(0.01)
+
+    # Verify it was sent via bulk transfer
+    mock_transfer.setBulk.assert_called_once_with(
+        0x02,
+        bytes([0x01, 0x02, 0x03]),
+        callback=sink.transfer_callback,
+    )
+    mock_transfer.submit.assert_called_once()
+
+    if sink.queue_task:
+        sink.queue_task.cancel()
+        try:
+            await sink.queue_task
+        except asyncio.CancelledError:
+            pass
+
+
+@pytest.mark.asyncio
+async def test_usb_packet_sink_iso_routing_with_iso_endpoint():
+    # Mock usb1 device and endpoints
+    mock_device = mock.Mock()
+    mock_bulk_out = mock.Mock()
+    mock_bulk_out.getAddress.return_value = 0x02
+    mock_iso_out = mock.Mock()
+    mock_iso_out.getMaxPacketSize.return_value = 64
+
+    # Scenario 2: Isochronous endpoints are enabled
+    mock_transfer_bulk = mock.Mock()
+    mock_transfer_iso = mock.Mock()
+
+    # getTransfer is called twice: once for bulk_or_control and once for isochronous
+    mock_device.getTransfer.side_effect = [mock_transfer_bulk, mock_transfer_iso]
+
+    sink = usb.UsbPacketSink(mock_device, mock_bulk_out, isochronous_out=mock_iso_out)
+    sink.start()
+
+    # Send HCI_ISO_DATA_PACKET
+    iso_packet = bytes([hci.HCI_ISO_DATA_PACKET, 0x01, 0x02, 0x03])
+    sink.on_packet(iso_packet)
+
+    # Yield control to let the queue processor run
+    await asyncio.sleep(0.01)
+
+    # Verify it was NOT sent via bulk transfer
+    mock_transfer_bulk.setBulk.assert_not_called()
+
+    if sink.queue_task:
+        sink.queue_task.cancel()
+        try:
+            await sink.queue_task
+        except asyncio.CancelledError:
+            pass