usbif: UAC descriptor parsing, the first half of Phase 5's host side

A UAC device does not announce its formats in its interface descriptors.
It announces them in class-specific descriptors interleaved between them,
so a host must walk the whole configuration blob before it can choose an
alternate setting and start streaming. That walk is this module.

It is Python, and in this package rather than the C module, for the
reason the package docstring already states: Python configures and
observes, C moves isochronous bytes. Choosing a format is configuration.
Doing it here means a mis-parsed descriptor costs a re-run instead of a
reflash, and lets the same parser be pointed at a descriptor captured
from any host, including one that is not the board.

Four things it gets right that a naive reader does not:

Sample rates are three-byte little-endian. A four-byte read returns a
plausible-looking wrong number rather than failing, which is the worst
kind of parser bug.

Alt 0 is excluded. By specification it carries no endpoint and exists so
a device can be configured while consuming no bus bandwidth; offering it
as a choosable stream would offer a format that is silent by design.

Feedback endpoints are excluded. They carry rate corrections, not
samples, and treating one as a stream produces a device that enumerates,
"works", and is silent.

A truncated or padded blob stops the walk rather than looping. Real
hardware returns both.

Eleven tests over a blob assembled byte by byte rather than captured, so
they exercise the real layout -- the three-byte rates, the audio
endpoint's 9-byte form, alt 0 present with no endpoint.

Paired with usbif's new host_desc(dev_id), which hands Python a hosted
device's whole active configuration descriptor including every
class-specific descriptor. UVC will need exactly the same door.

Author: bdbarnett

lib/usbif/uac.py

