diff --git a/README.md b/README.md index 0f050b3..c471c22 100644 --- a/README.md +++ b/README.md @@ -217,6 +217,35 @@ mpy-cross sub-make, which then plants its own qstr fragments in your build directory and breaks the link. Use `idf.py -B` for an out-of-tree build instead, which does not inherit the variable. +## Examples + +| File | Role | What it shows | +|---|---|---| +| [`soundcard.py`](examples/soundcard.py) | device | Class-compliant UAC sound card (C pump). Pair with `usb_speaker.py` | +| [`uac_pump.py`](examples/uac_pump.py) | device | Python FIFO pump -- inspectable path, not the shipping card | +| [`hid_keyboard.py`](examples/hid_keyboard.py) | device | Board types into the host | +| [`hid_mouse.py`](examples/hid_mouse.py) | device | Board moves the host cursor | +| [`ram_drive.py`](examples/ram_drive.py) | device | RAM disk via `msc_attach` | +| [`sd_drive.py`](examples/sd_drive.py) | device | SD card as a USB drive via `msc_attach_blockdev` | +| [`usbif_webcam.py`](examples/usbif_webcam.py) | device | Board is a UVC webcam (`cameraif` when present) | +| [`midi_harmonizer.py`](examples/midi_harmonizer.py) | device | MIDI effect: melody in, triads out | +| [`midi_harmonizer_ui.py`](examples/midi_harmonizer_ui.py) | device | Harmonizer with a touchscreen chord picker | +| [`midi_device_in.py`](examples/midi_device_in.py) | device | Prove the board receives MIDI from a host | +| [`midi_latency.py`](examples/midi_latency.py) | device | MIDI-to-audio round-trip timing | +| [`host_enum.py`](examples/host_enum.py) | host | Attach/detach via the portable API (board or desktop) | +| [`hid_host.py`](examples/hid_host.py) | host | USB keyboard → PyDevices key events (M1) | +| [`usb_serial.py`](examples/usb_serial.py) | host | CDC read/write to a USB-serial device | +| [`usb_speaker.py`](examples/usb_speaker.py) | host | Play through a hosted USB speaker / `soundcard.py` | +| [`usb_mic.py`](examples/usb_mic.py) | host | Capture from a hosted USB microphone | +| [`usb_drive_mount.py`](examples/usb_drive_mount.py) | host | Mount a flash drive and list files | +| [`usb_drive_log.py`](examples/usb_drive_log.py) | host | Append sensor lines to a hosted stick | +| [`midi_host.py`](examples/midi_host.py) | host | Host a MIDI keyboard; send a chord back | +| [`uvc_display.py`](examples/uvc_display.py) | host | Hosted webcam on the board's panel (MJPEG via `jpegio` when present) | +| [`costume_selftest.py`](examples/costume_selftest.py) | — | Validate every costume's descriptors without a host | + +Board-to-board pairings, VBUS warnings, and firmware holes (by issue number) +live in [`examples/README.md`](examples/README.md). + ## Tests The ring buffer is tested on the host, where a failure is a two-second answer diff --git a/examples/README.md b/examples/README.md new file mode 100644 index 0000000..165f52a --- /dev/null +++ b/examples/README.md @@ -0,0 +1,69 @@ +# usbif examples + +One script per use of the shipping API. Each role is class-compliant on its +own: a PC or a commercial peripheral is a valid other end. Board-to-board +pairing is the same scripts on two boards, not a special `pair_*` file. + +## Class × role + +| Class | Device (board presents as…) | Host (board drives…) | +|---|---|---| +| **UAC** | [`soundcard.py`](soundcard.py) (C pump); [`uac_pump.py`](uac_pump.py) (Python FIFO) | [`usb_speaker.py`](usb_speaker.py), [`usb_mic.py`](usb_mic.py) | +| **UVC** | [`usbif_webcam.py`](usbif_webcam.py) | [`uvc_display.py`](uvc_display.py) (MJPEG via `jpegio` when present) | +| **MIDI** | [`midi_harmonizer.py`](midi_harmonizer.py), [`midi_harmonizer_ui.py`](midi_harmonizer_ui.py), [`midi_device_in.py`](midi_device_in.py), [`midi_latency.py`](midi_latency.py) | [`midi_host.py`](midi_host.py) | +| **HID** | [`hid_keyboard.py`](hid_keyboard.py), [`hid_mouse.py`](hid_mouse.py) | [`hid_host.py`](hid_host.py) | +| **MSC** | [`sd_drive.py`](sd_drive.py), [`ram_drive.py`](ram_drive.py) | [`usb_drive_mount.py`](usb_drive_mount.py), [`usb_drive_log.py`](usb_drive_log.py) | +| **CDC** | built-in MicroPython console / costume bit | [`usb_serial.py`](usb_serial.py) | +| **enum** | — | [`host_enum.py`](host_enum.py) (portable API; board or desktop) | +| **self-test** | [`costume_selftest.py`](costume_selftest.py) | — | + +## Board-to-board pairings + +Roles follow the hardware each board has, not their size. **P4 is always the +device** in these pairings: its high-speed host detects no device +([usbif#3](https://github.com/PyDevices/usbif/issues/3)). + +| Device board | Script | Host board | Script | What is lent | +|---|---|---|---|---| +| **P4** | `soundcard.py` | **S3** | `usb_speaker.py` | Sound output -- the headline offload | +| **P4** | `usbif_webcam.py` | **S3** | `uvc_display.py` | Camera → panel | +| **P4** | `sd_drive.py` | **S3** | `usb_drive_mount.py` | Shared storage | +| **P4** / **S3** | `midi_harmonizer.py` | **S3** | `midi_host.py` | MIDI effect / instrument | +| **S3** | `hid_keyboard.py` | **S3** | `hid_host.py` | Control surface (needs two S3s, or S3-device + PC) | + +A PC can replace either end. That is the point of standard classes. + +### Headline: P4 as a USB sound card for an S3 + +1. Console on each board's **UART bridge**, not on the OTG port under test. +2. P4: `mpftp run examples/soundcard.py` -- enumerates as Speakers, C pump + into the ES8311. +3. S3: OTG adapter or powered hub on the host port (the Waveshare S3 touch + boards do not switch VBUS; see also [usbif#5](https://github.com/PyDevices/usbif/issues/5)). +4. S3: `mpftp run examples/usb_speaker.py` -- finds the P4, plays a 440 Hz tone. +5. Hear it on the P4's speaker. + +### Power + +Connecting two self-powered boards can **back-feed** VBUS and leave a board +silent and unflashable. Default link: a cable that omits VBUS, or a powered +hub between them. Unplug both USB cables and use UART alone to recover a +wedged board. + +## Firmware holes (do not invent an example) + +| Hole | Issue | Consequence for examples | +|---|---|---| +| Device UAC is speaker-only (no mic endpoint) | [usbif#7](https://github.com/PyDevices/usbif/issues/7) | `usb_mic.py` hosts a commercial mic; no device-side capture script | +| Portable `Device.functions()` omits video | [usbif#8](https://github.com/PyDevices/usbif/issues/8) | `usbif_webcam.py` pokes `_usbif.FN_VIDEO` directly | +| P4 high-speed host detects nothing | [usbif#3](https://github.com/PyDevices/usbif/issues/3) | Every pairing puts the P4 in device role | +| Host FIFO bias is build-time | [usbif#2](https://github.com/PyDevices/usbif/issues/2) | Stereo UAC host and UVC host compete; Bias-IN may cost stereo | + +## Conventions + +- Headers name the use, the board role, the other end, and whether UART must + hold the REPL (any costume that drops CDC cuts a native-USB session). +- Prefer `usbif.auto` / portable names where they cover the class; fall back + to `_usbif` for streaming surfaces the portable API does not yet wrap. +- No measured fps or latency numbers are claimed in headers for scripts that + have not been re-run in this pass -- cite prior evidence or stay quiet. diff --git a/examples/hid_host.py b/examples/hid_host.py new file mode 100644 index 0000000..a8acf58 --- /dev/null +++ b/examples/hid_host.py @@ -0,0 +1,72 @@ +"""A USB keyboard on the host port drives ordinary PyDevices key events. + +Milestone M1: the same application code, producing the same ``events.Key`` +records, as an SDL keyboard on the desktop. The host stack delivers raw boot +reports; ``usbif.hid_keyboard.KeyboardDecoder`` turns them into presses and +releases by diffing successive reports. + + mpftp run -d COM49 examples/hid_host.py + +**Pairing.** A commercial keyboard, or a PyDevices board running +``hid_keyboard.py``. Needs an S3 (or any board whose host mode works); P4 +host is blocked (usbif#3). + +**Rollover.** When more keys are held than the report can carry, the decoder +ignores the ErrorRollOver report rather than emitting garbage -- see the +module docstring in ``lib/usbif/hid_keyboard.py``. +""" + +import time + +import _usbif +import events +from usbif.hid_keyboard import KeyboardDecoder + + +def find_keyboard(timeout_ms=15000): + _usbif.host_start(("hid",)) + deadline = time.ticks_add(time.ticks_ms(), timeout_ms) + while time.ticks_diff(deadline, time.ticks_ms()) > 0: + for dev in _usbif.host_devices(): + if "hid" in dev[5]: + return dev[0] + time.sleep_ms(250) + return None + + +def main(seconds=30): + dev_id = find_keyboard() + if dev_id is None: + print("no HID device found") + _usbif.host_stop() + return + + print("HID device", dev_id, "-- type on it for %d s" % seconds) + _usbif.host_hid_open(dev_id) + decoder = KeyboardDecoder() + buf = bytearray(8) + t0 = time.ticks_ms() + try: + while time.ticks_diff(time.ticks_ms(), t0) < seconds * 1000: + n = _usbif.host_hid_read(buf) + if n >= 3: + for ev in decoder.feed(buf): + kind = "DOWN" if ev.type == events.KEYDOWN else "UP" + # Field names follow events.Key; fall back to repr if a + # firmware builds the record differently. + name = getattr(ev, "name", "?") + key = getattr(ev, "key", None) or 0 + mod = getattr(ev, "mod", 0) + scan = getattr(ev, "scancode", None) + print(" %s %-12s key=0x%02x mod=0x%02x scancode=%s" + % (kind, name, key, mod, scan)) + else: + time.sleep_ms(5) + finally: + _usbif.host_hid_close() + _usbif.host_stop() + print("done") + + +if __name__ == "__main__": + main() diff --git a/examples/hid_keyboard.py b/examples/hid_keyboard.py new file mode 100644 index 0000000..faf565c --- /dev/null +++ b/examples/hid_keyboard.py @@ -0,0 +1,83 @@ +"""The board is a USB keyboard: type a short string into the host. + +Presents as a boot keyboard (report ID 1) and sends HID reports that any +ordinary OS keyboard stack accepts -- Notepad, a browser address bar, a +terminal. No driver to install. + + mpftp run -d COM49 examples/hid_keyboard.py + +**Console.** Costume is CDC+HID so the REPL stays on the same cable. Run the +REPL on the UART bridge if you change the costume to HID alone. + +**LEDs.** After typing, the script prints ``hid_leds()`` -- the lock-key state +the host last set (caps / num / scroll). A control surface that wants to show +them reads the same call. + +**Pairing.** An S3 hosting HID (``hid_host.py``) can take this board as its +keyboard. A PC works the same way. P4 cannot be the host (usbif#3). +""" + +import time + +import _usbif + +# HID usage ids for a-z (0x04..) and a few extras. Boot protocol only. +_ALPHA = {c: 0x04 + i for i, c in enumerate("abcdefghijklmnopqrstuvwxyz")} +_EXTRA = { + " ": 0x2C, + "\n": 0x28, + ".": 0x37, + ",": 0x36, + "-": 0x2D, +} +MESSAGE = "hello from usbif\n" + + +def _send(report, retries=50): + """Submit one keyboard report; retry while the host has not polled.""" + for _ in range(retries): + if _usbif.hid_send(_usbif.HID_KEYBOARD, report): + return True + time.sleep_ms(2) + return False + + +def _tap(usage, modifier=0): + """Press then release one key.""" + down = bytes([modifier, 0, usage, 0, 0, 0, 0, 0]) + up = bytes(8) + if not _send(down): + return False + time.sleep_ms(30) + return _send(up) + + +def type_string(text): + for ch in text: + lower = ch.lower() + usage = _ALPHA.get(lower) or _EXTRA.get(ch) + if usage is None: + print("skip unsupported char %r" % ch) + continue + mod = 0x02 if ch.isalpha() and ch.isupper() else 0 # left shift + if not _tap(usage, mod): + print("host did not accept a report; is the keyboard mounted?") + return False + time.sleep_ms(20) + return True + + +def main(): + _usbif.dev_functions(_usbif.FN_CDC | _usbif.FN_HID) + print("costume: cdc+hid -- focus a text field on the host") + # Give the host time to configure us before the first report. + time.sleep_ms(1500) + + ok = type_string(MESSAGE) + leds = _usbif.hid_leds() + print("typed %r: %s" % (MESSAGE.strip(), "ok" if ok else "failed")) + print("hid_leds: 0x%02x (bit0=num bit1=caps bit2=scroll)" % leds) + + +if __name__ == "__main__": + main() diff --git a/examples/hid_mouse.py b/examples/hid_mouse.py new file mode 100644 index 0000000..55a21f7 --- /dev/null +++ b/examples/hid_mouse.py @@ -0,0 +1,62 @@ +"""The board is a USB mouse: nudge the host cursor, then put it back. + +Presents as a boot mouse (report ID 2) and sends relative motion reports. +Verified originally by watching Windows' cursor move 82 pixels and return -- +the host is the instrument, not a counter in this script. + + mpftp run -d COM49 examples/hid_mouse.py + +**Report layout.** Boot mouse behind a report ID: buttons, dx, dy, wheel -- +four signed bytes after the ID the C side prepends via ``hid_send``. + +**Pairing.** Same as ``hid_keyboard.py``: a PC, or an S3 running a HID host. +""" + +import time + +import _usbif + +STEPS = 20 +DELTA = 4 # pixels per report + + +def _send(report, retries=50): + for _ in range(retries): + if _usbif.hid_send(_usbif.HID_MOUSE, report): + return True + time.sleep_ms(2) + return False + + +def move(dx, dy): + # buttons=0, dx, dy, wheel=0. Values are signed 8-bit. + def s8(n): + return n & 0xFF + + return _send(bytes([0, s8(dx), s8(dy), 0])) + + +def main(): + _usbif.dev_functions(_usbif.FN_CDC | _usbif.FN_HID) + print("costume: cdc+hid -- watch the host cursor") + time.sleep_ms(1500) + + print("moving +x") + for _ in range(STEPS): + if not move(DELTA, 0): + print("host did not accept a report") + return + time.sleep_ms(20) + + print("moving -x (return)") + for _ in range(STEPS): + if not move(-DELTA, 0): + print("host did not accept a report") + return + time.sleep_ms(20) + + print("done; cursor should be back where it started") + + +if __name__ == "__main__": + main() diff --git a/examples/host_enum.py b/examples/host_enum.py new file mode 100644 index 0000000..ed198fa --- /dev/null +++ b/examples/host_enum.py @@ -0,0 +1,62 @@ +"""Enumerate USB devices and watch attach / detach. + +Uses the portable host API (``usbif.auto.host``), so the same script runs on +a board with the native module and on a desktop where the OS owns the bus. +Capabilities are discovered, never assumed -- an empty set is a valid answer. + + mpftp run -d COM49 examples/host_enum.py + python examples/host_enum.py # desktop + +Plug and unplug devices while it runs; each attach and detach is printed. +On an S3 host that means a powered hub or OTG adapter (no VBUS switching on +the Waveshare touch boards). P4 high-speed host is blocked (usbif#3). +""" + +import sys +import time + +import events +import usbif +from usbif import auto + + +def main(seconds=60): + host = auto.host() + caps = host.capabilities() + print("backend capabilities:", sorted(caps) if caps else "(none)") + host.start() + + print("currently attached:") + for info in host.devices(): + print(" ", usbif.describe(info)) + if not host.devices(): + print(" (none yet -- plug something in)") + + print("watching attach/detach for %d s ..." % seconds) + deadline = time.ticks_add(time.ticks_ms(), seconds * 1000) if hasattr(time, "ticks_ms") \ + else None + end = time.time() + seconds + try: + while True: + if deadline is not None: + if time.ticks_diff(deadline, time.ticks_ms()) <= 0: + break + elif time.time() >= end: + break + for event in host.poll(): + kind = "attach" if event.type == events.USBATTACH else "detach" + if event.type not in (events.USBATTACH, events.USBDETACH): + kind = str(event.type) + print("%s: %s" % (kind, usbif.describe(event.device))) + if host.overflowed: + print("warning: event buffer overflowed -- poll more often") + time.sleep(0.1) + except KeyboardInterrupt: + print("stopped") + finally: + host.stop() + + +if __name__ == "__main__": + secs = int(sys.argv[1]) if len(sys.argv) > 1 else 60 + main(secs) diff --git a/examples/midi_host.py b/examples/midi_host.py index 86d40f8..400fbc9 100644 --- a/examples/midi_host.py +++ b/examples/midi_host.py @@ -11,11 +11,13 @@ that parses or generates MIDI does not care which end it is on -- the USB-MIDI 32-bit packet framing lives in C and never reaches Python. -**Status: written against the driver, not yet run against hardware.** The -driver (`src/usbif_host_midi.c`) is new and compile-verified only; nobody -upstream ships a MIDI host driver for the IDF, so it is ours. Expect to -find things. What it does when it meets a real instrument is exactly what -wants recording in docs/phase0-findings.md. +**Proven on hardware.** A Donner keyboard delivered 989 channel messages +(notes with velocity, pitch bend across its range, CCs, channel-10 drums) +with zero bytes dropped. Host MIDI OUT was closed on a DIN loopback through +an M-Audio interface: all eight sent messages returned byte-exact, median +5 ms round trip. Nobody upstream ships an IDF MIDI host driver, so this one +is ours (`src/usbif_host_midi.c`); the numbers above are why the status line +no longer says "compile-verified only". """ import time diff --git a/examples/ram_drive.py b/examples/ram_drive.py new file mode 100644 index 0000000..e561921 --- /dev/null +++ b/examples/ram_drive.py @@ -0,0 +1,52 @@ +"""A RAM disk the host sees as a removable drive. + +``msc_attach`` serves a buffer the application supplies -- no SD card, no +filesystem on the board, just bytes the host can read and write. Contrast +with ``sd_drive.py``, which hands the host a real block device. + + mpftp run -d COM49 examples/ram_drive.py + +**What you get.** A small writable volume (default 64 KiB). Windows / macOS / +Linux will want to format it the first time; that is expected for a blank +buffer. After format, files copy on and off like any thumb drive. + +**One writer.** While the host has the drive, do not also mount the same +buffer locally. ``msc_status()`` reports eject; honour it before touching the +bytes yourself. + +**Console.** CDC+MSC so the REPL stays on the cable. +""" + +import time + +import _usbif + +# 128 x 512-byte blocks = 64 KiB. Small enough for SRAM on an S3, large enough +# that a host will format and mount it without complaining about capacity. +BLOCKS = 128 +BLOCK = 512 + + +def main(): + _usbif.msc_detach() + + buf = bytearray(BLOCKS * BLOCK) + _usbif.msc_attach(buf, True) + + _usbif.dev_functions(_usbif.FN_CDC | _usbif.FN_MSC) + attached, n_blocks, _ = _usbif.msc_status() + print("RAM disk attached:", attached, "blocks:", n_blocks, + "(%d KiB)" % (n_blocks * BLOCK // 1024)) + print("format it on the host the first time, then copy files") + + while True: + time.sleep_ms(5) + _, _, ejected = _usbif.msc_status() + if ejected: + print("host ejected; releasing") + _usbif.msc_detach() + return + + +if __name__ == "__main__": + main() diff --git a/examples/soundcard.py b/examples/soundcard.py new file mode 100644 index 0000000..6380f62 --- /dev/null +++ b/examples/soundcard.py @@ -0,0 +1,141 @@ +"""The board is a class-compliant USB sound card -- the shipping path. + +A PC (or another PyDevices board hosting UAC) sees Speakers (Espressif Device) +and plays through it. Audio moves from the isochronous endpoint to the board's +I2S codec in a C FreeRTOS task: Python configures and observes, C moves the +bytes. That is the vision's rule, and this is the example that follows it. + + mpftp run -d COM49 examples/soundcard.py + +**Pairing.** On a P4 this is the device half of the sound-card offload: an S3 +without its own codec runs ``usb_speaker.py`` as host and plays through this +board. A laptop works identically -- standard classes are the protocol. + +**Not this file.** ``uac_pump.py`` is the Python FIFO pump -- useful for +watching the buffer and for latency tools that need ``uac_read``. Do not use +it as the sound card; use this. + +**Console.** Costume is CDC+UAC so the REPL stays on the same cable. UART is +still the safer place for the REPL when iterating (a costume change that drops +CDC cuts a native-USB session mid-run). + +**Pins and the amp.** I2S pin numbers and codec bring-up come from +``board_peripherals``: the pump owns the I2S channel, not the ES8311. Without +enabling the speaker amp you get perfect byte counters and silence -- the +failure mode that ate an afternoon in Phase 4. +""" + +import time + +import board_peripherals as bp +import _usbif + +# Host advertises 48 kHz stereo; the board codec is typically 24 kHz mono. +# The C pump decimates. Match the board's own rate so pitch is right. +DEFAULT_VOLUME = 85 # digital gain on this hardware; see phase0 findings + + +def _i2s_pins(): + """(bclk, ws, dout, mclk) from board_peripherals, or raise usefully.""" + # Boards publish these under a few historical names; try them in order. + for names in ( + ("I2S_BCLK", "I2S_WS", "I2S_DOUT", "I2S_MCLK"), + ("bclk", "ws", "dout", "mclk"), + ("BCLK", "WS", "DOUT", "MCLK"), + ): + vals = [] + ok = True + for n in names: + if not hasattr(bp, n): + ok = False + break + vals.append(getattr(bp, n)) + if ok: + return tuple(vals) + pins = getattr(bp, "I2S_OUT_PINS", None) + if pins is not None and len(pins) >= 3: + bclk, ws, dout = pins[0], pins[1], pins[2] + mclk = pins[3] if len(pins) > 3 else -1 + return bclk, ws, dout, mclk + raise RuntimeError( + "board_peripherals does not publish I2S output pins " + "(looked for I2S_BCLK/I2S_WS/I2S_DOUT/I2S_MCLK and I2S_OUT_PINS). " + "Pass them to _usbif.uac_pump_start yourself, or extend the board " + "package." + ) + + +def _bring_up_codec(): + """Power the amp and set a sane volume before the pump starts.""" + # Prefer the board helper that wires power + volume correctly. + audio_out = getattr(bp, "audio_out", None) + if callable(audio_out): + out = audio_out() + try: + out.open() + except Exception: + pass + try: + out.set_volume(DEFAULT_VOLUME) + out.mute(False) + except Exception: + pass + return out + # Fall back: poke the codec through the private hooks uac_pump.py uses. + power = getattr(bp, "_output_power", None) + if callable(power): + try: + power(True) + except Exception: + pass + set_vol = getattr(bp, "_codec_call", None) + if callable(set_vol): + try: + set_vol("set_dac_volume", DEFAULT_VOLUME) + set_vol("dac_mute", False) + except Exception: + pass + return None + + +def main(): + bclk, ws, dout, mclk = _i2s_pins() + fmt = getattr(bp, "_FORMAT", None) + rate = getattr(fmt, "rate", 24000) if fmt is not None else 24000 + bits = getattr(fmt, "bits", 16) if fmt is not None else 16 + channels = getattr(fmt, "channels", 1) if fmt is not None else 1 + + out = _bring_up_codec() + + # Costume first so the host sees the sound card before we start the pump. + # CDC stays so the REPL survives on the same connector. + _usbif.dev_functions(_usbif.FN_CDC | _usbif.FN_AUDIO) + print("costume: cdc+uac -- look for Speakers on the host") + + kwargs = {"rate": rate, "bits": bits, "channels": channels} + if mclk is not None and mclk >= 0: + kwargs["mclk"] = mclk + _usbif.uac_pump_start(bclk, ws, dout, **kwargs) + print("C pump started: I2S bclk=%d ws=%d dout=%d rate=%d ch=%d" + % (bclk, ws, dout, rate, channels)) + print("play audio to this board from a PC, or from usb_speaker.py on an S3") + + try: + while True: + time.sleep_ms(1000) + running, moved, idle, timeouts, shed = _usbif.uac_pump_stats() + print("pump running=%s bytes=%d idle=%d timeouts=%d shed=%d" + % (running, moved, idle, timeouts, shed)) + except KeyboardInterrupt: + print("stopping") + finally: + _usbif.uac_pump_stop() + if out is not None: + try: + out.close() + except Exception: + pass + + +if __name__ == "__main__": + main() diff --git a/examples/uac_pump.py b/examples/uac_pump.py index cc6362e..5388e08 100644 --- a/examples/uac_pump.py +++ b/examples/uac_pump.py @@ -1,17 +1,19 @@ -# usbif: play what the host sends over USB Audio out of the board's codec. +# usbif: Python FIFO pump -- the inspectable path from USB Audio to the codec. # -# The isochronous endpoint is serviced in C on TinyUSB's task; this loop only -# moves already-buffered blocks from that FIFO to the I2S sink, a soft deadline -# set by FIFO depth rather than a per-frame one. Measured at 96-99% of the -# offered stream on an ESP32-P4, which is why the pump has not yet been moved -# into C -- it turned out not to be the bottleneck. +# This is NOT the shipping sound card. The production path is the C pump in +# `soundcard.py` (`_usbif.uac_pump_start`), which moves isochronous bytes +# without the interpreter. Keep this script for two jobs it still does better +# than the C pump: watching the FIFO fill from Python (latency tooling such as +# `midi_latency.py` also depends on the FIFO being readable), and proving the +# path end to end when you need every byte count in a log file. # -# What *was* the bottleneck: TinyUSB's example sizing for the software FIFO is -# a multiple of the endpoint packet, and at 24 kHz mono that packet is 8 bytes, -# giving a 256-byte FIFO -- about 5 ms. Reads averaged 19 bytes, a third of the -# stream was lost, and a consumer asking for 20 ms blocks got nothing at all -# because 20 ms never fit. usbif sizes it in milliseconds instead; see -# USBIF_AUDIO_FIFO_MS in src/usbif_tusb_ext.h. +# History, kept so the numbers make sense: TinyUSB's example FIFO sizing is a +# multiple of the endpoint packet; at 24 kHz mono that was ~5 ms and a third +# of the stream was lost under a Python consumer. usbif sizes the FIFO in +# milliseconds instead. Measured at 96-99% of the offered stream on an +# ESP32-P4 with this loop -- which is why the first pass left the pump in +# Python. The C pump later took over for the flagship; this file stayed as +# the transparent half. import time import _usbif diff --git a/examples/usb_mic.py b/examples/usb_mic.py new file mode 100644 index 0000000..c87c918 --- /dev/null +++ b/examples/usb_mic.py @@ -0,0 +1,101 @@ +"""Capture a few seconds from a hosted USB microphone. + +The host-side mirror of what a device-side mic would be -- except the device +side cannot present a microphone yet (speaker-only UAC; usbif#7). This script +drives a commercial USB mic (or any UAC capture device) through +``usbif.uac_audio.input``, the same ``PCMInput`` surface as an on-board ADC. + + mpftp run -d COM49 examples/usb_mic.py + +Prints peak and RMS over the captured buffer so you can see that real audio +arrived, not silence. Verified originally against a C-Media USB mic and +Brad's voice (3,014 packets, zero dropped). +""" + +import time + +import _usbif +from usbif import uac +from usbif import uac_audio + +SECONDS = 3 + + +def find_mic(timeout_ms=15000): + _usbif.host_start(("uac",)) + deadline = time.ticks_add(time.ticks_ms(), timeout_ms) + while time.ticks_diff(deadline, time.ticks_ms()) > 0: + for dev_id, streams in uac_audio.audio_devices(): + ins = [s for s in streams if s.direction == uac.IN] + if ins: + return dev_id, ins + time.sleep_ms(250) + return None, () + + +def _peak_rms(buf, sample_bytes=2): + """Peak absolute sample and RMS over int16 LE mono/stereo interleaved.""" + n = len(buf) // sample_bytes + if n == 0: + return 0, 0.0 + peak = 0 + acc = 0 + for i in range(0, len(buf), sample_bytes): + sample = buf[i] | (buf[i + 1] << 8) + if sample >= 0x8000: + sample -= 0x10000 + a = sample if sample >= 0 else -sample + if a > peak: + peak = a + acc += sample * sample + rms = (acc / n) ** 0.5 + return peak, rms + + +def main(): + dev_id, ins = find_mic() + if dev_id is None: + print("no USB audio input found") + print("attach a USB microphone") + print("note: a PyDevices board cannot present as a mic yet (usbif#7)") + _usbif.host_stop() + return + + print("audio device", dev_id) + for s in ins: + print(" ", uac.describe(s)) + + mic = uac_audio.input(dev_id) + mic.open() + fmt = getattr(mic, "format", None) or getattr(mic, "_format", None) + if fmt is None: + stream = mic.stream + rate = max(stream.rates) if stream.rates else 48000 + channels, bits = stream.channels, stream.bits + else: + rate, channels, bits = fmt.rate, fmt.channels, fmt.bits + print("capturing %d s at %d Hz %d ch %d-bit" % (SECONDS, rate, channels, bits)) + + frame = channels * (bits // 8) + want = int(rate * SECONDS) * frame + buf = bytearray(want) + view = memoryview(buf) + got = 0 + t0 = time.ticks_ms() + try: + while got < want and time.ticks_diff(time.ticks_ms(), t0) < (SECONDS + 2) * 1000: + n = mic.readinto(view[got:]) + if n <= 0: + time.sleep_ms(5) + continue + got += n + finally: + peak, rms = _peak_rms(buf[:got]) + print("got %d / %d bytes; peak=%d rms=%.1f; stats %r" + % (got, want, peak, rms, mic.stats())) + mic.close() + _usbif.host_stop() + + +if __name__ == "__main__": + main() diff --git a/examples/usb_serial.py b/examples/usb_serial.py new file mode 100644 index 0000000..c43c8ac --- /dev/null +++ b/examples/usb_serial.py @@ -0,0 +1,61 @@ +"""Talk to a USB serial device the board is hosting. + +Opens the first CDC-ACM device on the host port, writes a short line, and +prints whatever comes back. A USB-serial adapter, another MCU's CDC console, +or a PyDevices board presenting CDC all work -- standard class, no bespoke +protocol. + + mpftp run -d COM49 examples/usb_serial.py + +**Pairing.** Device side is ordinary CDC (MicroPython's built-in console, or +``dev_functions(FN_CDC)``). Host side is this script on an S3. +""" + +import time + +import _usbif + + +def find_cdc(timeout_ms=15000): + _usbif.host_start(("cdc",)) + deadline = time.ticks_add(time.ticks_ms(), timeout_ms) + while time.ticks_diff(deadline, time.ticks_ms()) > 0: + for dev in _usbif.host_devices(): + if "cdc" in dev[5]: + return dev[0] + time.sleep_ms(250) + return None + + +def main(seconds=15): + dev_id = find_cdc() + if dev_id is None: + print("no CDC device found") + _usbif.host_stop() + return + + print("CDC device", dev_id) + _usbif.host_cdc_open(dev_id) + msg = b"hello from usbif host\r\n" + n = _usbif.host_cdc_write(msg) + print("wrote", n, "bytes:", msg) + + buf = bytearray(256) + t0 = time.ticks_ms() + total = 0 + try: + while time.ticks_diff(time.ticks_ms(), t0) < seconds * 1000: + got = _usbif.host_cdc_read(buf) + if got: + total += got + print("rx:", bytes(buf[:got])) + else: + time.sleep_ms(20) + finally: + print("total bytes read:", total) + _usbif.host_cdc_close() + _usbif.host_stop() + + +if __name__ == "__main__": + main() diff --git a/examples/usb_speaker.py b/examples/usb_speaker.py new file mode 100644 index 0000000..02be8c5 --- /dev/null +++ b/examples/usb_speaker.py @@ -0,0 +1,101 @@ +"""Play a tone through a hosted USB speaker -- the S3 half of the sound card. + +A commercial USB audio interface, a headset, or a PyDevices board running +``soundcard.py`` all look the same here: ``usbif.uac_audio.output`` returns an +ordinary ``audiodev.PCMOutput``. The application never knows a bus is involved. + + mpftp run -d COM49 examples/usb_speaker.py + +**Headline pairing.** P4 runs ``soundcard.py`` (device, C pump into its ES8311). +This script on an S3 hosts that P4 and plays through it. A PC hosting the P4 +works too -- standard classes are the protocol. + +**Power.** S3 host needs a powered hub or OTG adapter. Two self-powered boards +can back-feed unless the cable omits VBUS; see ``examples/README.md``. + +**FIFO bias.** Hosted stereo playback competes with hosted video for DWC FIFO +space (usbif#2). On a Bias-IN build, mono may be what survives. +""" + +import math +import time + +import _usbif +from usbif import uac +from usbif import uac_audio + + +def find_speaker(timeout_ms=15000): + _usbif.host_start(("uac",)) + deadline = time.ticks_add(time.ticks_ms(), timeout_ms) + while time.ticks_diff(deadline, time.ticks_ms()) > 0: + for dev_id, streams in uac_audio.audio_devices(): + out = [s for s in streams if s.direction == uac.OUT] + if out: + return dev_id, out + time.sleep_ms(250) + return None, () + + +def tone_frames(rate, channels, bits, freq=440.0, seconds=2.0, volume=0.2): + """Generate a sine as int16 little-endian frames.""" + n = int(rate * seconds) + amp = int(32767 * volume) + # One frame = channels samples. + raw = bytearray(n * channels * (bits // 8)) + for i in range(n): + sample = int(amp * math.sin(2 * math.pi * freq * i / rate)) + # int16 LE + lo = sample & 0xFF + hi = (sample >> 8) & 0xFF + base = i * channels * 2 + for ch in range(channels): + raw[base + ch * 2] = lo + raw[base + ch * 2 + 1] = hi + return raw + + +def main(): + dev_id, outs = find_speaker() + if dev_id is None: + print("no USB audio output found") + print("attach a speaker, headset, or a board running soundcard.py") + _usbif.host_stop() + return + + print("audio device", dev_id) + for s in outs: + print(" ", uac.describe(s)) + + out = uac_audio.output(dev_id) + out.open() + fmt = getattr(out, "format", None) or getattr(out, "_format", None) + if fmt is None: + # Fall back to the stream the adapter negotiated. + stream = out.stream + rate = max(stream.rates) if stream.rates else 48000 + channels, bits = stream.channels, stream.bits + else: + rate, channels, bits = fmt.rate, fmt.channels, fmt.bits + print("playing 440 Hz for 2 s at %d Hz %d ch %d-bit" % (rate, channels, bits)) + + pcm = tone_frames(rate, channels, bits) + view = memoryview(pcm) + sent = 0 + try: + while sent < len(pcm): + n = out.write(view[sent:]) + if n <= 0: + time.sleep_ms(5) + continue + sent += n + # Let the ring drain so the tail of the tone is audible. + time.sleep_ms(500) + finally: + print("sent", sent, "bytes; stats", out.stats()) + out.close() + _usbif.host_stop() + + +if __name__ == "__main__": + main() diff --git a/examples/uvc_display.py b/examples/uvc_display.py index 4b73645..a7561da 100644 --- a/examples/uvc_display.py +++ b/examples/uvc_display.py @@ -9,27 +9,21 @@ mpremote run uvc_display.py -**Why the uncompressed format and not MJPEG.** A webcam offers far better -resolutions in MJPEG than uncompressed -- on the bench camera, 640x480 against -176x144 -- so MJPEG is the tempting choice. It needs a JPEG decoder, and there -is not one reachable from Python in this firmware today. LVGL is compiled in -and its TJPGD decoder is enabled and registered, but the MicroPython bindings -expose only the decoder *types*, and the widget route (hand an ``lv.image`` a -variable ``image_dsc_t``) does not decode -- verified with the signature -repaired and the binary decoder stood down. Worse for this use, LVGL's -``is_jpg()`` demands a JFIF header in the first ten bytes and a UVC frame does -not have one: it opens straight into a quantisation table and carries its APP0 -segment after the Huffman tables. - -A ``jpegio`` native module is being added for exactly this path. When it -lands, this example should grow an MJPEG branch and prefer it -- the frames -are already whole JPEGs, complete with Huffman tables. Until then: -uncompressed frames, converted here, blitted straight to ``display_drv``. The -picture is small and the pixels are honest. - -**Why it is upscaled by whole numbers.** Nearest-neighbour at an integer +**MJPEG when ``jpegio`` is present.** A webcam offers far better resolutions +in MJPEG than uncompressed -- on the bench camera, 640x480 against 176x144 -- +so MJPEG is preferred when the firmware has ``jpegio`` (displayif). Frames +arrive as whole JPEGs; ``jpegio.JpegDecoder`` sniffs SOI only, which is what +UVC needs (a UVC MJPEG frame is not JFIF-first, so LVGL's ``is_jpg()`` rejects +it). Without ``jpegio`` the example falls back to uncompressed YUY2, converted +here and blitted to ``display_drv``. + +**Why YUY2 is upscaled by whole numbers.** Nearest-neighbour at an integer factor is a few instructions per pixel and needs no line buffer beyond one row. Anything smoother is a real resampler, which is a different example. + +**Pairing.** A PyDevices board presenting as a webcam (``usbif_webcam.py`` on +a P4) is a valid camera for this script on an S3, the same way a Logitech is. +P4 high-speed host is blocked (usbif#3), so the host role here is an S3. """ import time @@ -54,6 +48,13 @@ # 100 ns units, which is how UVC counts frame intervals throughout. INTERVALS = (2000000, 1333333, 1000000, 666666, 333333) # 5, 7.5, 10, 15, 30 fps +try: + import jpegio + _HAVE_JPEGIO = True +except ImportError: + jpegio = None + _HAVE_JPEGIO = False + @micropython.viper def yuy2_row_to_rgb565(src: ptr8, dst: ptr16, width: int, scale: int): @@ -135,25 +136,41 @@ def find_camera(timeout_ms=10000): return None -def pick_mode(dev_id, formats, alts, max_w, max_h): - """Negotiate the largest uncompressed mode the bus can actually carry. +def pick_mode(dev_id, formats, alts, max_w, max_h, prefer_mjpeg): + """Negotiate the largest mode the bus can actually carry. Two separate gates, and they have to be asked in this order. Whether a mode *exists* is in the descriptors; how much bandwidth it costs is not -- only the camera can say, and it says it by answering PROBE. So each candidate is negotiated for real before it is accepted or rejected. + + When ``prefer_mjpeg`` is true, MJPEG modes are tried first; otherwise + only uncompressed encodings are considered. """ candidates = [] for fmt in formats: - if fmt.encoding == "mjpeg" or not fmt.frames: + is_mjpeg = fmt.encoding == "mjpeg" + if prefer_mjpeg: + if not is_mjpeg and not fmt.frames: + continue + if not is_mjpeg: + # Prefer MJPEG; keep uncompressed as a fallback pass below. + continue + else: + if is_mjpeg or not fmt.frames: + continue + if not fmt.frames: continue for frame in fmt.frames: if frame.width > max_w or frame.height > max_h: continue - candidates.append((frame.width * frame.height, fmt, frame)) - candidates.sort(key=lambda c: c[0], reverse=True) + # MJPEG first when asked: score by size, then prefer mjpeg in the + # sort key so equal sizes still land on the compressed path. + rank = 1 if is_mjpeg else 0 + candidates.append((rank, frame.width * frame.height, fmt, frame)) + candidates.sort(key=lambda c: (c[0], c[1]), reverse=True) - for _, fmt, frame in candidates: + for _, _, fmt, frame in candidates: for interval in INTERVALS: if frame.intervals and interval not in frame.intervals: continue @@ -173,22 +190,40 @@ def _offered(formats): print(" offered:", uvc.describe(fmt)) +def _scale_geometry(frame): + scale = min(display_drv.width // frame.width, + display_drv.height // frame.height) or 1 + out_w = frame.width * scale + out_h = frame.height * scale + x0 = (display_drv.width - out_w) // 2 + y0 = (display_drv.height - out_h) // 2 + return scale, out_w, out_h, x0, y0 + + dev_id = find_camera() picked = None if dev_id is None: print("no UVC camera found on the host port") else: print("camera is device", dev_id) + if _HAVE_JPEGIO: + print("jpegio present -- preferring MJPEG") + else: + print("jpegio absent -- uncompressed YUY2 only") blob = _usbif.host_desc(dev_id) formats = uvc.formats(blob) alts = uvc.alt_settings(blob) if not formats: print("camera declares no video formats") else: - picked = pick_mode(dev_id, formats, alts, - display_drv.width, display_drv.height) + if _HAVE_JPEGIO: + picked = pick_mode(dev_id, formats, alts, + display_drv.width, display_drv.height, True) + if picked is None: + picked = pick_mode(dev_id, formats, alts, + display_drv.width, display_drv.height, False) if picked is None: - print("no uncompressed mode fits this host's isochronous IN limit") + print("no mode fits this host's isochronous IN limit") _offered(formats) if picked is not None: @@ -196,74 +231,108 @@ def _offered(formats): print("streaming", uvc.describe(fmt, frame, interval)) print("payload %d B/frame on alt %d" % (payload, alt.alt)) - # Whole-number upscale, centred. A 176x144 frame becomes 528x432 on an - # 800x480 panel; a display smaller than the frame falls back to 1:1. - scale = min(display_drv.width // frame.width, - display_drv.height // frame.height) or 1 - out_w = frame.width * scale - out_h = frame.height * scale - x0 = (display_drv.width - out_w) // 2 - y0 = (display_drv.height - out_h) // 2 + scale, out_w, out_h, x0, y0 = _scale_geometry(frame) print("%dx%d upscaled x%d -> %dx%d at (%d, %d)" % (frame.width, frame.height, scale, out_w, out_h, x0, y0)) - # One *band* of `scale` identical rows, reused per source row. Two reasons - # it is a band and not a single row. Building the whole scaled frame would - # be out_w * out_h * 2 bytes -- close to half a megabyte at 528x432 -- for - # no benefit. And blit_rect byteswaps its buffer **in place** when the - # panel needs it, so blitting one row buffer `scale` times would swap it - # again on every call and leave every repeated row with its colours - # inverted. One buffer, one blit, one swap. - band_stride = out_w * 2 - band = bytearray(band_stride * scale) - band_mv = memoryview(band) src = bytearray(frame_bytes) - src_mv = memoryview(src) - src_stride = frame.width * 2 - whole_frame = src_stride * frame.height - _usbif.host_uvc_open(dev_id, fmt.interface, alt.alt, alt.endpoint, alt.max_packet, frame_bytes) display_drv.fill(0) _shown = 0 _t0 = time.ticks_ms() - def _tick(_=None): - """Blit a frame if one has arrived; cheap when none has. - - Scheduled rather than looped. A ``while True`` here would work and - would be wrong: it owns the interpreter, so touch, the REPL and - anything else the app is running never get a turn. ``app.every`` is - what makes this a program the board runs rather than a program that - takes the board over -- the same reason paint.py hands its drawing to - the app's event dispatch instead of polling. - """ - global _shown - n = _usbif.host_uvc_read_frame(src) - if n <= 0: - return - # A short frame means the camera sent less than a whole picture. - # Showing it would tear the bottom of one image across the top of the - # next, so skip it and leave the last good frame up. - if n < whole_frame: - return - for sy in range(frame.height): - yuy2_row_to_rgb565(src_mv[sy * src_stride:], band, - frame.width, scale) - for k in range(1, scale): - band_mv[k * band_stride:(k + 1) * band_stride] = \ - band_mv[0:band_stride] - display_drv.blit_rect(band, x0, y0 + sy * scale, out_w, scale) - # dotclockframebuffer double-buffers, so the back buffer is only - # promoted by show(). Called here, in the scheduled work, exactly as - # paint.py calls it from the handlers the app dispatches. - display_drv.show() - _shown += 1 - if _shown % 25 == 0: - dt = time.ticks_diff(time.ticks_ms(), _t0) - print("%d frames, %.1f fps, stats %r" - % (_shown, _shown * 1000 / dt, _usbif.host_uvc_stats())) - - # 10 ms, matching bouncing_balls. Frames arrive every 200 ms at 5 fps, so - # nearly every tick returns immediately. + if fmt.encoding == "mjpeg": + decoder = jpegio.JpegDecoder() + # Native-order RGB565, tight. Sized for the negotiated frame; a camera + # that sends a larger JPEG than it advertised is refused by decode. + rgb = bytearray(frame.width * frame.height * 2) + # Nearest-neighbour upscale into a band, same reason as the YUY2 path: + # blit_rect byteswaps in place, so one band / one blit / one swap. + band_stride = out_w * 2 + band = bytearray(band_stride * scale) + band_mv = memoryview(band) + rgb_mv = memoryview(rgb) + src_stride = frame.width * 2 + + def _tick(_=None): + global _shown + n = _usbif.host_uvc_read_frame(src) + if n <= 0: + return + try: + decoder.open(memoryview(src)[:n]) + except Exception: + return + if decoder.width != frame.width or decoder.height != frame.height: + return + try: + decoder.decode(rgb, scale=0) + except Exception: + return + # Integer upscale row by row into the band, then blit. + for sy in range(frame.height): + row = rgb_mv[sy * src_stride:(sy + 1) * src_stride] + if scale == 1: + display_drv.blit_rect(row, x0, y0 + sy, out_w, 1) + else: + # Expand horizontally into the first band row, then + # replicate vertically. + o = 0 + for px in range(0, src_stride, 2): + pix = row[px:px + 2] + for _k in range(scale): + band_mv[o:o + 2] = pix + o += 2 + for k in range(1, scale): + band_mv[k * band_stride:(k + 1) * band_stride] = \ + band_mv[0:band_stride] + display_drv.blit_rect(band, x0, y0 + sy * scale, out_w, scale) + display_drv.show() + _shown += 1 + if _shown % 25 == 0: + dt = time.ticks_diff(time.ticks_ms(), _t0) + print("%d frames, %.1f fps, stats %r" + % (_shown, _shown * 1000 / dt, _usbif.host_uvc_stats())) + + else: + # Uncompressed YUY2 path. + band_stride = out_w * 2 + band = bytearray(band_stride * scale) + band_mv = memoryview(band) + src_mv = memoryview(src) + src_stride = frame.width * 2 + whole_frame = src_stride * frame.height + + def _tick(_=None): + """Blit a frame if one has arrived; cheap when none has. + + Scheduled rather than looped. A ``while True`` here would work and + would be wrong: it owns the interpreter, so touch, the REPL and + anything else the app is running never get a turn. + """ + global _shown + n = _usbif.host_uvc_read_frame(src) + if n <= 0: + return + # A short frame means the camera sent less than a whole picture. + # Showing it would tear; skip and leave the last good frame up. + if n < whole_frame: + return + for sy in range(frame.height): + yuy2_row_to_rgb565(src_mv[sy * src_stride:], band, + frame.width, scale) + for k in range(1, scale): + band_mv[k * band_stride:(k + 1) * band_stride] = \ + band_mv[0:band_stride] + display_drv.blit_rect(band, x0, y0 + sy * scale, out_w, scale) + display_drv.show() + _shown += 1 + if _shown % 25 == 0: + dt = time.ticks_diff(time.ticks_ms(), _t0) + print("%d frames, %.1f fps, stats %r" + % (_shown, _shown * 1000 / dt, _usbif.host_uvc_stats())) + + # 10 ms. Frames arrive every 200 ms at 5 fps, so nearly every tick returns + # immediately. app.every(_tick, period=10, async_=app.timer_async)