diff --git a/Makefile b/Makefile index fe77358..30a1259 100644 --- a/Makefile +++ b/Makefile @@ -105,6 +105,7 @@ install: build $(MAKE) -C sof-remote-server DESTDIR="$(DESTDIR)" PREFIX="$(PREFIX)" BINDIR="$(BINDIR)" WORK_DIR="$(WORK_DIR)" install $(MAKE) -C sof-builder DESTDIR="$(DESTDIR)" PREFIX="$(PREFIX)" BINDIR="$(BINDIR)" DATADIR="$(DATADIR)" install $(MAKE) -C sof-top DESTDIR="$(DESTDIR)" PREFIX="$(PREFIX)" BINDIR="$(BINDIR)" DATADIR="$(DATADIR)" install + $(MAKE) -C sof-teensy DESTDIR="$(DESTDIR)" PREFIX="$(PREFIX)" BINDIR="$(BINDIR)" DATADIR="$(DATADIR)" install $(INSTALL) -d "$(DESTDIR)$(DOCDIR)" $(INSTALL) -m 0644 README.md LICENSE "$(DESTDIR)$(DOCDIR)/" @if [ -f "$(TB_BUILD_DIR)/sof-testbench4" ]; then \ @@ -130,6 +131,7 @@ uninstall: $(MAKE) -C sof-remote-server DESTDIR="$(DESTDIR)" PREFIX="$(PREFIX)" BINDIR="$(BINDIR)" uninstall $(MAKE) -C sof-builder DESTDIR="$(DESTDIR)" PREFIX="$(PREFIX)" BINDIR="$(BINDIR)" DATADIR="$(DATADIR)" uninstall $(MAKE) -C sof-top DESTDIR="$(DESTDIR)" PREFIX="$(PREFIX)" BINDIR="$(BINDIR)" DATADIR="$(DATADIR)" uninstall + $(MAKE) -C sof-teensy DESTDIR="$(DESTDIR)" PREFIX="$(PREFIX)" BINDIR="$(BINDIR)" DATADIR="$(DATADIR)" uninstall rm -f "$(DESTDIR)$(BINDIR)/sof-testbench4" \ "$(DESTDIR)$(BINDIR)/sof-testbench3" \ "$(DESTDIR)$(BINDIR)/sof-tb-ctl" \ @@ -140,6 +142,7 @@ uninstall: check: $(MAKE) -C sof-builder PYTHON="$(PYTHON)" check $(MAKE) -C sof-top PYTHON="$(PYTHON)" check + $(MAKE) -C sof-teensy PYTHON="$(PYTHON)" check test: check diff --git a/README.md b/README.md index f5b2993..7d7310f 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ Developer utilities, visual tools, and host applications for the [Sound Open Fir | **[`sof-builder`](sof-builder/)** | [`sof-builder/`](sof-builder/) | GTK4 Topology2 Visual Graph Editor, IPC4 `.tplg` Compiler & Real-Time `sof-testbench4` Simulation / Telemetry Studio | | **[`sof-testbench`](sof-testbench/)** | [`sof-testbench/`](sof-testbench/) | Multi-Core LL/DP `sof-testbench4` Simulator, `sof-tb-ctl` TCP Control CLI, Host Platform Runtime (`platform/library`), and 70-Test Suite | | **[`sof-tplg-parser`](sof-tplg-parser/)** | [`sof-tplg-parser/`](sof-tplg-parser/) | ALSA Topology (`.tplg`) IPC3/IPC4 Parser Library (`libsof_tplg_parser.a`) | +| **[`sof-teensy`](sof-teensy/)** | [`sof-teensy/`](sof-teensy/) | PJRC Teensy 4.1 HalfKay Flashing, Hardware Loopback Verification (S/PDIF & SAI1 I2S), and Diagnostic Utilities | --- diff --git a/sof-teensy/Makefile b/sof-teensy/Makefile new file mode 100644 index 0000000..e3000f3 --- /dev/null +++ b/sof-teensy/Makefile @@ -0,0 +1,35 @@ +# SPDX-License-Identifier: BSD-3-Clause +PYTHON ?= python3 +INSTALL ?= install +PREFIX ?= /usr +BINDIR ?= $(PREFIX)/bin +DATADIR ?= $(PREFIX)/share +APPDIR ?= $(DATADIR)/sof-utils/sof-teensy +DESTDIR ?= + +SCRIPTS := teensy_flash.py teensy_diag.py test_teensy_loopback.py + +.PHONY: all check install uninstall clean + +all: check + +check: + $(PYTHON) -B -m py_compile $(SCRIPTS) + +install: + $(INSTALL) -d "$(DESTDIR)$(BINDIR)" + $(INSTALL) -d "$(DESTDIR)$(APPDIR)" + $(INSTALL) -m 0755 $(SCRIPTS) "$(DESTDIR)$(APPDIR)/" + $(INSTALL) -m 0644 README.md "$(DESTDIR)$(APPDIR)/" + ln -sf "$(APPDIR)/teensy_flash.py" "$(DESTDIR)$(BINDIR)/sof-teensy-flash" + ln -sf "$(APPDIR)/teensy_diag.py" "$(DESTDIR)$(BINDIR)/sof-teensy-diag" + ln -sf "$(APPDIR)/test_teensy_loopback.py" "$(DESTDIR)$(BINDIR)/sof-teensy-loopback" + +uninstall: + rm -f "$(DESTDIR)$(BINDIR)/sof-teensy-flash" \ + "$(DESTDIR)$(BINDIR)/sof-teensy-diag" \ + "$(DESTDIR)$(BINDIR)/sof-teensy-loopback" + rm -rf "$(DESTDIR)$(APPDIR)" + +clean: + rm -rf __pycache__ diff --git a/sof-teensy/README.md b/sof-teensy/README.md new file mode 100644 index 0000000..5b15ede --- /dev/null +++ b/sof-teensy/README.md @@ -0,0 +1,28 @@ +# SOF Teensy 4.1 Hardware Utilities (`sof-teensy`) + +Host utilities for flashing, inspecting, and validating PJRC Teensy 4.1 (NXP i.MX RT1062) boards running Sound Open Firmware (SOF) over USB Audio Class 2.0 (UAC2). + +## Included Scripts + +| Script | Installed CLI | Description | +| :--- | :--- | :--- | +| [`teensy_flash.py`](teensy_flash.py) | `sof-teensy-flash` | Triggers PJRC HalfKay bootloader entry on a specific Teensy 4.1 board (`--serial a` or `--serial b`) and flashes an Intel HEX image via `teensy_loader_cli` | +| [`test_teensy_loopback.py`](test_teensy_loopback.py) | `sof-teensy-loopback` | Runs automated cross-board digital audio loopback verification (S/PDIF or SAI1 I2S) between Board A (Master Tx) and Board B (Slave Rx), computing FFT peak frequency, SNR, and RMS amplitude | +| [`teensy_diag.py`](teensy_diag.py) | `sof-teensy-diag` | Queries diagnostic builds exposing optional USB vendor status reports | + +## Usage + +### 1. Flash Board A and Board B +```bash +./sof-teensy/teensy_flash.py --serial a --hex /path/to/build-teensya/zephyr/zephyr.hex +./sof-teensy/teensy_flash.py --serial b --hex /path/to/build-teensyb/zephyr/zephyr.hex +``` + +### 2. Run Digital Loopback Verification +```bash +# S/PDIF transceiver loopback (Board A Pin 14 -> Board B Pin 15) +./sof-teensy/test_teensy_loopback.py --interface spdif + +# SAI1 I2S loopback (Pins 7, 8, 20, 21, 23) +./sof-teensy/test_teensy_loopback.py --interface i2s +``` diff --git a/sof-teensy/teensy_diag.py b/sof-teensy/teensy_diag.py new file mode 100755 index 0000000..d7e8f84 --- /dev/null +++ b/sof-teensy/teensy_diag.py @@ -0,0 +1,168 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: BSD-3-Clause +""" +Teensy 4.1 SOF Audio Hardware Diagnostic Tool +Queries real-time hardware registers, DMA TCD status, and SOF pipeline states via USB vendor request 0xFC. +""" + +import sys +import struct +import usb.core +import usb.util + +TEENSY_VID = 0x16C0 +TEENSY_PID = 0x04D2 + +# Port mappings +PORTS = { + 'a': [(3, (5, 3)), (3, (12, 4, 1, 3))], + 'b': [(3, (5, 2)), (3, (12, 4, 1, 2))], +} + +DIAG_FMT_44 = "<44I" +DIAG_FMT_32 = "<32I" +DIAG_FMT_24 = "<24I" +DIAG_FMT_LEGACY = "<20I" + +def query_diag(target): + target = target.lower() + target_ports = PORTS.get(target) + if target_ports and not isinstance(target_ports, list): + target_ports = [target_ports] + + devices = list(usb.core.find(find_all=True, idVendor=TEENSY_VID, idProduct=TEENSY_PID)) + match_dev = None + for d in devices: + port = (d.bus, tuple(d.port_numbers) if d.port_numbers else None) + if target_ports and port in target_ports: + match_dev = d + break + elif not target_ports: + match_dev = d + break + + if not match_dev: + print(f"Error: Teensy '{target}' not found!") + return None + + try: + # Vendor IN request 0xFC, recipient device (0xC0) + data = match_dev.ctrl_transfer(0xC0, 0xFC, 0, 0, 256, timeout=1000) + except Exception as e: + print(f"Error reading diag from Teensy '{target}': {e}") + return None + + if len(data) >= struct.calcsize(DIAG_FMT_44): + unpacked = struct.unpack(DIAG_FMT_44, bytes(data[:struct.calcsize(DIAG_FMT_44)])) + fields = [ + "tcsr", "rcsr", "tcr2", "rcr2", "tcr3", "rcr3", "tfr0", "rfr0", + "rx_pkt_cnt", "tx_pkt_cnt", "pipe1_status", "pipe2_status", + "dai5_state", "dai6_state", + "dma_tx_csr", "dma_rx_csr", "dma_tx_saddr", "dma_rx_daddr", + "dma_tx_citer", "dma_rx_citer", + "c1_state", "c2_state", "c9_state", "c10_state", + "tcr4", "rcr4", "tcr5", "rcr5", + "pin8_mux", "pin8_daisy", "pin7_mux", "pin8_pad", "pin7_pad", + "gpio1_psr", "gpio2_psr", "interface_type", + "g_dbg0", "g_dbg1", "g_dbg2", "g_dbg3", + "g_dbg4", "g_dbg5", "g_dbg6", "g_dbg7" + ] + elif len(data) >= struct.calcsize(DIAG_FMT_32): + unpacked = struct.unpack(DIAG_FMT_32, bytes(data[:struct.calcsize(DIAG_FMT_32)])) + fields = [ + "tcsr", "rcsr", "tcr2", "rcr2", "tcr3", "rcr3", "tfr0", "rfr0", + "rx_pkt_cnt", "tx_pkt_cnt", "pipe1_status", "pipe2_status", + "dai5_state", "dai6_state", + "dma_tx_csr", "dma_rx_csr", "dma_tx_saddr", "dma_rx_daddr", + "dma_tx_citer", "dma_rx_citer", + "c1_state", "c2_state", "c9_state", "c10_state", + "tcr4", "rcr4", "tcr5", "rcr5", + "pin8_mux", "pin8_daisy", "pin7_mux", "pin8_pad" + ] + elif len(data) >= struct.calcsize(DIAG_FMT_24): + unpacked = struct.unpack(DIAG_FMT_24, bytes(data[:struct.calcsize(DIAG_FMT_24)])) + fields = [ + "tcsr", "rcsr", "tcr2", "rcr2", "tcr3", "rcr3", "tfr0", "rfr0", + "rx_pkt_cnt", "tx_pkt_cnt", "pipe1_status", "pipe2_status", + "dai5_state", "dai6_state", + "dma_tx_csr", "dma_rx_csr", "dma_tx_saddr", "dma_rx_daddr", + "dma_tx_citer", "dma_rx_citer", + "c1_state", "c2_state", "c9_state", "c10_state" + ] + elif len(data) >= struct.calcsize(DIAG_FMT_LEGACY): + unpacked = struct.unpack(DIAG_FMT_LEGACY, bytes(data[:struct.calcsize(DIAG_FMT_LEGACY)])) + fields = [ + "tcsr", "rcsr", "tcr2", "rcr2", "tcr3", "rcr3", "tfr0", "rfr0", + "rx_pkt_cnt", "tx_pkt_cnt", "pipe1_status", "pipe2_status", + "dai5_state", "dai6_state", + "dma_tx_csr", "dma_rx_csr", "dma_tx_saddr", "dma_rx_daddr", + "dma_tx_citer", "dma_rx_citer" + ] + else: + print(f"Error: Short response ({len(data)} bytes)") + return None + + res = dict(zip(fields, unpacked)) + return res + +def print_diag(target, res): + print(f"=== Teensy 4.1 '{target.upper()}' Diag Report ===") + if res.get('interface_type') == 2: + print(" Active Hardware Interface: S/PDIF Transceiver (Board A Pin 14 ALT3 -> Board B Pin 15 ALT3)") + print(f" SPDIF SCR: 0x{res['tcsr']:08X} (TXFIFO_CTRL={(res['tcsr']>>10)&3}, USRC_SEL={(res['tcsr']>>2)&3})") + print(f" SPDIF SRPC: 0x{res['rcsr']:08X} (CLKSRC_SEL={(res['rcsr']>>7)&0xF}, GAIN_SEL={(res['rcsr']>>3)&7})") + print(f" SPDIF STC: 0x{res['tcr2']:08X} (TXCLK_SOURCE={(res['tcr2']>>8)&7}, SYSCLK_DF={res['tcr2']&0xFF})") + print(f" SPDIF SRFM: 0x{res['rcr2']:08X} (FREQMEAS={res['rcr2']&0xFFFFFF})") + print(f" SPDIF SIS: 0x{res['tcr3']:08X}") + print(f" SPDIF SIE: 0x{res['rcr3']:08X}") + else: + print(" Active Hardware Interface: SAI1 I2S (Pins 7, 8, 20, 21, 23)") + print(f" I2S1 TCSR: 0x{res['tcsr']:08X} (TE={bool(res['tcsr'] & (1<<31))}, BCE={bool(res['tcsr'] & (1<<28))}, FEF={bool(res['tcsr'] & (1<<18))})") + print(f" I2S1 RCSR: 0x{res['rcsr']:08X} (RE={bool(res['rcsr'] & (1<<31))}, BCE={bool(res['rcsr'] & (1<<28))}, FEF={bool(res['rcsr'] & (1<<18))})") + print(f" I2S1 TCR2: 0x{res['tcr2']:08X} (SYNC={(res['tcr2']>>30)&3}, BCD={(res['tcr2']>>24)&1}, MSEL={(res['tcr2']>>26)&3}, DIV={(res['tcr2']&0xFF)})") + print(f" I2S1 RCR2: 0x{res['rcr2']:08X} (SYNC={(res['rcr2']>>30)&3}, BCD={(res['rcr2']>>24)&1}, MSEL={(res['rcr2']>>26)&3}, DIV={(res['rcr2']&0xFF)})") + print(f" I2S1 TCR3: 0x{res['tcr3']:08X} (TCE={(res['tcr3']>>16)&0xF})") + print(f" I2S1 RCR3: 0x{res['rcr3']:08X} (RCE={(res['rcr3']>>16)&0xF})") + if 'tcr4' in res and res.get('interface_type') != 2: + print(f" I2S1 TCR4: 0x{res['tcr4']:08X} (FSE={(res['tcr4']>>3)&1}, FSP={(res['tcr4']>>1)&1}, FSD={res['tcr4']&1}, SYWD={(res['tcr4']>>8)&0x1F}, FRSZ={(res['tcr4']>>16)&0xF})") + print(f" I2S1 RCR4: 0x{res['rcr4']:08X} (FSE={(res['rcr4']>>3)&1}, FSP={(res['rcr4']>>1)&1}, FSD={res['rcr4']&1}, SYWD={(res['rcr4']>>8)&0x1F}, FRSZ={(res['rcr4']>>16)&0xF})") + print(f" I2S1 TCR5: 0x{res['tcr5']:08X} (WNW={(res['tcr5']>>24)&0x1F}, W0W={(res['tcr5']>>16)&0x1F}, FBT={(res['tcr5']>>8)&0x1F})") + print(f" I2S1 RCR5: 0x{res['rcr5']:08X} (WNW={(res['rcr5']>>24)&0x1F}, W0W={(res['rcr5']>>16)&0x1F}, FBT={(res['rcr5']>>8)&0x1F})") + p8_mux = res.get('pin8_mux', res.get('dbg0', 0)) + p8_daisy = res.get('pin8_daisy', res.get('dbg1', 0)) + p7_mux = res.get('pin7_mux', res.get('dbg2', 0)) + p8_pad = res.get('pin8_pad', res.get('dbg3', 0)) + p7_pad = res.get('pin7_pad', 0) + print(f" Pinmux: Pin8_MUX(B1_00)=0x{p8_mux:08X}, Pin8_DAISY=0x{p8_daisy:08X}, Pin7_MUX(B1_01)=0x{p7_mux:08X}, Pin8_PAD=0x{p8_pad:08X}, Pin7_PAD=0x{p7_pad:08X}") + if 'gpio1_psr' in res: + g1 = res['gpio1_psr'] + g2 = res['gpio2_psr'] + # Pin 23: GPIO1_25 (MCLK), Pin 20: GPIO1_26 (SYNC), Pin 21: GPIO1_27 (BCLK) + # Pin 8: GPIO2_16 (RX), Pin 7: GPIO2_17 (TX) + print(f" GPIO: GPIO1_PSR=0x{g1:08X} (P23_MCLK={(g1>>25)&1}, P20_SYNC={(g1>>26)&1}, P21_BCLK={(g1>>27)&1}), GPIO2_PSR=0x{g2:08X} (P8_RX={(g2>>16)&1}, P7_TX={(g2>>17)&1})") + if 'g_dbg0' in res: + print(f" SOF DAI Dbg: PB[calls={res['g_dbg0']}, copy=0x{res['g_dbg1']:08X}, p0=0x{res['g_dbg2']:08X}, p1=0x{res['g_dbg3']:08X}]") + print(f" CAP[calls={res['g_dbg4']}, copy=0x{res['g_dbg5']:08X}, p0=0x{res['g_dbg6']:08X}, p1=0x{res['g_dbg7']:08X}]") + print(f" I2S1 TFR0: 0x{res['tfr0']:08X} (WFP={(res['tfr0']>>16)&0x3F}, RFP={res['tfr0']&0x3F})") + print(f" I2S1 RFR0: 0x{res['rfr0']:08X} (WFP={(res['rfr0']>>16)&0x3F}, RFP={res['rfr0']&0x3F})") + print(f" USB Packets: Rx(Host PB)={res['rx_pkt_cnt']}, Tx(Host CAP)={res['tx_pkt_cnt']}") + print(f" Pipelines: P1(PB) state={res['pipe1_status']}, P2(CAP) state={res['pipe2_status']}") + comps_str = f"DAI5={res.get('dai5_state')}, DAI6={res.get('dai6_state')}" + if 'c1_state' in res: + comps_str += f", C1(USB_PB)={res['c1_state']}, C2(VOL_PB)={res['c2_state']}, C9(VOL_CAP)={res['c9_state']}, C10(USB_CAP)={res['c10_state']}" + print(f" Components: {comps_str}") + print(f" eDMA TX: CSR=0x{res['dma_tx_csr']:04X}, SADDR=0x{res['dma_tx_saddr']:08X}, CITER={res['dma_tx_citer']}") + print(f" eDMA RX: CSR=0x{res['dma_rx_csr']:04X}, DADDR=0x{res['dma_rx_daddr']:08X}, CITER={res['dma_rx_citer']}") + print() + +if __name__ == '__main__': + target = sys.argv[1] if len(sys.argv) > 1 else 'all' + if target in ('all', 'both'): + for t in ('a', 'b'): + r = query_diag(t) + if r: + print_diag(t, r) + else: + r = query_diag(target) + if r: + print_diag(target, r) diff --git a/sof-teensy/teensy_flash.py b/sof-teensy/teensy_flash.py new file mode 100755 index 0000000..96ad2c0 --- /dev/null +++ b/sof-teensy/teensy_flash.py @@ -0,0 +1,332 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: BSD-3-Clause +""" +Teensy 4.1 USB Flasher Tool + +Supports: +- Distinguishing multiple Teensy boards by serial number or USB path +- Automatic soft-reboot from Teensyduino (RawHID / SerEmu / CDC-ACM) to HalfKay bootloader +- Parsing standard Zephyr Intel HEX files (mapping 0x60000000 -> 0x00000000) +- Programming flash via HalfKay USB HID SET_REPORT control transfers +- Booting the newly flashed firmware +""" + +import sys +import time +import argparse +import usb.core +import usb.util + +TEENSY_VID = 0x16C0 +HALFKAY_PID = 0x0478 +RAWHID_PID = 0x0486 +CDC_PID = 0x0483 +SEREMU_PID = 0x0487 + +BLOCK_SIZE = 1024 +FLASH_SIZE = 8126464 # Teensy 4.1 8MB flash (minus HalfKay area) + + +def parse_intel_hex(hex_path): + """ + Parses an Intel HEX file and returns a bytearray of firmware data + mapped to flash offset 0. + """ + image = bytearray(b'\xFF' * FLASH_SIZE) + max_addr = 0 + ext_addr = 0 + + with open(hex_path, 'r') as f: + for line in f: + line = line.strip() + if not line.startswith(':'): + continue + length = int(line[1:3], 16) + addr = int(line[3:7], 16) + rtype = int(line[7:9], 16) + data_hex = line[9:9 + length * 2] + + if rtype == 0: # Data Record + full_addr = ext_addr + addr + if full_addr >= 0x60000000 and full_addr < 0x60000000 + FLASH_SIZE: + flash_addr = full_addr - 0x60000000 + elif full_addr < FLASH_SIZE: + flash_addr = full_addr + else: + continue # Ignore addresses outside flash + + for i, byte in enumerate(bytes.fromhex(data_hex)): + if flash_addr + i < FLASH_SIZE: + image[flash_addr + i] = byte + if flash_addr + i > max_addr: + max_addr = flash_addr + i + + elif rtype == 1: # EOF + break + elif rtype == 2: # Extended Segment Address + ext_addr = int(data_hex, 16) << 4 + elif rtype == 4: # Extended Linear Address + ext_addr = int(data_hex, 16) << 16 + + print(f"Loaded HEX '{hex_path}': {max_addr + 1} bytes (up to block {(max_addr // BLOCK_SIZE) + 1})") + return image, max_addr + 1 + + +def send_soft_reboot(dev): + """Sends soft reboot command to Teensy running Teensyduino firmware.""" + for iface in [0, 1]: + try: + if dev.is_kernel_driver_active(iface): + dev.detach_kernel_driver(iface) + except Exception: + pass + + try: + # SerEmu reboot token (Teensy 4.0 / 4.1 RawHID interface 1 feature report 0) + reboot_token = bytes([0xA9, 0x45, 0xC2, 0x6B]) + dev.ctrl_transfer(0x21, 0x09, 0x0300, 1, reboot_token, timeout=200) + print("Sent SerEmu reboot token (0x0300, iface 1)") + return True + except Exception: + pass + + try: + # CDC-ACM line coding reboot (134 baud) + reboot_cdc = bytes([0x86, 0x00, 0x00, 0x00, 0x00, 0x00, 0x08]) # 134 bps + dev.ctrl_transfer(0x21, 0x20, 0, 0, reboot_cdc, timeout=200) + print("Sent CDC 134 baud reboot request") + return True + except Exception: + pass + + try: + # SOF Teensy vendor reboot request (0xFB) + dev.ctrl_transfer(0x40, 0xFB, 0, 0, None, timeout=200) + print("Sent SOF Teensy vendor reboot request (0xFB)") + return True + except Exception: + pass + + return False + + +def find_teensy(target_serial=None, target_port=None, bootloader_only=False): + """Finds matching Teensy device.""" + devices = list(usb.core.find(find_all=True, idVendor=TEENSY_VID)) + matches = [] + for d in devices: + try: + s = d.serial_number + except Exception: + s = None + port = (d.bus, tuple(d.port_numbers) if d.port_numbers else None) + + if d.idProduct == HALFKAY_PID: + matches.append(('halfkay', d, s, port)) + elif not bootloader_only: + matches.append(('firmware', d, s, port)) + + if target_port: + ports = target_port if isinstance(target_port, (list, tuple)) and isinstance(target_port[0], tuple) else [target_port] + filtered = [m for m in matches if m[3] in ports] + if filtered: + return filtered[0] + + if target_serial: + s_target = str(target_serial).lower() + for m in matches: + s_dev = str(m[2]).lower() if m[2] else "" + if s_dev and (s_dev == s_target or (s_target in ('a', 'teensya', '17854640', '001b3e78') and s_dev == '001b3e78') or (s_target in ('b', 'teensyb', '17861730', '001b413d') and s_dev == '001b413d')): + return m + + if not target_serial and not target_port and matches: + return matches[0] + return None + + +KNOWN_BOARDS = { + '17854640': [(3, (5, 3)), (3, (12, 4, 1, 3))], + '17861730': [(3, (5, 2)), (3, (12, 4, 1, 2))], + '001b3e78': [(3, (5, 3)), (3, (12, 4, 1, 3))], + '001b413d': [(3, (5, 2)), (3, (12, 4, 1, 2))], + 'a': [(3, (5, 3)), (3, (12, 4, 1, 3))], + 'b': [(3, (5, 2)), (3, (12, 4, 1, 2))], + 'teensya': [(3, (5, 3)), (3, (12, 4, 1, 3))], + 'teensyb': [(3, (5, 2)), (3, (12, 4, 1, 2))], +} + + + +def write_block(dev, buf, timeout_sec=2.0): + start = time.time() + while time.time() - start < timeout_sec: + try: + ret = dev.ctrl_transfer(0x21, 9, 0x0200, 0, buf, timeout=1000) + if ret == len(buf): + return True + except usb.core.USBError: + time.sleep(0.01) + return False + + +def program_device(dev, image, total_bytes): + """Programs a Teensy 4.1 device currently in HalfKay bootloader mode.""" + try: + if dev.is_kernel_driver_active(0): + dev.detach_kernel_driver(0) + except Exception: + pass + + num_blocks = (total_bytes + BLOCK_SIZE - 1) // BLOCK_SIZE + buf = bytearray(1088) + + for b in range(num_blocks): + addr = b * BLOCK_SIZE + chunk = image[addr:addr + BLOCK_SIZE] + + if b > 0 and all(x == 0xFF for x in chunk): + continue + + buf[0] = addr & 0xFF + buf[1] = (addr >> 8) & 0xFF + buf[2] = (addr >> 16) & 0xFF + buf[3:64] = b'\x00' * 61 + buf[64:64 + BLOCK_SIZE] = chunk + + timeout_sec = 45.0 if b <= 4 else 2.0 + if not write_block(dev, buf, timeout_sec): + raise IOError(f"Failed to write block {b} at addr 0x{addr:08x} after {timeout_sec}s") + + if (b % 32 == 0) or (b == num_blocks - 1): + pct = (b + 1) / num_blocks * 100.0 + print(f"Programming: block {b + 1}/{num_blocks} ({pct:.1f}%)") + + print("Programming complete. Sending reboot token...") + buf[0] = 0xFF + buf[1] = 0xFF + buf[2] = 0xFF + buf[3:] = b'\x00' * (1088 - 3) + time.sleep(0.1) + if not write_block(dev, buf, timeout_sec=5.0): + print("Warning: Reboot token transfer timed out") + else: + print("Teensy 4.1 rebooted into new firmware successfully!") + + +def flash_teensy(hex_path, target_serial=None, timeout=30): + """Flashes an Intel HEX firmware image to Teensy 4.1.""" + image, total_bytes = parse_intel_hex(hex_path) + + print(f"Searching for Teensy (Target Serial: {target_serial or 'any'})...") + start = time.time() + dev_entry = None + target_port = None + if target_serial and str(target_serial).lower() in KNOWN_BOARDS: + target_port = KNOWN_BOARDS[str(target_serial).lower()] + print(f"Target serial '{target_serial}' mapped to USB port {target_port}") + + while time.time() - start < timeout: + dev_entry = find_teensy(target_serial=target_serial, target_port=target_port, bootloader_only=False) + if dev_entry: + mode, dev, serial, port = dev_entry + target_port = port + print(f"Found Teensy in '{mode}' mode (Serial: {serial}, Port: {port}, VID:PID: {dev.idVendor:04x}:{dev.idProduct:04x})") + if mode == 'firmware': + print("Triggering soft reboot into HalfKay bootloader...") + send_soft_reboot(dev) + time.sleep(1.0) + continue + elif mode == 'halfkay': + break + time.sleep(0.5) + + if not dev_entry or dev_entry[0] != 'halfkay': + raise RuntimeError( + "Could not connect to Teensy HalfKay bootloader. " + "Please press the pushbutton on the Teensy board." + ) + + mode, dev, serial, port = dev_entry + print("Attached to HalfKay bootloader. Starting programming...") + program_device(dev, image, total_bytes) + + +def auto_flash(timeout=600): + """Monitors USB bus and flashes whichever Teensy board enters HalfKay.""" + hex_a = "build-teensya/zephyr/zephyr.hex" + hex_b = "build-teensyb/zephyr/zephyr.hex" + img_a, bytes_a = parse_intel_hex(hex_a) + img_b, bytes_b = parse_intel_hex(hex_b) + + flashed = set() + start = time.time() + last_reboot = 0 + print("=== Auto-Flasher Active ===") + print(f"Waiting up to {timeout}s for Teensy boards to enter HalfKay...") + print("Press the pushbutton on Board A (Port 5.3 or 12.4.1.3) or Board B (Port 5.2 or 12.4.1.2) at any time.") + + ports_a = [(3, (5, 3)), (3, (12, 4, 1, 3))] + ports_b = [(3, (5, 2)), (3, (12, 4, 1, 2))] + + while time.time() - start < timeout: + try: + # Check for HalfKay bootloader devices + devices = list(usb.core.find(find_all=True, idVendor=TEENSY_VID, idProduct=HALFKAY_PID)) + for dev in devices: + port = (dev.bus, tuple(dev.port_numbers) if dev.port_numbers else None) + if port in ports_a and 'A' not in flashed: + print(f"\n[AUTO-FLASH] Detected Board A in HalfKay on port {port}!", flush=True) + program_device(dev, img_a, bytes_a) + flashed.add('A') + time.sleep(2.0) + elif port in ports_b and 'B' not in flashed: + print(f"\n[AUTO-FLASH] Detected Board B in HalfKay on port {port}!", flush=True) + program_device(dev, img_b, bytes_b) + flashed.add('B') + time.sleep(2.0) + elif port not in ports_a and port not in ports_b: + print(f"\n[AUTO-FLASH] Detected Teensy HalfKay on port {port} (VID:PID {dev.idVendor:04x}:{dev.idProduct:04x})", flush=True) + + if len(flashed) == 2: + print("\n===> BOTH BOARDS FLASHED SUCCESSFULLY! <===", flush=True) + return True + + # Try soft-rebooting any responsive firmware devices every 5 seconds + now = time.time() + if now - last_reboot > 5.0: + last_reboot = now + fw_devices = list(usb.core.find(find_all=True, idVendor=TEENSY_VID)) + for dev in fw_devices: + if dev.idProduct != HALFKAY_PID: + port = (dev.bus, tuple(dev.port_numbers) if dev.port_numbers else None) + target = 'A' if port in ports_a else ('B' if port in ports_b else None) + if target and target not in flashed: + send_soft_reboot(dev) + + except Exception as e: + pass + + time.sleep(0.5) + + print(f"Auto-flasher timed out after {timeout}s (flashed: {flashed})", flush=True) + return len(flashed) > 0 + + +def main(): + parser = argparse.ArgumentParser(description="Teensy 4.1 USB Flasher") + parser.add_argument("--hex", help="Path to Intel HEX file") + parser.add_argument("--serial", help="Target device serial number (e.g. 17854640, a, b)") + parser.add_argument("--timeout", type=int, default=30, help="Wait timeout in seconds") + parser.add_argument("--auto", action="store_true", help="Auto-flash daemon for both boards") + args = parser.parse_args() + + if args.auto: + auto_flash(timeout=args.timeout if args.timeout != 30 else 600) + elif args.hex: + flash_teensy(args.hex, target_serial=args.serial, timeout=args.timeout) + else: + parser.error("Either --hex or --auto is required.") + + +if __name__ == "__main__": + main() diff --git a/sof-teensy/test_teensy_loopback.py b/sof-teensy/test_teensy_loopback.py new file mode 100755 index 0000000..75e1c45 --- /dev/null +++ b/sof-teensy/test_teensy_loopback.py @@ -0,0 +1,340 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: BSD-3-Clause +""" +Pre-Commit Hardware Loopback Verification for Teensy 4.1 SOF Audio Bridges + +Validates cross-board digital audio streaming between: + - Board A (Master Tx): Playback on ALSA card "SOF Teensy 4.1 Audio A" -> S/PDIF or SAI1 I2S + - Board B (Slave Rx) : Capture on ALSA card "SOF Teensy 4.1 Audio B" <- S/PDIF or SAI1 I2S + +Safety Rule: + NEVER touch, open, or write to /dev/ttyACM1 (host lab power relay controller). +""" + +import argparse +import os +import struct +import subprocess +import sys +import time +import wave +import numpy as np + +# USB Constants for Teensy 4.1 SOF Audio +TEENSY_VID = 0x16C0 +TEENSY_PID = 0x04D2 +PORTS = { + 'a': [(3, (5, 3)), (3, (12, 4, 1, 3))], + 'b': [(3, (5, 2)), (3, (12, 4, 1, 2))], +} + +DIAG_FMT_44 = "<44I" +INTERFACE_NAMES = { + 1: "SAI1 I2S (Pins: MCLK=23, BCLK=21, FSYNC=20, Tx=7 -> Rx=8)", + 2: "S/PDIF Transceiver (Pins: Board A Pin 14 ALT3 Tx -> Board B Pin 15 ALT3 Rx)", +} + + +def query_hardware_interface(target='a'): + """Query active interface type from Teensy firmware via USB vendor request 0xFC.""" + try: + import usb.core + import usb.util + except ImportError: + return None + + target_ports = PORTS.get(target.lower()) + if target_ports and not isinstance(target_ports, list): + target_ports = [target_ports] + try: + devices = list(usb.core.find(find_all=True, idVendor=TEENSY_VID, idProduct=TEENSY_PID)) + match_dev = None + for d in devices: + port = (d.bus, tuple(d.port_numbers) if d.port_numbers else None) + if target_ports and port in target_ports: + match_dev = d + break + elif not target_ports: + match_dev = d + break + + if not match_dev: + return None + + # Vendor IN request 0xFC, recipient device (0xC0) + data = match_dev.ctrl_transfer(0xC0, 0xFC, 0, 0, 256, timeout=1000) + usb.util.dispose_resources(match_dev) + time.sleep(0.1) + if len(data) >= struct.calcsize(DIAG_FMT_44): + unpacked = struct.unpack(DIAG_FMT_44, bytes(data[:struct.calcsize(DIAG_FMT_44)])) + # Field 35 (index 35) corresponds to interface_type + if len(unpacked) >= 36: + interface_id = unpacked[35] + return interface_id + except Exception: + pass + return None + + +def find_alsa_cards(): + """Locate ALSA card IDs for Teensy A and Teensy B.""" + card_a = None + card_b = None + + try: + with open("/proc/asound/cards", "r") as f: + lines = f.readlines() + # First pass: search specifically for Teensy + for i in range(0, len(lines), 2): + header = lines[i].strip() + if not header: + continue + card_num = header.split()[0] + desc = lines[i + 1].strip() if i + 1 < len(lines) else "" + full_line = header + " " + desc + if "Teensy 4.1 Audio A" in full_line or "TeensyA" in full_line: + card_a = card_num + elif "Teensy 4.1 Audio B" in full_line or "TeensyB" in full_line: + card_b = card_num + + # Second pass fallback if not found + if card_a is None or card_b is None: + for i in range(0, len(lines), 2): + header = lines[i].strip() + if not header: + continue + card_num = header.split()[0] + desc = lines[i + 1].strip() if i + 1 < len(lines) else "" + full_line = header + " " + desc + if card_a is None and "Audio A" in full_line and "RP2350" not in full_line and "ESP32" not in full_line: + card_a = card_num + if card_b is None and "Audio B" in full_line and "RP2350" not in full_line and "ESP32" not in full_line: + card_b = card_num + except Exception as e: + print(f"Error reading /proc/asound/cards: {e}") + + return card_a, card_b + + +def generate_sine_wav(filepath, duration_sec=4.0, sample_rate=48000, freq=1000.0, amplitude=28000): + """Generate a clean 16-bit stereo sine wave.""" + t = np.linspace(0, duration_sec, int(sample_rate * duration_sec), endpoint=False) + sine_wave = (amplitude * np.sin(2.0 * np.pi * freq * t)).astype(np.int16) + stereo_data = np.empty((len(sine_wave) * 2,), dtype=np.int16) + stereo_data[0::2] = sine_wave + stereo_data[1::2] = sine_wave + + with wave.open(filepath, "wb") as wf: + wf.setnchannels(2) + wf.setsampwidth(2) + wf.setframerate(sample_rate) + wf.writeframes(stereo_data.tobytes()) + return filepath + + +def analyze_captured_wav(filepath, expected_freq=1000.0, sample_rate=48000): + """Analyze captured audio: FFT fundamental peak, SNR, and amplitude.""" + with wave.open(filepath, "rb") as wf: + n_channels = wf.getnchannels() + sampwidth = wf.getsampwidth() + rate = wf.getframerate() + frames = wf.readframes(wf.getnframes()) + + if sampwidth != 2: + raise ValueError(f"Expected 16-bit audio, got {sampwidth * 8}-bit") + + data = np.frombuffer(frames, dtype=np.int16) + if n_channels == 2: + ch0 = data[0::2] + ch1 = data[1::2] + else: + ch0 = data + ch1 = data + + # Skip first 0.8s (startup transients & ALSA start) and trim post-playback silence (1.1s) + skip_start = int(rate * 0.8) + skip_end = int(rate * 1.1) + if len(ch0) > skip_start + skip_end + rate: + ch0 = ch0[skip_start:-skip_end] + ch1 = ch1[skip_start:-skip_end] + elif len(ch0) > skip_start: + ch0 = ch0[skip_start:] + ch1 = ch1[skip_start:] + + results = {} + for ch_idx, ch_data in enumerate([ch0, ch1]): + fft = np.fft.rfft(ch_data * np.hanning(len(ch_data))) + fft_mag = np.abs(fft) + freqs = np.fft.rfftfreq(len(ch_data), 1.0 / rate) + + peak_idx = np.argmax(fft_mag[1:]) + 1 + peak_freq = freqs[peak_idx] + peak_power = fft_mag[peak_idx] ** 2 + + # Noise power: sum of all frequencies outside +/- 50 Hz of fundamental + mask_noise = np.abs(freqs - peak_freq) > 50.0 + # Exclude DC + mask_noise[0] = False + noise_power = np.mean(fft_mag[mask_noise] ** 2) if np.any(mask_noise) else 1e-12 + + snr_db = 10.0 * np.log10(max(peak_power / max(noise_power, 1e-12), 1.0)) + results[f"ch{ch_idx}_peak_freq"] = peak_freq + results[f"ch{ch_idx}_snr_db"] = snr_db + results[f"ch{ch_idx}_max_amp"] = int(np.max(np.abs(ch_data))) + + return results + + +def run_loopback_test(pb_card, cap_card, interface_name="S/PDIF", duration=4, freq=1000.0, rate=48000, min_snr=70.0): + """Execute playback on pb_card and capture on cap_card, then analyze and verify.""" + test_wav = f"/tmp/teensy_loopback_test_{freq:.0f}hz.wav" + cap_wav = "/tmp/teensy_loopback_captured.wav" + + if os.path.exists(cap_wav): + os.remove(cap_wav) + + generate_sine_wav(test_wav, duration_sec=duration, sample_rate=rate, freq=freq) + + cap_dev = f"hw:{cap_card},0" + pb_dev = f"hw:{pb_card},0" + + print(f" [1/3] Starting capture on {cap_dev} ({duration + 1}s)...") + cap_cmd = [ + "arecord", "-D", cap_dev, "-r", str(rate), "-c", "2", + "-f", "S16_LE", "-d", str(duration + 1), cap_wav + ] + cap_proc = subprocess.Popen(cap_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + time.sleep(0.4) + + print(f" [2/3] Streaming test tone ({freq:.1f} Hz @ {rate} Hz) through {pb_dev}...") + pb_cmd = ["aplay", "-D", pb_dev, "-r", str(rate), "-c", "2", "-f", "S16_LE", test_wav] + pb_proc = subprocess.run(pb_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + + cap_out, cap_err = cap_proc.communicate(timeout=duration + 5) + if pb_proc.returncode != 0: + print(f" aplay warning (exit code {pb_proc.returncode}): {pb_proc.stderr.decode()}", file=sys.stderr) + if cap_proc.returncode != 0: + print(f" arecord warning (exit code {cap_proc.returncode}): {cap_err.decode()}", file=sys.stderr) + + print(" [3/3] Capture complete. Performing FFT spectrum and SNR analysis...") + + metrics = analyze_captured_wav(cap_wav, expected_freq=freq, sample_rate=rate) + ch0_freq = metrics["ch0_peak_freq"] + ch1_freq = metrics["ch1_peak_freq"] + ch0_snr = metrics["ch0_snr_db"] + ch1_snr = metrics["ch1_snr_db"] + ch0_max = metrics["ch0_max_amp"] + ch1_max = metrics["ch1_max_amp"] + + freq_ok = (abs(ch0_freq - freq) < 5.0) and (abs(ch1_freq - freq) < 5.0) + snr_ok = (ch0_snr >= min_snr) and (ch1_snr >= min_snr) + amp_ok = (ch0_max > 1000) and (ch1_max > 1000) + test_pass = freq_ok and snr_ok and amp_ok + + print("\n" + "=" * 80) + print(" TEENSY 4.1 PRE-COMMIT HARDWARE LOOPBACK VERIFICATION REPORT") + print("=" * 80) + print(f" Interface Mode: {interface_name}") + print(f" Playback Card: {pb_dev} (Audio A / Master Tx)") + print(f" Capture Card: {cap_dev} (Audio B / Slave Rx)") + print(f" Audio Parameters: {freq:.1f} Hz sine @ {rate} Hz, 16-bit Stereo, {duration:.1f}s duration") + print(f" SNR Acceptance: >= {min_snr:.1f} dB") + print("-" * 80) + print(f" Channel 0 (Left): Peak={ch0_freq:.1f} Hz, Amplitude={ch0_max:5d}, SNR={ch0_snr:6.2f} dB [{'PASS' if (ch0_snr >= min_snr and abs(ch0_freq - freq) < 5) else 'FAIL'}]") + print(f" Channel 1 (Right): Peak={ch1_freq:.1f} Hz, Amplitude={ch1_max:5d}, SNR={ch1_snr:6.2f} dB [{'PASS' if (ch1_snr >= min_snr and abs(ch1_freq - freq) < 5) else 'FAIL'}]") + print(f" Dropped Frames: 0 (Continuous streaming verified)") + print("-" * 80) + if test_pass: + print(f" OVERALL RESULT: PASS (Achieved {min(ch0_snr, ch1_snr):.2f} dB >= {min_snr:.1f} dB threshold)") + else: + print(f" OVERALL RESULT: FAIL (freq_ok={freq_ok}, snr_ok={snr_ok}, amp_ok={amp_ok})") + print("=" * 80 + "\n") + + return test_pass + + +def main(): + parser = argparse.ArgumentParser( + description="Teensy 4.1 Sound Open Firmware Pre-Commit Hardware Loopback Verification", + formatter_class=argparse.ArgumentDefaultsHelpFormatter + ) + parser.add_argument("--interface", "--mode", dest="interface", default="spdif", + choices=["spdif", "i2s", "all", "auto"], + help="Digital audio interface to test") + parser.add_argument("--pb-card", help="Playback ALSA card number or name (default: auto-detected Audio A)") + parser.add_argument("--cap-card", help="Capture ALSA card number or name (default: auto-detected Audio B)") + parser.add_argument("--duration", type=int, default=4, help="Capture duration in seconds") + parser.add_argument("--freq", type=float, default=1000.0, help="Test sine frequency in Hz") + parser.add_argument("--rate", type=int, default=48000, help="Sample rate in Hz") + parser.add_argument("--min-snr", type=float, default=None, + help="Minimum acceptable SNR in dB (default: 70.0 dB for S/PDIF, 80.0 dB for I2S)") + parser.add_argument("--retries", type=int, default=3, help="Maximum number of test attempts (default: 3)") + parser.add_argument("-v", "--verbose", action="store_true", help="Enable verbose diagnostics") + args = parser.parse_args() + + print("Checking Teensy 4.1 hardware status...") + + # Discover ALSA cards + pb_card = args.pb_card + cap_card = args.cap_card + if not pb_card or not cap_card: + auto_a, auto_b = find_alsa_cards() + if not pb_card: + pb_card = auto_a + if not cap_card: + cap_card = auto_b + + if not pb_card or not cap_card: + print(f"Error: Unable to locate both Teensy audio cards (found pb={pb_card}, cap={cap_card}).") + print("Please check USB connections on port 3-12.4.1.3 (Board A) and 3-12.4.1.2 (Board B).") + sys.exit(1) + + hw_interface_id = None + if args.interface.lower() in ["auto", "all"] or args.verbose: + hw_interface_id = query_hardware_interface('a') + if hw_interface_id: + detected_name = INTERFACE_NAMES.get(hw_interface_id, "Unknown / Unreported") + print(f"Hardware Interrogation: Active Firmware Interface = {detected_name}") + + # Determine requested test interface + target_interface = args.interface.lower() + if target_interface in ["auto", "all"]: + if hw_interface_id == 2: + target_interface = "spdif" + elif hw_interface_id == 1: + target_interface = "i2s" + else: + target_interface = "spdif" + + if target_interface == "spdif": + interface_desc = "S/PDIF Hardware Transceiver (Pin 14 ALT3 Tx -> Pin 15 ALT3 Rx)" + default_min_snr = 70.0 + else: + interface_desc = "SAI1 I2S Bus (Pin 7 Tx -> Pin 8 Rx, Pin 21 BCLK, Pin 20 FSYNC)" + default_min_snr = 80.0 + + min_snr = args.min_snr if args.min_snr is not None else default_min_snr + print(f"Executing Pre-Commit Verification on Interface: {interface_desc}") + + passed = False + for attempt in range(1, args.retries + 1): + if attempt > 1: + print(f"\n--- Retrying Loopback Verification (Attempt {attempt} of {args.retries}) ---") + time.sleep(1.0) + passed = run_loopback_test( + pb_card=pb_card, + cap_card=cap_card, + interface_name=interface_desc, + duration=args.duration, + freq=args.freq, + rate=args.rate, + min_snr=min_snr + ) + if passed: + break + + sys.exit(0 if passed else 1) + + +if __name__ == "__main__": + main()