@@ -0,0 +1,208 @@
+"""USB Audio Class descriptor parsing, for the host role.
+
+A UAC device does not announce its formats in its interface descriptors. It
+announces them in *class-specific* descriptors interleaved between them, and a
+host has to walk the whole configuration blob to learn what the device can do
+before it can choose an alternate setting and start streaming. That walk is
+this module.
+
+It lives in Python, and in this package rather than in the C module, for the
+reason the package docstring already gives: **Python configures and observes;
+C moves isochronous bytes.** Choosing a format is configuration. Doing it here
+means a mis-parsed descriptor costs a re-run instead of a reflash, and means
+the same parser can be pointed at a descriptor captured from any host --
+including one from a machine that is not the board.
+
+Only UAC 1.0 is handled, which is what class-compliant devices that work
+without a driver actually implement, and what the ESP32's full-speed host can
+carry. UAC 2.0's descriptors differ enough to be a separate reader rather than
+a flag on this one.
+
+The shape of a configuration blob, for the reader who has not stared at one::
+
+    CONFIGURATION
+      INTERFACE  class=1 subclass=1   AudioControl
+        CS_INTERFACE ...              topology: terminals and units
+      INTERFACE  class=1 subclass=2   AudioStreaming, alt 0  -- always silent
+        (no endpoint: alt 0 exists so a device can consume no bandwidth)
+      INTERFACE  class=1 subclass=2   AudioStreaming, alt 1
+        CS_INTERFACE AS_GENERAL       which terminal, which format tag
+        CS_INTERFACE FORMAT_TYPE      channels, bit depth, sample rates
+        ENDPOINT     isochronous      address, packet size, interval
+      INTERFACE  ... alt 2, alt 3     the same again at other formats
+
+Choosing a format therefore means choosing an *alternate setting*, which is
+why :func:`streams` returns one record per alt rather than one per interface.
+"""
+
+try:
+    from collections import namedtuple
+except ImportError:  # pragma: no cover - ucollections on older firmware
+    from ucollections import namedtuple
+
+# Descriptor types
+DT_INTERFACE = 0x04
+DT_ENDPOINT = 0x05
+DT_CS_INTERFACE = 0x24
+DT_CS_ENDPOINT = 0x25
+
+# Audio interface class and subclasses
+CLASS_AUDIO = 0x01
+SUBCLASS_AUDIOCONTROL = 0x01
+SUBCLASS_AUDIOSTREAMING = 0x02
+
+# AudioStreaming class-specific interface descriptor subtypes
+AS_GENERAL = 0x01
+AS_FORMAT_TYPE = 0x02
+
+# Endpoint attribute bits
+EP_XFER_MASK = 0x03
+EP_XFER_ISOC = 0x01
+EP_SYNC_MASK = 0x0C
+EP_SYNC_ASYNC = 0x04
+EP_SYNC_ADAPTIVE = 0x08
+EP_SYNC_SYNC = 0x0C
+EP_USAGE_MASK = 0x30
+EP_USAGE_FEEDBACK = 0x10
+
+IN, OUT = "in", "out"
+
+STREAM_FIELDS = ("interface", "alt", "endpoint", "direction", "rates",
+                 "channels", "bits", "frame_bytes", "max_packet", "interval",
+                 "sync", "terminal")
+
+UacStream = namedtuple("UacStream", " ".join(STREAM_FIELDS))  # noqa: PYI024
+
+
+def descriptors(blob):
+    """Walk a configuration blob, yielding ``(length, type, memoryview)``.
+
+    Stops at the first zero-length descriptor rather than looping forever: a
+    truncated or padded blob is a real thing to receive from real hardware,
+    and a parser that hangs on one is worse than a parser that stops early.
+    """
+    view = memoryview(blob)
+    offset = 0
+    end = len(view)
+    while offset + 2 <= end:
+        length = view[offset]
+        if length < 2 or offset + length > end:
+            return
+        yield length, view[offset + 1], view[offset:offset + length]
+        offset += length
+
+
+def _rates(body, offset, count):
+    """Sample rates from a FORMAT_TYPE_I descriptor.
+
+    ``bSamFreqType`` is 0 for a continuous min..max pair, or a count of
+    discrete rates. Both are three-byte little-endian, which is the detail
+    that makes a naive 4-byte read return plausible nonsense.
+    """
+    out = []
+    for i in range(count if count else 2):
+        base = offset + i * 3
+        if base + 3 > len(body):
+            break
+        out.append(body[base] | (body[base + 1] << 8) | (body[base + 2] << 16))
+    return tuple(out)
+
+
+def streams(blob):
+    """Every AudioStreaming alternate setting that can actually carry audio.
+
+    Alt 0 is deliberately excluded: by specification it has no endpoint and
+    exists so a device can be configured while consuming no bus bandwidth.
+    Returning it as a choosable stream would offer a format that is silent by
+    design, which is the sort of thing that looks like a driver bug later.
+
+    Feedback endpoints are excluded for the same reason -- they carry rate
+    corrections, not samples.
+    """
+    found = []
+    itf = alt = None
+    pending = None
+
+    for length, dtype, body in descriptors(blob):
+        if dtype == DT_INTERFACE and length >= 9:
+            if pending is not None:
+                found.append(pending)
+                pending = None
+            itf, alt = body[2], body[3]
+            if body[5] == CLASS_AUDIO and body[6] == SUBCLASS_AUDIOSTREAMING and alt != 0:
+                pending = {"interface": itf, "alt": alt, "terminal": None,
+                           "rates": (), "channels": 0, "bits": 0, "frame_bytes": 0,
+                           "endpoint": None, "direction": None,
+                           "max_packet": 0, "interval": 0, "sync": None}
+            continue
+
+        if pending is None:
+            continue
+
+        if dtype == DT_CS_INTERFACE and length >= 3:
+            subtype = body[2]
+            if subtype == AS_GENERAL and length >= 7:
+                pending["terminal"] = body[3]
+            elif subtype == AS_FORMAT_TYPE and length >= 8:
+                pending["channels"] = body[4]
+                pending["frame_bytes"] = body[5]      # bSubframeSize
+                pending["bits"] = body[6]             # bBitResolution
+                pending["rates"] = _rates(body, 8, body[7])
+        elif dtype == DT_ENDPOINT and length >= 7:
+            attrs = body[3]
+            if (attrs & EP_XFER_MASK) != EP_XFER_ISOC:
+                continue
+            if (attrs & EP_USAGE_MASK) == EP_USAGE_FEEDBACK:
+                continue                              # rate corrections, not audio
+            address = body[2]
+            pending["endpoint"] = address
+            pending["direction"] = IN if address & 0x80 else OUT
+            pending["max_packet"] = body[4] | (body[5] << 8)
+            pending["interval"] = body[6]
+            sync = attrs & EP_SYNC_MASK
+            pending["sync"] = {EP_SYNC_ASYNC: "async", EP_SYNC_ADAPTIVE: "adaptive",
+                               EP_SYNC_SYNC: "sync"}.get(sync, "none")
+
+    if pending is not None:
+        found.append(pending)
+
+    return tuple(UacStream(**s) for s in found if s["endpoint"] is not None)
+
+
+def has_audio(blob):
+    """True if this configuration offers any AudioStreaming interface."""
+    for length, dtype, body in descriptors(blob):
+        if dtype == DT_INTERFACE and length >= 9 \
+                and body[5] == CLASS_AUDIO and body[6] == SUBCLASS_AUDIOSTREAMING:
+            return True
+    return False
+
+
+def choose(streams_found, direction, rate=None, channels=None, bits=None):
+    """Pick the best stream for a direction, or ``None``.
+
+    Preference order, most specific first: an exact match on everything the
+    caller asked for, then the highest rate available, then the widest bit
+    depth. A caller that asks for nothing gets the device's best offer, which
+    is what "select it like any other output" should mean.
+    """
+    candidates = [s for s in streams_found if s.direction == direction]
+    if rate is not None:
+        candidates = [s for s in candidates if rate in s.rates]
+    if channels is not None:
+        candidates = [s for s in candidates if s.channels == channels]
+    if bits is not None:
+        candidates = [s for s in candidates if s.bits == bits]
+    if not candidates:
+        return None
+    return max(candidates, key=lambda s: (max(s.rates) if s.rates else 0, s.bits,
+                                          s.channels))
+
+
+def describe(stream):
+    """One-line human description of a stream, for logs and REPL use."""
+    rates = "/".join(str(r) for r in stream.rates) if stream.rates else "?"
+    return "itf {} alt {}: {} {} Hz x {}ch x {}bit, ep {:#04x} {} {}B/{}ms".format(
+        stream.interface, stream.alt, stream.direction, rates, stream.channels,
+        stream.bits, stream.endpoint, stream.sync, stream.max_packet,
+        stream.interval)

pydevices-desktop.toml

@@ -62,6 +62,7 @@
 "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/usbif/linux_usb.py" = "/lib/usbif/linux_usb.py"
 "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/usbif/native_midi.py" = "/lib/usbif/native_midi.py"
 "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/usbif/native_usb.py" = "/lib/usbif/native_usb.py"
+"https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/usbif/uac.py" = "/lib/usbif/uac.py"
 "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/usbif/win_midi.py" = "/lib/usbif/win_midi.py"
 "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/usbif/win_usb.py" = "/lib/usbif/win_usb.py"
 "https://raw.githubusercontent.com/PyDevices/pydevices/main/utils/usdl2.py" = "/lib/usdl2.py"

tests/test_usbif.py

@@ -1014,3 +1014,139 @@ def test_ids_round_trip(self):
         self._install()
         self.assertEqual(self.mod._split_id("dev:midi"), ("dev", None))
         self.assertEqual(self.mod._split_id("host:12"), ("host", 12))
+
+
+def _uac_blob(rates=(48000,), channels=1, bits=16, ep=0x81, attrs=0x05,
+              max_packet=192, extra_alt=None):
+    """A realistic UAC 1.0 configuration blob.
+
+    Assembled byte by byte rather than captured, so the test exercises the
+    real descriptor layout including the parts a naive reader gets wrong: the
+    three-byte sample rates, the audio endpoint's 9-byte form, and alt 0
+    existing with no endpoint at all.
+    """
+    def itf(number, alt, n_eps, subclass):
+        return bytes([9, 0x04, number, alt, n_eps, 0x01, subclass, 0x00, 0x00])
+
+    def rate_bytes(value):
+        return bytes([value & 0xFF, (value >> 8) & 0xFF, (value >> 16) & 0xFF])
+
+    parts = [bytes([9, 0x02, 0, 0, 2, 1, 0, 0x80, 50])]        # CONFIGURATION
+    parts.append(itf(0, 0, 0, 0x01))                            # AudioControl
+    parts.append(bytes([9, 0x24, 0x01, 0x00, 0x01, 9, 0, 1, 1]))  # CS header
+    parts.append(itf(1, 0, 0, 0x02))                            # AS alt 0: silent
+    parts.append(itf(1, 1, 1, 0x02))                            # AS alt 1
+    parts.append(bytes([7, 0x24, AS_GENERAL_SUBTYPE, 0x02, 1, 0x01, 0x00]))
+    fmt = bytes([8 + 3 * len(rates), 0x24, 0x02, 0x01, channels,
+                 bits // 8, bits, len(rates)])
+    for value in rates:
+        fmt += rate_bytes(value)
+    parts.append(fmt)
+    parts.append(bytes([9, 0x05, ep, attrs, max_packet & 0xFF,
+                        (max_packet >> 8) & 0xFF, 0x01, 0x00, 0x00]))
+    if extra_alt:
+        parts.extend(extra_alt)
+    blob = b"".join(parts)
+    return blob[:2] + bytes([len(blob) & 0xFF, (len(blob) >> 8) & 0xFF]) + blob[4:]
+
+
+AS_GENERAL_SUBTYPE = 0x01
+
+
+class TestUacDescriptorParsing(unittest.TestCase):
+    """UAC 1.0 descriptor reading -- how a host learns what a device can do.
+
+    A UAC device announces its formats in class-specific descriptors
+    interleaved between its interfaces, so choosing a format means walking the
+    whole configuration blob and then choosing an alternate setting. These
+    assertions cover the parts that fail quietly rather than loudly.
+    """
+
+    def test_a_simple_microphone_parses(self):
+        from usbif import uac
+
+        found = uac.streams(_uac_blob())
+        self.assertEqual(len(found), 1)
+        stream = found[0]
+        self.assertEqual(stream.direction, uac.IN)
+        self.assertEqual(stream.rates, (48000,))
+        self.assertEqual(stream.channels, 1)
+        self.assertEqual(stream.bits, 16)
+        self.assertEqual(stream.endpoint, 0x81)
+        self.assertEqual(stream.max_packet, 192)
+        self.assertEqual(stream.sync, "async")
+
+    def test_sample_rates_are_three_byte_little_endian(self):
+        # A four-byte read here returns a plausible-looking wrong number
+        # rather than failing, which is the worst kind of parser bug.
+        from usbif import uac
+
+        (stream,) = uac.streams(_uac_blob(rates=(44100, 48000, 96000)))
+        self.assertEqual(stream.rates, (44100, 48000, 96000))
+
+    def test_alt_zero_is_not_offered_as_a_stream(self):
+        # By specification alt 0 has no endpoint and exists so a device can be
+        # configured while using no bandwidth. Offering it would be offering a
+        # format that is silent by design.
+        from usbif import uac
+
+        for stream in uac.streams(_uac_blob()):
+            self.assertNotEqual(stream.alt, 0)
+
+    def test_a_feedback_endpoint_is_not_mistaken_for_audio(self):
+        # Feedback endpoints carry rate corrections, not samples. Treating one
+        # as a stream would produce a device that "works" and is silent.
+        from usbif import uac
+
+        found = uac.streams(_uac_blob(ep=0x82, attrs=0x11))  # isoc | feedback
+        self.assertEqual(found, ())
+
+    def test_a_bulk_endpoint_is_ignored(self):
+        from usbif import uac
+
+        self.assertEqual(uac.streams(_uac_blob(attrs=0x02)), ())
+
+    def test_an_output_endpoint_is_reported_as_out(self):
+        from usbif import uac
+
+        (stream,) = uac.streams(_uac_blob(ep=0x02))
+        self.assertEqual(stream.direction, uac.OUT)
+
+    def test_has_audio_detects_a_streaming_interface(self):
+        from usbif import uac
+
+        self.assertTrue(uac.has_audio(_uac_blob()))
+        self.assertFalse(uac.has_audio(bytes([9, 0x02, 9, 0, 0, 1, 0, 0x80, 50])))
+
+    def test_choose_prefers_the_best_offer_when_asked_for_nothing(self):
+        from usbif import uac
+
+        found = uac.streams(_uac_blob(rates=(8000, 48000)))
+        picked = uac.choose(found, uac.IN)
+        self.assertEqual(max(picked.rates), 48000)
+
+    def test_choose_filters_on_what_the_caller_asked_for(self):
+        from usbif import uac
+
+        found = uac.streams(_uac_blob(rates=(44100, 48000)))
+        self.assertIsNotNone(uac.choose(found, uac.IN, rate=44100))
+        self.assertIsNone(uac.choose(found, uac.IN, rate=192000))
+        self.assertIsNone(uac.choose(found, uac.OUT))
+
+    def test_a_truncated_blob_stops_rather_than_looping(self):
+        # Real hardware returns short and padded descriptors. A parser that
+        # hangs on one is worse than a parser that stops early.
+        from usbif import uac
+
+        blob = _uac_blob()
+        self.assertEqual(uac.streams(blob[:len(blob) // 2]), ())
+        self.assertEqual(list(uac.descriptors(b"\x00\x00")), [])
+        self.assertEqual(list(uac.descriptors(b"")), [])
+
+    def test_describe_names_the_essentials(self):
+        from usbif import uac
+
+        (stream,) = uac.streams(_uac_blob())
+        text = uac.describe(stream)
+        for fragment in ("48000", "1ch", "16bit", "in"):
+            self.assertIn(fragment, text)