path: root/test/hil/hil_test.py

#!/usr/bin/env python3
#
# The MIT License (MIT)
#
# Copyright (c) 2023 HiFiPhile
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.

# Host setup (required: a missing tool fails its test rather than skipping it):
#   - System packages: sudo apt install mtools libmtp9 alsa-utils iperf
#       mtools     - read_disk_file (device/cdc_msc, device/msc_dual_lun)
#       libmtp9    - pymtp ctypes load (device/mtp); Debian 13 uses libmtp9t64
#       alsa-utils - arecord (device/audio_test_freertos)
#       iperf      - throughput tests (device/net_lwip_*)
#   - device/usbtest: usbtest kernel module + testusb binary (kernel tools/usb/testusb.c) on PATH,
#     plus sudo for modprobe / sysfs writes
#   - Python packages: pip install -r requirements.txt
#
# udev rules :
# ACTION=="add", SUBSYSTEM=="tty", SUBSYSTEMS=="usb", MODE="0666", PROGRAM="/bin/sh -c 'echo $$ID_SERIAL_SHORT | rev | cut -c -8 | rev'", SYMLINK+="ttyUSB_%c.%s{bInterfaceNumber}"
# ACTION=="add", SUBSYSTEM=="block", SUBSYSTEMS=="usb", ENV{ID_FS_USAGE}=="filesystem", MODE="0666", PROGRAM="/bin/sh -c 'echo $$ID_SERIAL_SHORT | rev | cut -c -8 | rev'", RUN{program}+="/usr/bin/systemd-mount --no-block --automount=yes --collect $devnode /media/blkUSB_%c.%s{bInterfaceNumber}"

import argparse
import io
import itertools
import os
import random
import re
import select
import sys
import time
import signal
from contextlib import redirect_stdout
from pathlib import Path
from typing import Any, TypedDict, NotRequired, cast

import serial
import subprocess
import json
import glob
import multiprocessing
from multiprocessing import TimeoutError as MpTimeoutError

# Raw Lock/Semaphore objects passed via Pool initargs are inheritable only under the fork
# start method (spawn/forkserver pickle them and fail at Pool creation) — pin it so a
# future interpreter default change cannot break the run at startup.
_mp = multiprocessing.get_context('fork')
Pool, Lock, Semaphore, Manager = _mp.Pool, _mp.Lock, _mp.Semaphore, _mp.Manager
import hashlib
import ctypes
from pymtp import MTP
import string

# --- per-board dev-session locks (see test/hil/board_lock.py) ------------
BOARD_LOCK_DIR = '/tmp/tinyusb-hil-locks'

def acquire_board_lock(board_name):
    """Take this board's flock for the duration of its flash+test.
    Returns an open file handle (keep it referenced; closing releases it),
    or None when HIL_NO_BOARD_LOCK=1 or the lock dir is unusable (fail-open:
    locking must never break a test run by itself).
    Raises RuntimeError only when another session holds the board."""
    import fcntl
    if os.environ.get('HIL_NO_BOARD_LOCK') == '1':
        return None  # user-authorized bypass — see board_lock.py / hil skill
    try:
        os.makedirs(BOARD_LOCK_DIR, exist_ok=True)
        fd = os.open(os.path.join(BOARD_LOCK_DIR, f'{board_name}.lock'),
                     os.O_RDWR | os.O_CREAT, 0o666)
        fh = os.fdopen(fd, 'r+')
    except OSError as e:
        # odd lock dir (perms, path collision): proceed unlocked, but say so —
        # a silent fail-open is indistinguishable from the intentional bypass
        print(f'warning: board lock unavailable for {board_name} ({e}); proceeding unlocked',
              flush=True)
        return None
    try:
        fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except OSError:
        try:
            info = fh.read(500).strip()
        except (OSError, UnicodeDecodeError):
            info = ''
        fh.close()
        raise RuntimeError(f'board locked: {info or "unknown holder"}')
    # announce ourselves so the other side's conflict message is truthful;
    # best-effort — the flock itself is already held
    try:
        fh.truncate(0)
        fh.seek(0)
        json.dump({'pid': os.getpid(), 'reason': 'hil_test.py',
                   'since': time.strftime('%Y-%m-%dT%H:%M:%S%z')}, fh)
        fh.flush()
    except OSError:
        pass
    return fh


# Enumeration wait budget. The first attempt gets ENUM_TIMEOUT; retry attempts get the
# shorter ENUM_TIMEOUT_RETRY - the board was just re-flashed again, and a device that is
# going to enumerate shows up within a few seconds, so a failing test costs ~3-5x a
# passing one instead of 10-30x. Per-attempt value is set by test_example(); each pool
# worker is its own process, so a module global is safe.
ENUM_TIMEOUT = 8
ENUM_TIMEOUT_RETRY = 4
_enum_timeout = ENUM_TIMEOUT


def enum_timeout() -> int:
    """Enumeration wait budget for the current test attempt."""
    return _enum_timeout


def wait_until(predicate, step: float = 1.0):
    """Poll predicate under the per-attempt enum budget. Deadline-based so a slow predicate
    body (subprocess, libmtp scan) counts against the budget. Returns the first truthy
    predicate value, or None on timeout."""
    deadline = time.monotonic() + enum_timeout()
    while True:
        r = predicate()
        if r:
            return r
        if time.monotonic() >= deadline:
            return None
        time.sleep(step)

STATUS_OK = "\033[32mOK\033[0m"
STATUS_FAILED = "\033[31mFailed\033[0m"
STATUS_SKIPPED = "\033[33mSkipped\033[0m"

# Plain (non-ANSI) cell symbols for the markdown matrix report (hil_report.md).
# A missing binary is reported as skipped too.
REPORT_CELL = {'pass': '✅', 'fail': '❌', 'skip': '⚪'}


class TestFail(AssertionError):
    """Fail a test but still surface a metric string in its report cell (e.g. usbtest's '❌ 29/30'
    instead of a bare ❌). The cell metric is icon-prefixed so render/tally treat it as a failure."""
    def __init__(self, msg: str, metric: str | None = None):
        super().__init__(msg)
        self.metric = metric


verbose = False
PROFILE = os.environ.get('HIL_PROFILE') == '1'  # timestamped logs + permit/flash timing + ctrl-map dump
test_only = []
board_test = {}
build_dir = 'cmake-build'
skip_flash = False
print_lock = None
shuffle_seed = None  # per-run seed for the per-board test-order shuffle (HIL_SHUFFLE_SEED to replay)

# Per-host-controller concurrency (see controller_of/controller_slot below): a usbtest battery
# saturates its DUT's host controller, so batteries and flashes are budgeted per controller.
# - uPD720201 cards need their latest firmware (>= 2.0.2.6; RAM-uploaded, reloads every
#   power cycle): ROM firmware dies under battery + re-enumeration churn, and usbtest.py
#   refuses the unlink-stress cases on it.
# - widths (profiled 2026-07-13/14): wall time 22.2/14.3/12.5/10.8 min at usbtest width
#   1/2/3/4, plateau after; flash width beyond 8 only adds flasher-hub contention;
#   battery case failures start at 12/8 (bandwidth stretch on shared leaf-hub uplinks).
# - a marginal DUT port bouncing during concurrent batteries can wedge/kill a uPD720201
#   ("xHCI host not responding to stop endpoint command"): fix the port/cable or pull
#   the board, don't lower the widths (2026-07-16: every death traced to one board's port).
FLASH_PARALLEL = int(os.getenv('HIL_FLASH_PARALLEL', '8'))
USBTEST_PARALLEL = int(os.getenv('HIL_USBTEST_PARALLEL', '4'))
CONTROLLER_SLOTS = 12  # lock slots; controllers are assigned to slots on first sight
usbtest_sems = None  # CONTROLLER_SLOTS semaphores: per-slot usbtest-battery permits
flash_sems = None       # CONTROLLER_SLOTS semaphores: per-slot flash permits
controller_map = None         # shared dict: 'pci:<addr>' -> slot, 'uid:<uid>' -> pci addr cache
controller_meta = None        # guards slot assignment in controller_map
controller_hints = {}         # static uid -> pci from the last run's cache (read-only per worker)


def init_worker(lock, seed, b_mutexes, f_sems, cmap, cmeta, hints_by_uid):
    global print_lock, shuffle_seed, usbtest_sems, flash_sems, controller_map, controller_meta, controller_hints
    print_lock = lock
    shuffle_seed = seed
    usbtest_sems = b_mutexes
    flash_sems = f_sems
    controller_map = cmap
    controller_meta = cmeta
    controller_hints = hints_by_uid


def log_line(msg: str) -> None:
    if PROFILE:
        msg = f'{time.time():.3f} {msg}'
    out = sys.__stdout__ if sys.__stdout__ is not None else sys.stdout
    if print_lock is not None:
        with print_lock:
            print(msg, file=out, flush=True)
    else:
        print(msg, file=out, flush=True)


# -------------------------------------------------------------
# Per-controller scheduling
# -------------------------------------------------------------
def controller_of(uid: str):
    """Resolve a DUT uid to its root host controller's PCI address, or None if the device
    is not enumerated (e.g. parked in board_test firmware with USB off). Successful
    resolutions are cached — cabling does not change mid-run. Dual-port parts (e.g.
    CH32V307 usbhs/usbfs variants) share one uid and one cache entry: budgeting is only
    exact when both ports sit on the same controller (true on this rig)."""
    if controller_map is None:
        return None
    cached = controller_map.get(f'uid:{uid}')
    if cached:
        return cached
    for f in glob.glob('/sys/bus/usb/devices/*/serial'):
        d = os.path.dirname(f)
        try:
            if open(f).read().strip().lower() != uid.lower():
                continue
            bus = int(open(os.path.join(d, 'busnum')).read())
            root = os.path.realpath(f'/sys/bus/usb/devices/usb{bus}')
            m = re.findall(r'[0-9a-f]{4}:[0-9a-f]{2}:[0-9a-f]{2}\.[0-9a-f]', root)
            if m:
                controller_map[f'uid:{uid}'] = m[-1]
                return m[-1]
        except (OSError, ValueError):
            continue
    return None


def controller_slot(pci: str) -> int:
    """Map a controller PCI address to a lock slot (assigned on first sight)."""
    key = f'pci:{pci}'
    with controller_meta:
        slot = controller_map.get(key)
        if slot is None:
            slot = controller_map.get('nslots', 0)
            if slot >= CONTROLLER_SLOTS:
                slot = 0  # more controllers than slots: overflow shares slot 0 (safe, over-serialized)
            else:
                controller_map['nslots'] = slot + 1
            controller_map[key] = slot
        return slot


class controller_permit:
    """Context manager: one permit from `sems` on the board's controller slot. If the
    controller is unknown, fail closed: take one permit from EVERY slot, in order, so the
    operation respects the budget wherever it might land. `warn_unknown` logs that fallback
    (used by usbtest, where the device is expected to be enumerated by the caller)."""
    def __init__(self, sems, uid: str, warn_unknown: bool = False):
        self.sems = sems
        self.slots = None
        self.uid = uid
        if sems is None:
            return
        pci = controller_of(uid)
        if pci is None and not warn_unknown:
            # last-run cabling hint, flash budgeting only: a mis-budgeted flash is harmless,
            # but a battery must never trust a stale hint (it could stack two batteries on
            # one controller). In practice only a board's first flash lands here - batteries
            # assert enumeration before taking their permit.
            pci = controller_hints.get(uid)
        if pci is None and warn_unknown:
            log_line(f'warning: cannot resolve {uid} to a host controller; '
                     'taking a permit on every slot (over-serialized)')
        self.slots = [controller_slot(pci)] if pci else list(range(CONTROLLER_SLOTS))

    def __enter__(self):
        if self.slots:
            t0 = time.monotonic()
            taken = []
            try:
                for s in self.slots:
                    self.sems[s].acquire()
                    taken.append(s)
                # stays inside the try: if this raises (e.g. broken stdout), the permits
                # must be released - a failed __enter__ never gets its __exit__
                if PROFILE and time.monotonic() - t0 > 1.0:
                    log_line(f'[prof] permit wait {time.monotonic() - t0:.1f}s '
                             f'(uid {self.uid}, slots {self.slots})')
            except BaseException:
                for s in reversed(taken):
                    self.sems[s].release()
                raise
        return self

    def __exit__(self, *exc):
        if self.slots:
            for s in reversed(self.slots):
                self.sems[s].release()
        return False


def flash_permit(uid: str) -> controller_permit:
    return controller_permit(flash_sems, uid)


def usbtest_permit(uid: str) -> controller_permit:
    return controller_permit(usbtest_sems, uid, warn_unknown=True)


def compact_output(raw: str) -> str:
    if not raw:
        return ''
    lines = [ln.strip() for ln in raw.replace('\r', '\n').split('\n') if ln.strip()]
    return ' | '.join(lines)

class FlasherCfg(TypedDict):
    name: str
    uid: str
    args: str


class AttachedDevCfg(TypedDict, total=False):
    vid_pid: str
    serial: str
    is_cdc: bool
    is_msc: bool
    block_count: int
    block_size: int


class TestsCfg(TypedDict, total=False):
    device: bool
    dual: bool
    host: bool
    only: list[str]
    skip: list[str]
    dev_attached: list[AttachedDevCfg]


class BuildCfg(TypedDict, total=False):
    args: list[str]


class VariantCfg(TypedDict, total=False):
    name: str           # build dir (cmake-build-<name>) and HIL report row
    flags: str          # raw CFLAGS, e.g. "-DCFG_TUD_DWC2_DMA_ENABLE=1"
    defines: list[str]  # cmake -D defines, e.g. ["RHPORT_DEVICE=1"] (vs flags which are compiler-only)


class Board(TypedDict):
    name: str
    uid: str
    tests: TestsCfg
    flasher: FlasherCfg
    build: NotRequired[BuildCfg]
    variant: NotRequired[list[VariantCfg]]
    toolchain: NotRequired[str]  # CI build bucket override, e.g. "riscv-gcc" (consumed by hil_ci_set_matrix.py)


class HilConfig(TypedDict):
    boards: list[Board]

CMD_TIMEOUT = int(os.getenv('HIL_CMD_TIMEOUT', '180'))
POOL_TIMEOUT = int(os.getenv('HIL_POOL_TIMEOUT', '4200'))  # usbtest batteries are serialized fleet-wide, lengthening the tail
SERIAL_READ_TIMEOUT = float(os.getenv('HIL_SERIAL_READ_TIMEOUT', '5'))
SERIAL_WRITE_TIMEOUT = float(os.getenv('HIL_SERIAL_WRITE_TIMEOUT', '10'))


def cmd_stdout_text(out: Any) -> str:
    if out is None:
        return ''
    if isinstance(out, bytes):
        return out.decode('utf-8', errors='ignore')
    return str(out)

WCH_RISCV_CONTENT = """
adapter driver wlinke
adapter speed 6000
transport select sdi

wlink_set_address 0x00000000
set _CHIPNAME wch_riscv
sdi newtap $_CHIPNAME cpu -irlen 5 -expected-id 0x00001

set _TARGETNAME $_CHIPNAME.cpu

target create $_TARGETNAME.0 wch_riscv -chain-position $_TARGETNAME
$_TARGETNAME.0 configure  -work-area-phys 0x20000000 -work-area-size 10000 -work-area-backup 1
set _FLASHNAME $_CHIPNAME.flash

flash bank $_FLASHNAME wch_riscv 0x00000000 0 0 0 $_TARGETNAME.0

echo "Ready for Remote Connections"
"""

MSC_README_TXT = \
b"This is tinyusb's MassStorage Class demo.\r\n\r\n\
If you find any bugs or get any questions, feel free to file an\r\n\
issue at github.com/hathach/tinyusb"

# -------------------------------------------------------------
# Path
# -------------------------------------------------------------
OPENCOD_ADI_PATH = Path.home() / 'app' / 'openocd_adi'
TINYUSB_ROOT = Path(__file__).resolve().parents[2]

# get usb serial by id
def get_serial_dev(id, vendor_str, product_str, ifnum):
    if vendor_str and product_str:
        # known vendor and product
        vendor_str = vendor_str.replace(' ', '_')
        product_str = product_str.replace(' ', '_')
        return f'/dev/serial/by-id/usb-{vendor_str}_{product_str}_{id}-if{ifnum:02d}'
    else:
        # just use id: mostly for cp210x/ftdi flasher
        pattern = f'/dev/serial/by-id/usb-*_{id}-if*'
        port_list = glob.glob(pattern)
        if len(port_list) == 0:
            raise RuntimeError(f'No serial device found for {pattern}')
        return port_list[0]


# get usb disk by id
def get_disk_dev(id, vendor_str, lun):
    return f'/dev/disk/by-id/usb-{vendor_str}_Mass_Storage_{id}-0:{lun}'


def get_hid_dev(id, vendor_str, product_str, event):
    return f'/dev/input/by-id/usb-{vendor_str}_{product_str}_{id}-{event}'


def get_alsa_capture_dev(id):
    pattern = f'/dev/snd/by-id/usb-*_{id}-*'
    for dev in glob.glob(pattern):
        try:
            link = os.path.basename(os.path.realpath(dev))
        except OSError:
            continue
        m = re.match(r'controlC(\d+)', link)
        if m:
            return f'hw:{m.group(1)},0'
    return None


def open_serial_dev(port: str):
    timeout = enum_timeout()
    ser = None
    while timeout > 0:
        if os.path.exists(port):
            try:
                # write_timeout: a wedged device otherwise blocks ser.write() forever,
                # hanging the worker until the pool/job timeout kills the whole run
                ser = serial.Serial(port, baudrate=115200, timeout=SERIAL_READ_TIMEOUT,
                                    write_timeout=SERIAL_WRITE_TIMEOUT)
                break
            except serial.SerialException:
                print(f'serial {port} not reaady {timeout} sec')
                pass
        time.sleep(0.1)
        timeout -= 0.1

    assert timeout > 0, f'Cannot open port f{port}' if os.path.exists(port) else f'Port {port} not existed'
    assert ser is not None
    return ser


def serial_write_all(ser: serial.Serial, data: bytes):
    # write_timeout is a total deadline for the whole call (pyserial keeps partial progress
    # internally). A timeout means the device stopped draining — treat it as fatal: pyserial
    # loses the partial-write count on raise, so retrying would duplicate bytes on the wire.
    try:
        ser.write(data)
    except serial.SerialTimeoutException:
        raise AssertionError(f'Serial write timeout after {SERIAL_WRITE_TIMEOUT:.1f}s')


def read_disk_file(uid: str, lun: int, fname: str) -> bytes:
    # Reads a file from a FAT volume on a block device without mounting it.
    # Requires mtools: `apt install mtools` (no pip dependency).
    dev = get_disk_dev(uid, 'TinyUSB', lun)
    last_err = None

    def try_read():
        nonlocal last_err
        if not os.path.exists(dev):
            return None
        try:
            data = subprocess.check_output(
                ['mtype', '-i', dev, f'::/{fname}'], stderr=subprocess.PIPE)
            assert data, f'Cannot read file {fname} from {dev}'
            return data
        except subprocess.CalledProcessError as e:
            last_err = e.stderr.decode(errors='replace').strip()
            return None

    data = wait_until(try_read)
    if data is None:
        raise AssertionError(f'mtype failed on {dev}: {last_err}' if last_err else f'Storage {dev} not existed')
    return data


def open_mtp_dev(uid):
    mtp = MTP()

    def try_open():
        # unmount gio/gvfs MTP mount which blocks libmtp from accessing the device
        subprocess.run(f"gio mount -u mtp://TinyUsb_TinyUsb_Device_{uid}/",
                       shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        for raw in mtp.detect_devices():
            mtp.device = mtp.mtp.LIBMTP_Open_Raw_Device(ctypes.byref(raw))
            if mtp.device:
                sn = mtp.get_serialnumber().decode('utf-8')
                if sn == uid:
                    return mtp
                mtp.disconnect()
        return None

    return wait_until(try_open)


def get_printer_dev(id: str, vendor_str, product_str, ifnum: int):
    """Find /dev/usb/lpX by matching USB serial, vendor, product, and interface number via sysfs"""
    vendor_str = vendor_str.replace(' ', '_') if vendor_str else ''
    product_str = product_str.replace(' ', '_') if product_str else ''
    for lp in glob.glob('/sys/class/usbmisc/lp*'):
        try:
            sn = open(f'{lp}/device/../serial').read().strip()
            if sn == id:
                return f'/dev/usb/{os.path.basename(lp)}'
        except (FileNotFoundError, PermissionError, ValueError):
            pass
    return None


def open_printer_dev(id: str, vendor_str, product_str, ifnum: int) -> str:
    """Wait for printer device to enumerate and return its path"""
    def try_find():
        lp_dev = get_printer_dev(id, vendor_str, product_str, ifnum)
        return lp_dev if lp_dev and os.path.exists(lp_dev) else None

    lp_dev = wait_until(try_find)
    assert lp_dev, f'Printer device not found for {id} if{ifnum:02d}'
    return lp_dev


# -------------------------------------------------------------
# Flashing firmware
# -------------------------------------------------------------
def run_cmd(cmd: str, cwd: str | None = None, timeout: int = CMD_TIMEOUT) -> subprocess.CompletedProcess:
    popen_kwargs = {
        'cwd': cwd,
        'shell': True,
        'stdout': subprocess.PIPE,
        'stderr': subprocess.STDOUT,
        'text': True,
        'encoding': 'utf-8',
        'errors': 'replace',
    }
    if os.name != 'nt':
        popen_kwargs['preexec_fn'] = os.setsid

    p = subprocess.Popen(cmd, **popen_kwargs)
    try:
        out, _ = p.communicate(timeout=timeout)
        r = subprocess.CompletedProcess(args=cmd, returncode=p.returncode, stdout=out)
    except subprocess.TimeoutExpired as ex:
        if os.name != 'nt':
            try:
                os.killpg(p.pid, signal.SIGKILL)
            except ProcessLookupError:
                pass
        else:
            p.kill()
        out, _ = p.communicate()
        timeout_out = ex.stdout or out or b''
        title = f'COMMAND TIMEOUT ({timeout}s): {cmd}'
        print()
        if os.getenv('CI'):
            print(f"::group::{title}")
            print(cmd_stdout_text(timeout_out))
            print(f"::endgroup::")
        else:
            print(title)
            print(cmd_stdout_text(timeout_out))
        return subprocess.CompletedProcess(args=cmd, returncode=124, stdout=timeout_out)

    if r.returncode != 0:
        title = f'COMMAND FAILED: {cmd}'
        print()
        if os.getenv('CI'):
            print(f"::group::{title}")
            print(cmd_stdout_text(r.stdout))
            print(f"::endgroup::")
        else:
            print(title)
            print(cmd_stdout_text(r.stdout))
    elif verbose:
        print(cmd)
        print(cmd_stdout_text(r.stdout))
    return r


def flash_jlink(board: Board, firmware: str) -> subprocess.CompletedProcess:
    flasher = board['flasher']
    script = ['halt', 'r', f'loadfile {firmware}.elf', 'r', 'go', 'exit']
    f_jlink = Path(f'{board["name"]}_{Path(firmware).name}.jlink')
    with f_jlink.open('w') as f:
        f.writelines(f'{s}\n' for s in script)
    ret = run_cmd(f'JLinkExe -USB {flasher["uid"]} {flasher["args"]} -if swd -JTAGConf -1,-1 -speed auto -NoGui 1 -ExitOnError 1 -CommandFile {f_jlink}')
    f_jlink.unlink(missing_ok=True)
    return ret


def reset_jlink(board: Board) -> subprocess.CompletedProcess:
    flasher = board['flasher']
    script = ['halt', 'r', 'go', 'exit']
    f_jlink = Path(f'{board["name"]}_reset.jlink')
    if not f_jlink.exists():
        with f_jlink.open('w') as f:
            f.writelines(f'{s}\n' for s in script)
    ret = run_cmd(f'JLinkExe -USB {flasher["uid"]} {flasher["args"]} -if swd -JTAGConf -1,-1 -speed auto -NoGui 1 -ExitOnError 1 -CommandFile {f_jlink}')
    return ret


def flash_stlink(board, firmware):
    flasher = board['flasher']
    return run_cmd(f'STM32_Programmer_CLI --connect port=swd sn={flasher["uid"]} --write {firmware}.elf --go')


def reset_stlink(board):
    flasher = board['flasher']
    return run_cmd(f'STM32_Programmer_CLI --connect port=swd sn={flasher["uid"]} --rst --go')

def flash_stflash(board, firmware):
    flasher = board['flasher']
    ret = run_cmd(f'st-flash --serial {flasher["uid"]} write {firmware}.bin 0x8000000')
    return ret


def reset_stflash(board):
    flasher = board['flasher']
    return subprocess.CompletedProcess(args=['dummy'], returncode=0)


def flash_openocd(board, firmware):
    flasher = board['flasher']
    ret = run_cmd(f'openocd -c "tcl_port disabled" -c "gdb_port disabled" -c "adapter serial {flasher["uid"]}" '
                  f'{flasher["args"]} -c "init; halt; program {firmware}.elf verify; reset; exit"')
    return ret


def reset_openocd(board):
    flasher = board['flasher']
    ret = run_cmd(f'openocd -c "tcl_port disabled" -c "gdb_port disabled" -c "adapter serial {flasher["uid"]}" '
                  f'{flasher["args"]} -c "init; reset run; exit"')
    return ret


def flash_openocd_wch(board, firmware):
    flasher = board['flasher']
    f_wch = f"wch-riscv_{board['uid']}.cfg"
    if not os.path.exists(f_wch):
        with open(f_wch, 'w') as file:
            file.write(WCH_RISCV_CONTENT)

    ret = run_cmd(f'openocd_wch -c "adapter serial {flasher["uid"]}" -f {f_wch} '
                  f'-c "program {firmware}.elf reset exit"')
    return ret


def reset_openocd_wch(board):
    flasher = board['flasher']
    f_wch = f"wch-riscv_{board['uid']}.cfg"
    if not os.path.exists(f_wch):
        with open(f_wch, 'w') as file:
            file.write(WCH_RISCV_CONTENT)

    ret = run_cmd(f'openocd_wch -c "adapter serial {flasher["uid"]}" -f {f_wch} -c "program reset exit"')
    return ret


def flash_openocd_adi(board: Board, firmware: str) -> subprocess.CompletedProcess:
    flasher = board['flasher']
    openocd = OPENCOD_ADI_PATH / 'src' / 'openocd'
    tcl_dir = OPENCOD_ADI_PATH / 'tcl'
    ret = run_cmd(f'{openocd} -c "adapter serial {flasher["uid"]}" -s {tcl_dir} '
                  f'{flasher["args"]} -c "program {firmware}.elf reset exit"')
    return ret


def reset_openocd_adi(board: Board) -> subprocess.CompletedProcess:
    flasher = board['flasher']
    openocd = OPENCOD_ADI_PATH / 'src' / 'openocd'
    tcl_dir = OPENCOD_ADI_PATH / 'tcl'
    ret = run_cmd(f'{openocd} -c "adapter serial {flasher["uid"]}" -s {tcl_dir} '
                  f'{flasher["args"]} -c "program reset exit"')
    return ret


def flash_wlink_rs(board, firmware):
    flasher = board['flasher']
    # wlink use index for probe selection and lacking usb serial support
    ret = run_cmd(f'wlink flash {firmware}.elf')
    return ret


def reset_wlink_rs(board):
    flasher = board['flasher']
    # wlink use index for probe selection and lacking usb serial support
    ret = run_cmd(f'wlink reset')
    return ret


def flash_esptool(board: Board, firmware: str) -> subprocess.CompletedProcess:
    flasher = board['flasher']
    port = get_serial_dev(flasher["uid"], None, None, 0)
    fw_dir = Path(f'{firmware}.bin').parent
    with (fw_dir / 'config.env').open() as f:
        idf_target = json.load(f)['IDF_TARGET']
    with (fw_dir / 'flash_args').open() as f:
        flash_args = f.read().strip().replace('\n', ' ')
    command = (f'esptool --chip {idf_target} -p {port} {flasher["args"]} '
               f'--before=default_reset --after=hard_reset write_flash {flash_args}')
    ret = run_cmd(command, cwd=str(fw_dir))
    return ret


def reset_esptool(board):
    flasher = board['flasher']
    return subprocess.CompletedProcess(args=['dummy'], returncode=0)


def flash_uniflash(board, firmware):
    flasher = board['flasher']
    ret = run_cmd(f'dslite.sh {flasher["args"]} -f {firmware}.hex')
    return ret


def reset_uniflash(board):
    flasher = board['flasher']
    return subprocess.CompletedProcess(args=['dummy'], returncode=0)


def flash_lm4flash(board, firmware):
    # TI Tiva-C / Stellaris ICDI: lightweight lm4flash, resets and runs after write
    flasher = board['flasher']
    ret = run_cmd(f'lm4flash -s {flasher["uid"]} {flasher["args"]} {firmware}.bin')
    return ret


def reset_lm4flash(board):
    # lm4flash has no reset-only mode; it resets+runs on flash, so reset is a no-op
    flasher = board['flasher']
    return subprocess.CompletedProcess(args=['dummy'], returncode=0)


# -------------------------------------------------------------
# Tests: dual
# -------------------------------------------------------------
def test_dual_host_info_to_device_cdc(board):
    uid = board['uid']
    declared_devs = [f'{d["vid_pid"]}_{d["serial"]}' for d in board['tests']['dev_attached']]
    port = get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
    ser = open_serial_dev(port)
    ser.timeout = 0.1

    # read until all expected devices are enumerated
    data = b''
    timeout = enum_timeout()
    while timeout > 0:
        new_data = ser.read(ser.in_waiting or 1)
        if new_data:
            data += new_data
        # check if all devices found
        enum_dev_sn = []
        for l in data.decode('utf-8', errors='ignore').splitlines():
            vid_pid_sn = re.search(r'ID ([0-9a-fA-F]+):([0-9a-fA-F]+) SN (\w+)', l)
            if vid_pid_sn:
                enum_dev_sn.append(f'{vid_pid_sn.group(1)}_{vid_pid_sn.group(2)}_{vid_pid_sn.group(3)}')
        if set(declared_devs).issubset(set(enum_dev_sn)):
            break
        time.sleep(0.1)
        timeout -= 0.1
    ser.close()

    if len(data) == 0:
        assert False, 'No data from device'
    lines = data.decode('utf-8', errors='ignore').splitlines()

    enum_dev_sn = []
    for l in lines:
        vid_pid_sn = re.search(r'ID ([0-9a-fA-F]+):([0-9a-fA-F]+) SN (\w+)', l)
        if vid_pid_sn:
            print(f'\r\n  {l} ', end='')
            enum_dev_sn.append(f'{vid_pid_sn.group(1)}_{vid_pid_sn.group(2)}_{vid_pid_sn.group(3)}')

    if set(declared_devs) != set(enum_dev_sn):
        failed_msg = f'Expected {declared_devs}, Enumerated {enum_dev_sn}'
        print('\n'.join(lines))
        assert False, failed_msg
    return 0


# -------------------------------------------------------------
# Tests: host
# -------------------------------------------------------------
def test_host_device_info(board):
    flasher = board['flasher']
    declared_devs = [f'{d["vid_pid"]}_{d["serial"]}' for d in board['tests']['dev_attached']]

    port = get_serial_dev(flasher["uid"], None, None, 0)
    ser = open_serial_dev(port)
    ser.timeout = 0.1

    # reset device since we can miss the first line
    ret = globals()[f'reset_{flasher["name"].lower()}'](board)
    assert ret.returncode == 0, 'Failed to reset device'

    # read until all expected devices are enumerated
    data = b''
    timeout = enum_timeout()
    while timeout > 0:
        new_data = ser.read(ser.in_waiting or 1)
        if new_data:
            data += new_data
        # check if all devices found
        enum_dev_sn = []
        for l in data.decode('utf-8', errors='ignore').splitlines():
            vid_pid_sn = re.search(r'ID ([0-9a-fA-F]+):([0-9a-fA-F]+) SN (\w+)', l)
            if vid_pid_sn:
                enum_dev_sn.append(f'{vid_pid_sn.group(1)}_{vid_pid_sn.group(2)}_{vid_pid_sn.group(3)}')
        if set(declared_devs).issubset(set(enum_dev_sn)):
            break
        time.sleep(0.1)
        timeout -= 0.1
    ser.close()

    if len(data) == 0:
        assert False, 'No data from device'
    lines = data.decode('utf-8', errors='ignore').splitlines()

    enum_dev_sn = []
    for l in lines:
        vid_pid_sn = re.search(r'ID ([0-9a-fA-F]+):([0-9a-fA-F]+) SN (\w+)', l)
        if vid_pid_sn:
            print(f'\r\n  {l} ', end='')
            enum_dev_sn.append(f'{vid_pid_sn.group(1)}_{vid_pid_sn.group(2)}_{vid_pid_sn.group(3)}')

    if set(declared_devs) != set(enum_dev_sn):
        failed_msg = f'Expected {declared_devs}, Enumerated {enum_dev_sn}'
        print('\n'.join(lines))
        assert False, failed_msg
    return 0


def check_msc_info(lines, msc_devs):
    """Print MSC info and verify block_count/block_size against config"""
    inquiry = ''
    disk_size = ''
    for l in lines:
        if re.match(r'^[A-Za-z].*\s+(rev\s+|[0-9])', l) and 'Disk Size' not in l:
            inquiry = l.strip()
        if 'Disk Size' in l:
            disk_size = l.strip()
    if inquiry or disk_size:
        print(f'\r\n  {inquiry} {disk_size} ', end='')
    # Verify block_count and block_size from "Disk Size: COUNT SIZE-byte blocks: N MB"
    if disk_size and msc_devs:
        m = re.match(r'Disk Size:\s+(\d+)\s+(\d+)-byte blocks', disk_size)
        if m:
            actual_count = int(m.group(1))
            actual_size = int(m.group(2))
            for dev in msc_devs:
                exp_count = dev.get('block_count')
                exp_size = dev.get('block_size')
                if exp_count and actual_count == exp_count:
                    assert actual_size == exp_size, (
                        f'MSC block_size mismatch: expected {exp_size}, got {actual_size}')
                    break


def test_host_cdc_msc_hid(board):
    flasher = board['flasher']
    dev_attached = board['tests'].get('dev_attached', [])
    cdc_devs = [d for d in dev_attached if d.get('is_cdc')]
    msc_devs = [d for d in dev_attached if d.get('is_msc')]
    if not cdc_devs and not msc_devs:
        return 'skipped'

    port = get_serial_dev(flasher["uid"], None, None, 0)
    ser = open_serial_dev(port)
    ser.timeout = 0.1

    # reset device to catch mount messages
    ret = globals()[f'reset_{flasher["name"].lower()}'](board)
    assert ret.returncode == 0, 'Failed to reset device'

    # Wait for all expected mount messages
    data = b''
    timeout = enum_timeout()
    wait_cdc = len(cdc_devs) > 0
    wait_msc = len(msc_devs) > 0
    while timeout > 0:
        new_data = ser.read(ser.in_waiting or 1)
        if new_data:
            data += new_data
            cdc_ok = (not wait_cdc) or (b'CDC Interface is mounted' in data)
            msc_ok = (not wait_msc) or (b'Disk Size' in data)
            if cdc_ok and msc_ok:
                break
        time.sleep(0.1)
        timeout -= 0.1

    # Lookup serial chip name from vid_pid
    vid_pid_name = {
        '0403_6001': 'FTDI', '0403_6010': 'FTDI', '0403_6011': 'FTDI', '0403_6014': 'FTDI',
        '10c4_ea60': 'CP210x', '10c4_ea70': 'CP210x',
        '067b_2303': 'PL2303', '067b_23a3': 'PL2303',
        '1a86_7523': 'CH340', '1a86_7522': 'CH340',
        '1a86_55d3': 'CH9102', '1a86_55d4': 'CH9102',
    }

    lines = data.decode('utf-8', errors='ignore').splitlines()

    # Verify and print CDC mount
    if cdc_devs:
        assert b'CDC Interface is mounted' in data, 'CDC device not mounted on host'
        dev = cdc_devs[0]
        chip_name = vid_pid_name.get(dev['vid_pid'], dev['vid_pid'])
        for l in lines:
            if 'CDC Interface is mounted' in l:
                print(f'\r\n  {chip_name}: {l} ', end='')

    # Verify and print MSC mount (inquiry + disk size)
    if msc_devs:
        assert b'MassStorage device is mounted' in data, 'MSC device not mounted on host'
        assert b'Disk Size' in data, 'MSC Disk Size not reported'
        check_msc_info(lines, msc_devs)

    # CDC echo test via flasher serial
    if not cdc_devs:
        ser.close()
        return

    time.sleep(2)
    ser.read(ser.in_waiting)
    ser.reset_input_buffer()

    def rand_ascii(length):
        return "".join(random.choices(string.ascii_letters + string.digits, k=length)).encode("ascii")

    packet_size = 64

    # Echo test: write random 1-packet_size chunks, wait for echo before sending next
    echo_len = 1024
    echo_data = rand_ascii(echo_len)
    ser.reset_input_buffer()
    offset = 0
    while offset < echo_len:
        chunk_size = min(random.randint(1, packet_size), echo_len - offset)
        serial_write_all(ser, echo_data[offset:offset + chunk_size])
        # wait until this chunk is echoed back
        echo = b''
        t_end = time.monotonic() + 1.0
        while time.monotonic() < t_end and len(echo) < chunk_size:
            rd = ser.read(chunk_size - len(echo))
            if rd:
                echo += rd
        expected = echo_data[offset:offset + chunk_size]
        assert echo == expected, (f'CDC echo mismatch at offset {offset} ({chunk_size} bytes):\n'
                                  f'  expected: {expected}\n  received: {echo}')
        offset += chunk_size

    ser.close()


def test_host_msc_file_explorer(board):
    flasher = board['flasher']
    msc_devs = [d for d in board['tests'].get('dev_attached', []) if d.get('is_msc')]
    if not msc_devs:
        return 'skipped'

    port = get_serial_dev(flasher["uid"], None, None, 0)
    ser = open_serial_dev(port)
    ser.timeout = 0.1

    # reset device to catch mount messages
    ret = globals()[f'reset_{flasher["name"].lower()}'](board)
    assert ret.returncode == 0, 'Failed to reset device'

    # Wait for MSC mount (Disk Size message)
    data = b''
    timeout = enum_timeout()
    while timeout > 0:
        new_data = ser.read(ser.in_waiting or 1)
        if new_data:
            data += new_data
            if b'Disk Size' in data:
                break
        time.sleep(0.1)
        timeout -= 0.1
    assert b'Disk Size' in data, 'MSC device not mounted'
    lines = data.decode('utf-8', errors='ignore').splitlines()
    check_msc_info(lines, msc_devs)

    # Send "cat README.TXT" and check response (optional — file may not exist on all drives)
    time.sleep(1)
    ser.reset_input_buffer()
    for ch in 'cat README.TXT\r':
        serial_write_all(ser, ch.encode())
        time.sleep(0.002)

    resp = b''
    t = 10.0
    while t > 0:
        rd = ser.read(max(1, ser.in_waiting))
        if rd:
            resp += rd
        if b'>' in resp and resp.rstrip().endswith(b'>'):
            break
        time.sleep(0.05)
        t -= 0.05

    resp_text = resp.decode('utf-8', errors='ignore')
    if MSC_README_TXT.decode() in resp_text:
        print('README.TXT matched ', end='')

    # MSC throughput test: send dd command to read sectors
    time.sleep(0.5)
    ser.reset_input_buffer()
    for ch in 'dd 1024\r':
        serial_write_all(ser, ch.encode())
        time.sleep(0.002)

    # Read dd output until prompt
    resp = b''
    t = 30.0
    while t > 0:
        rd = ser.read(max(1, ser.in_waiting))
        if rd:
            resp += rd
        if b'KB/s' in resp and b'>' in resp:
            break
        time.sleep(0.05)
        t -= 0.05

    resp_text = resp.decode('utf-8', errors='ignore')
    speed = None
    for line in resp_text.splitlines():
        if 'KB/s' in line:
            print(f'{line.strip()} ', end='')
            m = re.search(r'([\d.]+)\s*([KMG]B/s)', line)  # MSC read speed for the report cell
            if m:
                speed = f'{m.group(1)} {m.group(2)}'
            break

    ser.close()
    assert speed is not None, 'MSC read produced no speed report (dd stalled or failed)'
    return speed


def test_host_msc_file_explorer_freertos(board):
    return test_host_msc_file_explorer(board)


# -------------------------------------------------------------
# Tests: device
# -------------------------------------------------------------
def test_device_board_test(board):
    # Dummy test
    pass


def test_device_cdc_dual_ports(board):
    uid = board['uid']
    port = [
        get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0),
        get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 2)
    ]
    ser = [open_serial_dev(p) for p in port]

    def rand_ascii(length):
        return "".join(random.choices(string.ascii_letters + string.digits, k=length)).encode("ascii")

    sizes = [32, 64, 128, 256, 512, random.randint(2000, 5000)]

    def write_and_check(writer, payload : bytes):
        payload_len = len(payload)
        for s in ser:
            s.reset_input_buffer()
        rd0 = b''
        rd1 = b''
        offset = 0
        # Write in chunks of random 1-64 bytes (device has 64-byte buffer)
        while offset < payload_len:
            chunk_size = min(random.randint(1, 64), payload_len - offset)
            serial_write_all(ser[writer], payload[offset:offset + chunk_size])
            rd0 += ser[0].read(chunk_size)
            rd1 += ser[1].read(chunk_size)
            offset += chunk_size
        assert rd0 == payload.lower(), f'Port0 wrong data ({payload_len}): expected {payload.lower()}... was {rd0}'
        assert rd1 == payload.upper(), f'Port1 wrong data ({payload_len}): expected {payload.upper()}... was {rd1}'

    for size in sizes:
        payload0 = rand_ascii(size)
        write_and_check(0, payload0)

        payload1 = rand_ascii(size)
        write_and_check(1, payload1)
    ser[0].close()
    ser[1].close()


def test_device_cdc_msc(board):
    uid = board['uid']
    # CDC Echo test
    port = get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
    ser = open_serial_dev(port)

    def rand_ascii(length):
        return "".join(random.choices(string.ascii_letters + string.digits, k=length)).encode("ascii")

    sizes = [32, 64, 128, 256, 512, random.randint(2000, 5000)]
    for size in sizes:
        test_str = rand_ascii(size)
        rd_str = b''
        offset = 0
        # Write in chunks of random 1-64 bytes (device has 64-byte buffer)
        while offset < size:
            chunk_size = min(random.randint(1, 64), size - offset)
            serial_write_all(ser, test_str[offset:offset + chunk_size])
            rd_str += ser.read(chunk_size)
            offset += chunk_size
        assert rd_str == test_str, f'CDC wrong data ({size} bytes):\n  expected: {test_str}\n  received: {rd_str}'
    ser.close()

    # MSC Block test
    data = read_disk_file(uid, 0, 'README.TXT')
    assert data == MSC_README_TXT, f'MSC wrong data in README.TXT\n expected: {MSC_README_TXT.decode()}\n received: {data.decode()}'


def test_device_cdc_msc_freertos(board):
    test_device_cdc_msc(board)


def test_device_cdc_msc_throughput(board):
    uid = board['uid']

    def parse_speed(dd_output):
        for line in dd_output.splitlines():
            m = re.search(r'([\d.]+)\s+([kMG]?B)/s', line)
            if m:
                return f'{float(m.group(1)):.1f} {m.group(2)}ps'
        return '?'

    # Wait for MSC disk enumeration
    dev = get_disk_dev(uid, 'TinyUSB', 0)
    timeout = enum_timeout()
    while timeout > 0:
        if os.path.exists(dev):
            break
        time.sleep(0.1); timeout -= 0.1
    assert timeout > 0, f'Disk {dev} not found'

    # Wait for CDC tty enumeration
    tty = get_serial_dev(uid, 'TinyUSB', 'Throughput', 0)
    timeout = enum_timeout()
    while timeout > 0:
        if os.path.exists(tty):
            break
        time.sleep(0.1); timeout -= 0.1
    assert timeout > 0, f'CDC tty {tty} not found'

    # Detect speed (12 Mbps FS / 480 Mbps HS) for payload scaling
    is_fs = False
    for f in glob.glob('/sys/bus/usb/devices/*/serial'):
        try:
            if open(f).read().strip().lower() == uid.lower():
                is_fs = (open(os.path.join(os.path.dirname(f), 'speed')).read().strip() == '12')
                break
        except (OSError, ValueError):
            pass

    # Put tty in raw mode so dd sees pure binary throughput.
    rs = run_cmd(f'timeout 30 stty -F {tty} raw -echo')
    assert rs.returncode == 0, f'stty failed: {cmd_stdout_text(rs.stdout)}'

    # Payload aim: ~5 s per direction at FS (~830 kB/s), much less at HS.
    msc_count = 2 if is_fs else 16    # bs=1M
    cdc_count = 16 if is_fs else 128  # bs=64K

    tmp_file = f'/tmp/cdc_msc_tp_{uid}.bin'

    rw = run_cmd(f'timeout 30 dd if=/dev/zero of={tty} bs=64K count={cdc_count} 2>&1')
    assert rw.returncode == 0, f'CDC dd write failed: {cmd_stdout_text(rw.stdout)}'
    cdc_w = parse_speed(cmd_stdout_text(rw.stdout))

    rr = run_cmd(f'timeout 30 dd if={tty} of=/dev/null bs=64K count={cdc_count} iflag=fullblock 2>&1')
    assert rr.returncode == 0, f'CDC dd read failed: {cmd_stdout_text(rr.stdout)}'
    cdc_r = parse_speed(cmd_stdout_text(rr.stdout))

    rmr = run_cmd(f'dd if={dev} of={tmp_file} bs=1M count={msc_count} iflag=direct 2>&1')
    assert rmr.returncode == 0, f'MSC dd read failed: {cmd_stdout_text(rmr.stdout)}'
    msc_r = parse_speed(cmd_stdout_text(rmr.stdout))

    rmw = run_cmd(f'dd if={tmp_file} of={dev} bs=1M count={msc_count} oflag=direct 2>&1')
    assert rmw.returncode == 0, f'MSC dd write failed: {cmd_stdout_text(rmw.stdout)}'
    msc_w = parse_speed(cmd_stdout_text(rmw.stdout))

    try:
        os.remove(tmp_file)
    except OSError:
        pass

    print(f'  CDC read {cdc_r} write {cdc_w}, MSC read {msc_r} write {msc_w}  ', end='')

    # compact read/write speeds for the report cell, e.g. "✅ C 652/422k M 1.1M/783k"
    # (C=CDC, M=MSC; the unit is shown once when both sides share it)
    def short(s):
        return (s.split()[0].rstrip('0').rstrip('.') + s.split()[-1][0]) if ' ' in s else s

    def pair(r, w):
        r, w = short(r), short(w)
        if r[-1:] == w[-1:] and r[-1:].isalpha():
            r = r[:-1]
        return f'{r}/{w}'

    return f'{REPORT_CELL["pass"]} C {pair(cdc_r, cdc_w)} M {pair(msc_r, msc_w)}'


def test_device_dfu(board):
    uid = board['uid']

    # Wait device enum. Deadline-based: dfu-util -l itself takes ~1 s per call, which a
    # per-iteration countdown would not charge against the budget.
    deadline = time.monotonic() + enum_timeout()
    found = False
    while time.monotonic() < deadline:
        ret = run_cmd(f'dfu-util -l')
        stdout = cmd_stdout_text(ret.stdout)
        if f'serial="{uid}"' in stdout and 'Found DFU: [cafe:400b]' in stdout:
            found = True
            break
        time.sleep(1)

    assert found, 'Device not available'

    f_dfu0 = f'dfu0_{uid}'
    f_dfu1 = f'dfu1_{uid}'

    # Test upload
    try:
        os.remove(f_dfu0)
        os.remove(f_dfu1)
    except OSError:
        pass

    ret = run_cmd(f'dfu-util -S {uid} -a 0 -U {f_dfu0}')
    assert ret.returncode == 0, 'Upload failed'

    ret = run_cmd(f'dfu-util -S {uid} -a 1 -U {f_dfu1}')
    assert ret.returncode == 0, 'Upload failed'

    with open(f_dfu0) as f:
        assert 'Hello world from TinyUSB DFU! - Partition 0' in f.read(), 'Wrong uploaded data'

    with open(f_dfu1) as f:
        assert 'Hello world from TinyUSB DFU! - Partition 1' in f.read(), 'Wrong uploaded data'

    os.remove(f_dfu0)
    os.remove(f_dfu1)


def test_device_dfu_runtime(board):
    uid = board['uid']
    # Wait device enum (deadline-based, see test_device_dfu)
    deadline = time.monotonic() + enum_timeout()
    found = False
    while time.monotonic() < deadline:
        ret = run_cmd(f'dfu-util -l')
        stdout = cmd_stdout_text(ret.stdout)
        if f'serial="{uid}"' in stdout and 'Found Runtime: [cafe:400c]' in stdout:
            found = True
            break
        time.sleep(1)

    assert found, 'Device not available'


def test_device_hid_boot_interface(board):
    uid = board['uid']
    kbd = get_hid_dev(uid, 'TinyUSB', 'TinyUSB_Device', 'event-kbd')
    mouse1 = get_hid_dev(uid, 'TinyUSB', 'TinyUSB_Device', 'if01-event-mouse')
    mouse2 = get_hid_dev(uid, 'TinyUSB', 'TinyUSB_Device', 'if01-mouse')
    # Wait device enum
    timeout = enum_timeout()
    while timeout > 0:
        if os.path.exists(kbd) and os.path.exists(mouse1) and os.path.exists(mouse2):
            break
        time.sleep(1)
        timeout = timeout - 1

    assert timeout > 0, 'HID device not available'


def test_device_hid_composite_freertos(id):
    # TODO implement later
    pass


def test_device_printer_to_cdc(board):
    import threading

    uid = board['uid']

    # Wait for CDC port and printer device
    cdc_port = get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
    ser = open_serial_dev(cdc_port)
    lp_dev = open_printer_dev(uid, 'TinyUSB', 'TinyUSB_Device', 2)

    # Test 0: Verify IEEE 1284 Device ID from sysfs
    expected_id = 'MFG:TinyUSB;MDL:Printer to CDC;CMD:PS;CLS:PRINTER;'
    lp_name = os.path.basename(lp_dev)
    sysfs_id_path = f'/sys/class/usbmisc/{lp_name}/device/ieee1284_id'
    if os.path.exists(sysfs_id_path):
        with open(sysfs_id_path) as f:
            ieee1284_id = f.read().strip()
        if ieee1284_id:
            assert ieee1284_id == expected_id, (f'IEEE 1284 ID mismatch:\n'
                                                f'  expected: {expected_id}\n  got: {ieee1284_id}')

    def rand_ascii(length):
        return "".join(random.choices(string.ascii_letters + string.digits, k=length)).encode("ascii")

    sizes = [32, 64, 128, 256, 512, random.randint(2000, 5000)]

    # flush any stale data
    ser.reset_input_buffer()

    # Test 1: Printer -> CDC with multiple sizes, write in random 1-64 byte chunks
    LP_WRITE_TIMEOUT = 5.0  # seconds; firmware may stall draining the printer OUT endpoint
    for size in sizes:
        test_data = rand_ascii(size)
        ser.reset_input_buffer()
        rd = b''
        offset = 0
        lp_fd = os.open(lp_dev, os.O_WRONLY | os.O_NONBLOCK)
        try:
            while offset < size:
                chunk_size = min(random.randint(1, 64), size - offset)
                buf = test_data[offset:offset + chunk_size]
                written = 0
                while written < len(buf):
                    _, wr, _ = select.select([], [lp_fd], [], LP_WRITE_TIMEOUT)
                    assert wr, f'Printer write timeout after {LP_WRITE_TIMEOUT}s (firmware not draining OUT endpoint)'
                    n = os.write(lp_fd, buf[written:])
                    written += n
                rd += ser.read(chunk_size)
                offset += chunk_size
        finally:
            os.close(lp_fd)
        # read any remaining bytes (fullspeed devices may need extra time)
        while len(rd) < size:
            remaining = ser.read(size - len(rd))
            if not remaining:
                break
            rd += remaining
        assert rd == test_data, (f'Printer->CDC wrong data ({size} bytes):\n'
                                 f'  expected: {test_data[:64]}\n  received: {rd[:64]}')

    # Test 2: CDC -> Printer with multiple sizes, write in random 1-64 byte chunks
    # Use a thread to read from printer since /dev/usb/lp read blocks
    ser.reset_input_buffer()
    time.sleep(0.5)
    for size in sizes:
        test_data = rand_ascii(size)
        rd_result = [b'', None]  # [data, error]
        reader_ready = threading.Event()

        def lp_reader():
            try:
                rd = b''
                fd = os.open(lp_dev, os.O_RDONLY)
                reader_ready.set()
                try:
                    while len(rd) < size:
                        chunk = os.read(fd, min(64, size - len(rd)))
                        if not chunk:
                            break
                        rd += chunk
                finally:
                    os.close(fd)
                rd_result[0] = rd
            except Exception as e:
                rd_result[1] = e
                reader_ready.set()

        reader = threading.Thread(target=lp_reader, daemon=True)
        reader.start()
        # wait for reader to open lp device before writing
        reader_ready.wait(timeout=5)
        time.sleep(0.1)

        # Write to CDC in small chunks with flush to avoid overflowing device FIFO
        offset = 0
        while offset < size:
            chunk_size = min(random.randint(1, 64), size - offset)
            serial_write_all(ser, test_data[offset:offset + chunk_size])
            time.sleep(0.01)
            offset += chunk_size

        reader.join(timeout=10)
        assert not reader.is_alive(), f'CDC->Printer timeout ({size} bytes)'
        assert rd_result[1] is None, f'CDC->Printer read error: {rd_result[1]}'
        assert rd_result[0] == test_data, (f'CDC->Printer wrong data ({size} bytes):\n'
                                           f'  expected: {test_data[:64]}\n  received: {rd_result[0][:64]}')
        time.sleep(0.2)

    ser.close()


def test_device_mtp(board):
    uid = board['uid']

    # --- BEFORE: mute C-level stderr for libmtp vid/pid warnings ---
    fd = sys.stderr.fileno()
    _saved = os.dup(fd)
    _null = os.open(os.devnull, os.O_WRONLY)
    os.dup2(_null, fd)

    mtp = open_mtp_dev(uid)

    # --- AFTER: restore stderr ---
    os.dup2(_saved, fd)
    os.close(_null)
    os.close(_saved)

    if mtp is None or mtp.device is None:
        assert False, 'MTP device not found'

    try:
        assert b"TinyUSB" == mtp.get_manufacturer(), 'MTP wrong manufacturer'
        assert b"MTP Example" == mtp.get_modelname(), 'MTP wrong model'
        assert b'1.0' == mtp.get_deviceversion(), 'MTP wrong version'
        assert b'TinyUSB MTP' == mtp.get_devicename(), 'MTP wrong device name'

        # read and compare readme.txt and logo.png
        f1_expect = b'TinyUSB MTP Filesystem example'
        f2_md5_expect = '40ef23fc2891018d41a05d4a0d5f822f' # md5sum of logo.png
        f1 = uid.encode("utf-8") + b'_file1'
        f2 = uid.encode("utf-8") + b'_file2'
        f3 = uid.encode("utf-8") + b'_file3'
        mtp.get_file_to_file(1, f1)
        with open(f1, 'rb') as file:
            f1_data = file.read()
            os.remove(f1)
            assert f1_data == f1_expect, 'MTP file1 wrong data'
        mtp.get_file_to_file(2, f2)
        with open(f2, 'rb') as file:
            f2_data = file.read()
            os.remove(f2)
            assert f2_md5_expect == hashlib.md5(f2_data).hexdigest(), 'MTP file2 wrong data'
        # test send file
        with open(f3, "wb") as file:
            f3_data = os.urandom(random.randint(1024, 3*1024))
            file.write(f3_data)
            file.close()
            fid = mtp.send_file_from_file(f3, b'file3')
            f3_readback = f3 + b'_readback'
            mtp.get_file_to_file(fid, f3_readback)
            with open(f3_readback, 'rb') as f:
                f3_rb_data = f.read()
                os.remove(f3_readback)
                assert f3_rb_data == f3_data, 'MTP file3 wrong data'
            os.remove(f3)
            mtp.delete_object(fid)
    finally:
        mtp.disconnect()


def test_device_net_lwip_webserver(board):
    # MAC hard-coded in examples/device/net_lwip_webserver/src/main.c; Linux names the
    # USB network interface enx<MAC_lowercase_no_colons>. Device IP is 192.168.7.1 and
    # the example runs an iperf2 TCP server on port 5001 (INCLUDE_IPERF).
    import socket
    mac_no_colons = '0202846a9600'
    iface = 'enx' + mac_no_colons
    device_ip = '192.168.7.1'
    iperf_port = 5001

    # Wait for the host to get an IPv4 address in the device's subnet (DHCP served by the device).
    # USB enum + DHCP serve can take longer on the CI HIL hardware than on local — give it 30s.
    iface_timeout = 30
    deadline = time.monotonic() + iface_timeout
    host_ip = None
    while time.monotonic() < deadline:
        ret = subprocess.run(['ip', '-o', '-4', 'addr', 'show', iface],
                             capture_output=True, text=True, timeout=2)
        m = re.search(r'inet (192\.168\.7\.\d+)/', ret.stdout) if ret.returncode == 0 else None
        if m:
            host_ip = m.group(1)
            break
        time.sleep(0.5)
    assert host_ip, f'USB net iface {iface} did not come up with 192.168.7.x within {iface_timeout}s'

    # Poll the iperf TCP port until the device is accepting. The net stack comes up a bit
    # after DHCP completes; iperf server binding isn't instantaneous after reflash.
    deadline = time.monotonic() + enum_timeout()
    last_err = None
    while time.monotonic() < deadline:
        try:
            with socket.create_connection((device_ip, iperf_port), timeout=1):
                last_err = None
                break
        except OSError as e:
            last_err = e
            time.sleep(0.3)
    assert last_err is None, f'iperf TCP {device_ip}:{iperf_port} not accepting within {enum_timeout()}s: {last_err}'

    # Throughput: 5-second iperf2 TCP test, CSV output for stable parsing.
    # iperf2 CSV final summary line: timestamp,src_ip,src_port,dst_ip,dst_port,id,interval,bytes,bps
    ret = subprocess.run(['iperf', '-c', device_ip, '-t', '5', '-y', 'C'],
                         capture_output=True, text=True, timeout=30)
    stderr = ret.stderr.strip()
    stdout = ret.stdout.strip()
    assert ret.returncode == 0, f'iperf rc={ret.returncode}: stderr={stderr!r} stdout={stdout!r}'
    lines = [l for l in stdout.splitlines() if l]
    assert lines, f'iperf produced no output (rc={ret.returncode}, stderr={stderr!r})'
    try:
        bps = int(lines[-1].split(',')[-1])
    except (ValueError, IndexError) as e:
        raise AssertionError(f'could not parse iperf output: {lines[-1]!r} ({e})')
    mbps = bps / 1e6
    print(f'  iperf {mbps:5.1f} Mbps', end='')

    # Reject implausibly low throughput - a working USB-net link should clear this easily.
    assert mbps >= 1.0, f'iperf throughput too low: {mbps:.2f} Mbps'


def test_device_msc_dual_lun(board):
    uid = board['uid']

    # Read README from LUN 0
    data0 = read_disk_file(uid, 0, 'README0.TXT')
    readme0 = b"LUN0: " + MSC_README_TXT
    assert data0 == readme0, f'MSC LUN0 wrong data in README0.TXT\n  expected: {readme0}\n  received: {data0}'

    # Read README from LUN 1
    data1 = read_disk_file(uid, 1, 'README1.TXT')
    readme1 = b"LUN1: " + MSC_README_TXT
    assert data1 == readme1, f'MSC LUN1 wrong data in README1.TXT\n  expected: {readme1}\n  received: {data1}'


def test_device_midi_test(board):
    uid = board['uid']

    # Find MIDI device via /dev/snd/by-id using board UID
    timeout = enum_timeout()
    midi_port = None
    while timeout > 0:
        pattern = f'/dev/snd/by-id/usb-*_{uid}-*'
        devs = glob.glob(pattern)
        if devs:
            # by-id entry points to controlCX, derive card number for midiCXD0
            link = os.path.basename(os.readlink(devs[0]))  # e.g. "controlC2"
            card_num = link.replace('controlC', '')
            midi_path = f'/dev/snd/midiC{card_num}D0'
            if os.path.exists(midi_path):
                midi_port = midi_path
                break
        time.sleep(1)
        timeout -= 1
    assert midi_port is not None, f'MIDI device not found for {uid}'

    # Read MIDI messages and verify note on/off
    import select
    with open(midi_port, 'rb') as f:
        notes = []
        # Read for up to 3 seconds to capture a few notes (286ms interval)
        end_time = time.monotonic() + 3
        while time.monotonic() < end_time:
            ready, _, _ = select.select([f], [], [], 0.5)
            if ready:
                data = f.read(64)
                if data:
                    # Parse MIDI bytes: note_on = 0x90, note_off = 0x80
                    i = 0
                    while i + 2 < len(data):
                        status = data[i]
                        if (status & 0xF0) == 0x90:  # Note On
                            notes.append(data[i + 1])
                            i += 3
                        elif (status & 0xF0) == 0x80:  # Note Off
                            i += 3
                        else:
                            i += 1

    assert len(notes) >= 2, f'Expected at least 2 MIDI notes, got {len(notes)}'
    # Verify notes are from the expected sequence
    note_sequence = [
        74, 78, 81, 86, 90, 93, 98, 102, 57, 61, 66, 69, 73, 78, 81, 85,
        88, 92, 97, 100, 97, 92, 88, 85, 81, 78, 74, 69, 66, 62, 57, 62,
        66, 69, 74, 78, 81, 86, 90, 93, 97, 102, 97, 93, 90, 85, 81, 78,
        73, 68, 64, 61, 56, 61, 64, 68, 74, 78, 81, 86, 90, 93, 98, 102
    ]
    for n in notes:
        assert n in note_sequence, f'Unexpected MIDI note {n}'


def test_device_audio_test_freertos(board):
    uid = board['uid']

    if os.name == 'nt':
        return 'skipped'

    pcm = None
    timeout = enum_timeout()
    while timeout > 0:
        pcm = get_alsa_capture_dev(uid)
        if pcm:
            break
        time.sleep(1)
        timeout -= 1

    assert pcm is not None, f'ALSA capture device not found for {uid}'

    raw_path = f'/tmp/tinyusb_audio_{uid}.raw'
    cmd = [
        'arecord',
        '-D', pcm,
        '-q',
        '-f', 'S16_LE',
        '-c', '1',
        '-r', '48000',
        '-d', '2',
        '-t', 'raw',
        raw_path,
    ]

    ret = subprocess.run(cmd, capture_output=True, text=True, timeout=20)
    assert ret.returncode == 0, f'arecord failed: {ret.stderr.strip() or ret.stdout.strip()}'

    try:
        with open(raw_path, 'rb') as f:
            raw = f.read()
    finally:
        try:
            os.remove(raw_path)
        except OSError:
            pass

    assert len(raw) >= 48000, f'Captured too little audio: {len(raw)} bytes'
    assert (len(raw) % 2) == 0, f'Invalid 16-bit audio length: {len(raw)}'

    sample_count = len(raw) // 2
    samples = [int.from_bytes(raw[i:i + 2], 'little', signed=False) for i in range(0, len(raw), 2)]
    assert sample_count > 1024, f'Not enough samples captured: {sample_count}'

    # The firmware sends a continuous uint16 ramp. Using ALSA hw: capture bypasses
    # PulseAudio processing, so most adjacent samples should differ by exactly 1.
    total_diffs = sample_count - 1
    one_step = 0
    near_step = 0
    for i in range(total_diffs):
        d = (samples[i + 1] - samples[i]) & 0xFFFF
        if d == 1:
            one_step += 1
        if d in (0, 1, 2, 47, 48, 49):
            near_step += 1

    one_ratio = one_step / total_diffs
    near_ratio = near_step / total_diffs
    assert one_ratio >= 0.85, f'Unexpected audio pattern (strict ratio={one_ratio:.3f})'
    assert near_ratio >= 0.98, f'Unexpected audio pattern (relaxed ratio={near_ratio:.3f})'

    print(f'  ALSA {pcm} strict={one_ratio:.3f} relaxed={near_ratio:.3f}', end='')


def test_device_hid_generic_inout(board):
    uid = board['uid']
    import hid  # cython-hidapi (pip: hidapi, apt: python3-hid)

    # Find HID device by UID (VID=0xCafe)
    timeout = enum_timeout()
    dev = None
    while timeout > 0:
        for d in hid.enumerate(0xCafe):
            if d['serial_number'] == uid:
                dev = d
                break
        if dev:
            break
        time.sleep(1)
        timeout -= 1
    assert dev is not None, f'HID device not found for {uid}'

    h = hid.device()
    h.open(dev['vendor_id'], dev['product_id'], uid)
    try:
        # Echo test: send random data and verify echo
        for size in [8, 32, 63]:
            # Report ID (0) + payload, padded to 64 bytes
            payload = bytes([random.randint(1, 255) for _ in range(size)])
            report = bytes([0]) + payload + bytes(64 - size)
            h.write(report)
            echo = h.read(64, 2000)
            assert echo and len(echo) >= size, (
                f'HID echo timeout or short read ({size} bytes)')
            assert bytes(echo[:size]) == payload, (
                f'HID echo wrong data ({size} bytes):\n'
                f'  expected: {payload.hex()}\n  received: {bytes(echo[:size]).hex()}')
    finally:
        h.close()


def test_device_usbtest(board):
    # Run the Linux testusb tier-4 battery (test/hil/usbtest.py) against the enumerated cafe:4010
    # device; surface the pass count in the report cell ("✅ 30/30", or "❌ 29/30" on a partial).
    uid = board['uid']

    def usbtest_enumerated():
        # match VID:PID too, not just the serial: right after flashing, the previous example's
        # enumeration (same serial, different PID) can linger and would fail usbtest.py's lookup
        for f in glob.glob('/sys/bus/usb/devices/*/serial'):
            d = os.path.dirname(f)
            try:
                if (open(f).read().strip().lower() == uid.lower()
                        and open(os.path.join(d, 'idVendor')).read().strip() == 'cafe'
                        and open(os.path.join(d, 'idProduct')).read().strip() == '4010'):
                    return True
            except OSError:
                pass
        return False

    end = time.monotonic() + enum_timeout()
    while time.monotonic() < end and not usbtest_enumerated():
        time.sleep(0.2)
    # fail before usbtest_permit: an absent device would otherwise queue on the battery
    # mutex for minutes behind real batteries just to have usbtest.py report "no device"
    if not usbtest_enumerated():
        # 0/30 rather than a bare cell: the battery never ran (30 = standard case count)
        raise TestFail(f'no cafe:4010 device with serial {uid}',
                       metric=f'{REPORT_CELL["fail"]} 0/30')
    # settle: right after flashing the enumeration can bounce once (and on dual-port parts like
    # CH32V307 the other port's stale usbtest node — same serial and PID — lingers a moment);
    # running testusb into that gap sees the device drop mid-case
    time.sleep(3)

    # --keep-binding is required for concurrent batteries: usbtest.py's cleanup unbinds
    # EVERY usbtest-bound interface (releasing stale same-PID grabs), which would kill a
    # peer battery mid-run under USBTEST_PARALLEL > 1; the unbind path has also wedged a
    # host xHCI (usb_hcd_alloc_bandwidth) on this rig. Leaving bindings is harmless with
    # unique example PIDs - the next example re-enumerates under a different PID and binds
    # its normal driver. usbtest_permit budgets USBTEST_PARALLEL batteries per controller.
    script = Path(__file__).resolve().parent / 'usbtest.py'
    cmd = f'python3 "{script}" --serial "{uid}" --json --keep-binding --timeout 60'
    with usbtest_permit(uid):
        r = run_cmd(cmd, timeout=200)
    out = cmd_stdout_text(r.stdout)
    brace = out.find('{')
    try:
        data = json.loads(out[brace:])
        passed, failed = int(data['passed']), int(data['failed'])
    except (ValueError, KeyError, json.JSONDecodeError):
        raise TestFail(f'usbtest did not run: {compact_output(out) or cmd_stdout_text(r.stderr)}',
                       metric=f'{REPORT_CELL["fail"]} 0/30')

    total = passed + failed
    if failed == 0 and total > 0:
        return f'{REPORT_CELL["pass"]} {passed}/{total}'
    bad = [c.get('num') for c in data.get('cases', []) if c.get('status') != 'PASS']
    raise TestFail(f'usbtest {passed}/{total} (cases failed: {bad})',
                   metric=f'{REPORT_CELL["fail"]} {passed}/{total}')


# -------------------------------------------------------------
# Main
# -------------------------------------------------------------

# The per-board run order is shuffled (see test_board).
# Every example carries a unique hardcoded idProduct (see its usb_descriptors.c)

# device tests
device_tests = [
    'device/cdc_dual_ports',
    'device/cdc_msc',
    'device/dfu',
    'device/cdc_msc_throughput',
    'device/audio_test_freertos',
    'device/dfu_runtime',
    'device/cdc_msc_freertos',
    'device/hid_boot_interface',
    'device/msc_dual_lun',
    'device/hid_generic_inout',
    'device/printer_to_cdc',
    'device/midi_test',
    'device/mtp',
    'device/usbtest',  # cafe:4010, unique PID; runs the Linux testusb tier-4 battery via usbtest.py
    # 'device/net_lwip_webserver',  # disabled for PR #3605: USB net iface enum is flaky on the CI HIL host
]

dual_tests = [
    'dual/host_info_to_device_cdc',
]

host_test = [
    'host/cdc_msc_hid',
    'host/msc_file_explorer',
    'host/msc_file_explorer_freertos',
    'host/device_info',
]


def find_firmware(variant: str, example: str):
    """Locate a built example's firmware base path (no extension) under
    cmake-build-<variant>/<example>/. Accepts the single-config layout (firmware
    directly in the example dir) or Ninja Multi-Config (a per-config subdir like
    RelWithDebInfo/). Returns the base Path, or None if not built."""
    fw_dir = TINYUSB_ROOT / build_dir / f'cmake-build-{variant}' / example
    base = Path(example).name
    if fw_dir.is_dir():
        for cand in [fw_dir / base, fw_dir / 'RelWithDebInfo' / base,
                     *(p.with_suffix('') for p in sorted(fw_dir.glob(f'*/{base}.elf')))]:
            if cand.with_suffix('.elf').exists() or cand.with_suffix('.bin').exists():
                return cand
    return None


def test_example(board: Board, variant: str, example: str) -> tuple[int, str, str | None]:
    """
    Test example firmware
    :param board: board dict
    :param variant: build variant name = build dir (cmake-build-<variant>) and report row
    :param example: example name
    :return: (err_count, status, metric) where err_count is 0 on success/skip or
             1 on failure, status is one of 'pass'/'fail'/'skip' (a missing binary
             counts as 'skip'), and metric is an optional string a test returns to
             show in its report cell instead of the pass symbol (e.g. speed)
    """
    err_count = 0
    result_status = 'fail'
    metric = None

    test_name = f'{variant:40} {example:30} ...'

    fw_name = find_firmware(variant, example)
    if fw_name is None:
        log_line(f'{test_name} Skip (no binary)')
        return 0, 'skip', None

    if verbose:
        log_line(f'Flashing {fw_name}.elf')

    # flash firmware (unless --skip-flash), then run the test. Both may fail randomly,
    # retry a few times.
    global _enum_timeout
    start_s = time.time()
    flash_ok = True
    last_err = ''
    last_detail = ''
    for i in range(max_retry):
        _enum_timeout = ENUM_TIMEOUT if i == 0 else ENUM_TIMEOUT_RETRY
        attempt_out = io.StringIO()
        with redirect_stdout(attempt_out):
            if not skip_flash:
                with flash_permit(board['uid']):
                    t_flash = time.monotonic()
                    ret = globals()[f'flash_{board["flasher"]["name"].lower()}'](board, str(fw_name))
                    if PROFILE:
                        log_line(f'[prof] {variant} {example} flash attempt {i + 1}: '
                                 f'{time.monotonic() - t_flash:.1f}s rc={ret.returncode}')
                flash_ok = (ret.returncode == 0)
            if flash_ok:
                try:
                    tret = globals()[f'test_{example.replace("/", "_")}'](board)
                    last_detail = compact_output(attempt_out.getvalue())
                    if tret == 'skipped':
                        status = STATUS_SKIPPED
                        result_status = 'skip'
                    else:
                        status = STATUS_OK
                        result_status = 'pass'
                        # a test may return a string to show in its report cell (e.g. speed)
                        metric = tret if isinstance(tret, str) else None
                    msg = f'{test_name}  {status}'
                    if last_detail:
                        msg += f'  {last_detail}'
                    msg += f'  in {time.time() - start_s:.1f}s'
                    log_line(msg)
                    break
                except Exception as e:
                    last_err = str(e)
                    last_detail = compact_output(attempt_out.getvalue())
                    if i == max_retry - 1:
                        err_count += 1
                        # a failing test may still carry a metric to show in its cell (e.g. "❌ 29/30")
                        metric = getattr(e, 'metric', None)
                        msg = f'{test_name}  {STATUS_FAILED}: {e}'
                        if last_detail:
                            msg += f'  {last_detail}'
                        msg += f'  in {time.time() - start_s:.1f}s'
                        log_line(msg)
                    else:
                        msg = f'{test_name}  retry {i+2}/{max_retry}: test failed: {e}'
                        if last_detail:
                            msg += f'  {last_detail}'
                        log_line(msg)
                        time.sleep(0.5)
            else:
                last_err = 'Flash failed'
                last_detail = compact_output(attempt_out.getvalue())
                if i < max_retry - 1:
                    msg = f'{test_name}  retry {i+2}/{max_retry}: flash failed'
                    if last_detail:
                        msg += f'  {last_detail}'
                    log_line(msg)
                time.sleep(0.5)

    if not flash_ok:
        err_count += 1
        msg = f'{test_name}  Flash {STATUS_FAILED}'
        if last_err:
            msg += f': {last_err}'
        if last_detail:
            msg += f'  {last_detail}'
        msg += f'  in {time.time() - start_s:.1f}s'
        log_line(msg)

    return err_count, result_status, metric


def build_board(board: Board) -> tuple[str, int]:
    """Build firmware for this board via tools/build.py.
    Honors board config's variant list and build.args defines.
    Output goes to cmake-build/cmake-build-<variant>/ (tools/build.py layout)."""
    name = board['name']
    bcfg = cast(BuildCfg, board.get('build', {}))
    extra_defs = bcfg.get('args', [])
    variants = board.get('variant') or [{'name': name, 'flags': ''}]

    failed = 0
    for v in variants:
        cmd = [sys.executable, str(TINYUSB_ROOT / 'tools' / 'build.py'), '-b', name]
        for d in extra_defs:
            cmd += ['-D', d]
        if v['name'] != name:
            cmd += ['--build-name', v['name']]
        for d in v.get('defines', []):
            cmd += ['-D', d]
        for tok in v.get('flags', '').split():
            cmd += [f'--cflag={tok}']
        if verbose:
            cmd.append('-v')
            print(f'  + {" ".join(cmd)}')
        r = subprocess.run(cmd, cwd=TINYUSB_ROOT)
        if r.returncode != 0:
            failed += 1
    return name, failed


def test_board(board: Board) -> tuple[str, int, list[str], list, float]:
    name = board['name']
    flasher = board['flasher']

    try:
        _lock_fh = acquire_board_lock(name)
    except RuntimeError as e:
        log_line(f'{name:25} {STATUS_FAILED}: {e}')
        # visible report row so the ❌ matches the exit code; failed-tests stays
        # empty so a re-run repeats the whole board (no bogus -bt test filter)
        return name, 1, [], [(name, {'board-locked': 'fail'}, None)], 0.0
    # after the lock: flock wait behind a concurrent run is not board cost
    t_board = time.monotonic()
    try:
        # default to all tests
        test_list = []

        if name in board_test:
            test_list = board_test[name]
        elif len(test_only) > 0:
            # Explicit -t: filter against the board's capabilities so a device-only
            # board doesn't try to run host/dual tests (the test functions need a
            # `dev_attached` entry in the board config that won't exist).
            board_tests = board.get('tests', {})
            if 'only' in board_tests:
                allowed = set(board_tests['only'])
                test_list = [t for t in test_only if t in allowed]
            else:
                for t in test_only:
                    category = t.split('/', 1)[0]
                    if board_tests.get(category) is True:
                        test_list.append(t)
        else:
            if 'tests' in board:
                board_tests = board['tests']
                if board_tests.get('device') is True:
                    test_list += list(device_tests)
                if board_tests.get('dual') is True:
                    test_list += dual_tests
                if board_tests.get('host') is True:
                    test_list += host_test
                if 'only' in board_tests:
                    test_list = board_tests['only']
                if 'skip' in board_tests:
                    for skip in board_tests['skip']:
                        if skip in test_list:
                            test_list.remove(skip)
                            log_line(f'{name:25} {skip:30} ... Skip')

        err_count = 0
        failed_tests = []
        rows = []  # list of (row_label, {example: status}, duration) — one row per build variant
        # a -t/-bt filtered run times only a subset; report no duration so an accumulate
        # re-run keeps the previous full-run value
        partial = bool(test_only) or name in board_test
        variants = board.get('variant') or [{'name': name, 'flags': ''}]

        prev_last = None  # last test of the previous variant: the variant boundary is an adjacency too
        for v in variants:
            vname = v['name']
            # Shuffle each (board, variant)'s run order — de-synchronizes the worker pool so
            # usbtest batteries and flash churn spread across the timeline instead of convoying,
            # and surfaces order-dependent bugs. Seeded for replay (HIL_SHUFFLE_SEED, logged by
            # main). Unique per-example PIDs make any two different examples re-enumerate; only
            # the variant boundary can repeat the same example (same PID) — swap it away.
            run_list = list(test_list)
            if shuffle_seed is not None and len(run_list) > 1:
                random.Random(f'{shuffle_seed}:{name}:{vname}').shuffle(run_list)
                if run_list[0] == prev_last:
                    run_list[0], run_list[-1] = run_list[-1], run_list[0]
            if run_list:
                prev_last = run_list[-1]
            t_variant = time.monotonic()
            cells = {}
            for test in run_list:
                ec, status, metric = test_example(board, vname, test)
                err_count += ec
                cells[test] = metric if metric else status
                if ec > 0:
                    failed_tests.append(test)
            dur = f'{time.monotonic() - t_variant:.0f}s' if run_list and not partial else None
            rows.append((vname, cells, dur))

        # board duration excludes the teardown park-flash below; a partial (filtered)
        # run reports 0.0 so it never overwrites a cached full-run duration
        t_total = 0.0 if partial else time.monotonic() - t_board

        # flash board_test last to disable board's usb (skipped when --skip-flash is set);
        # this is teardown/park, not a test — not recorded in the report
        if not skip_flash:
            test_example(board, variants[0]['name'], 'device/board_test')

        return name, err_count, sorted(set(failed_tests)), rows, t_total
    finally:
        if _lock_fh:
            try:
                # clear our pid record before dropping the flock: this worker
                # process lives on (pool reuse), so a stale record would make
                # board_lock.py's pid-liveness checks report a freed board as
                # still locked for the rest of the run
                _lock_fh.truncate(0)
            except OSError:
                pass
            _lock_fh.close()


REPORT_MD = 'hil_report.md'
REPORT_JSON = 'hil_report.json'
# controller hints learned from previous runs: uid -> {'name', 'pci', 'duration'}. Only
# 'pci' is consumed (dispatch order and first-flash budgeting, never battery
# serialization); name/duration are informational. PCI addresses are boot-stable (bus
# numbers are not), so the cache survives reboots and only goes stale on re-cabling.
CONTROLLER_CACHE = Path.home() / '.cache' / 'tinyusb-hil' / 'controller_cache.json'


def schedule_boards(boards: list, pci_of_uid: dict) -> list:
    """Dispatch order: round-robin across host controllers so every controller's
    serialized usbtest battery chain is fed from t=0 instead of one card's boards
    convoying at the head of the queue. Boards without a controller hint form their
    own bucket; config order is kept within a bucket."""
    buckets = {}
    for b in boards:
        buckets.setdefault(pci_of_uid.get(b['uid'], '?'), []).append(b)
    return [b for grp in itertools.zip_longest(*buckets.values()) for b in grp if b is not None]


def render_matrix(rows_all: list) -> str:
    """Render rows (list of (row_label, {example: status}, duration)) as an aligned
    markdown matrix: columns = tests (bare names) centered, boards left-aligned,
    per-row duration as the trailing column."""
    seen = set()
    for _, cells, _ in rows_all:
        seen.update(cells)
    if not seen:
        return 'No tests were run.'

    # metric-bearing columns pinned first (usbtest score, throughput, explorer read speed),
    # the rest alphabetical by bare test name: stable regardless of the (shuffled) execution order
    pinned = ['usbtest', 'cdc_msc_throughput', 'msc_file_explorer', 'msc_file_explorer_freertos']

    def col_key(t):
        name = t.rsplit('/', 1)[-1]
        return (pinned.index(name) if name in pinned else len(pinned), name, t)

    columns = sorted(seen, key=col_key)
    headers = [c.rsplit('/', 1)[-1] for c in columns] + ['duration']  # bare example names

    def cell(cells, col):
        v = cells.get(col)
        if v is None:
            return ''
        return REPORT_CELL.get(v, v)  # status symbol, or a metric string (e.g. speed) verbatim

    rows_vals = [(lbl, [cell(cells, c) for c in columns] + [dur or ''])
                 for lbl, cells, dur in rows_all]
    board_hdr = 'Board'
    board_w = max([len(board_hdr)] + [len(lbl) for lbl, _ in rows_vals])
    col_w = [max([len(h)] + [len(vals[i]) for _, vals in rows_vals])
             for i, h in enumerate(headers)]

    def line(label, values):
        padded = [label.ljust(board_w)] + [v.center(w) for v, w in zip(values, col_w)]
        return '| ' + ' | '.join(padded) + ' |'

    header = line(board_hdr, headers)
    sep = '| ' + '-' * board_w + ' | ' + ' | '.join(':' + '-' * (w - 2) + ':' for w in col_w) + ' |'
    body = [line(lbl, vals) for lbl, vals in rows_vals]

    # tally run cells (blank/not-run cells are absent from the dicts). A cell is a bare status
    # ('pass'/'fail'/'skip') or a metric string that carries its own icon (e.g. "❌ 29/30" is a
    # fail, "✅ 30/30" / "✅ CDC …" a pass), so classify by the leading icon.
    def cell_kind(v):
        if v == 'fail' or (isinstance(v, str) and v.startswith(REPORT_CELL['fail'])):
            return 'fail'
        if v == 'skip' or (isinstance(v, str) and v.startswith(REPORT_CELL['skip'])):
            return 'skip'
        return 'pass'
    kinds = [cell_kind(v) for _, cells, _ in rows_all for v in cells.values()]
    failed = kinds.count('fail')
    skipped = kinds.count('skip')
    passed = kinds.count('pass')
    summary = (f'**{REPORT_CELL["pass"]} {passed} passed · {REPORT_CELL["fail"]} {failed} failed · '
               f'{REPORT_CELL["skip"]} {skipped} skipped · blank not run**')

    return summary + '\n\n' + '\n'.join([header, sep] + body)


def accumulate_report(mret: list, report_dir: Path, fresh: bool) -> str:
    """Merge this run's results into hil_report.json in report_dir, then (re)write
    the markdown matrix to hil_report.md. `fresh` (a full run, no --accumulate/-bt)
    starts a new report; otherwise a re-run accumulates so boards/tests that
    already passed are preserved while re-run cells are updated. Returns the md."""
    acc = {}  # ordered {row_label: [cells dict, duration str|None]}
    jpath = report_dir / REPORT_JSON
    if not fresh and jpath.is_file():
        try:
            saved = json.loads(jpath.read_text())
            # CI keys the report dir by run id, so the sidecar can only have been
            # written by an earlier attempt of the same run
            for entry in saved.get('rows', []):
                acc[entry['board']] = [dict(entry['cells']), entry.get('duration')]
        except (ValueError, KeyError, TypeError):
            pass  # corrupt/old sidecar: start fresh

    # merge this run: current cells override prior for boards/tests that ran; a filtered
    # run reports duration None, keeping the previous full-run value
    for name, _, _, rows, _ in mret:
        if rows and not any('board-locked' in cells for _, cells, _ in rows):
            # board ran for real this time: clear a stale lock-failure cell
            # (its row is keyed by board name; test rows may be variant names)
            stale = acc.get(name)
            if stale is not None:
                stale[0].pop('board-locked', None)
                if not stale[0]:
                    # variant-keyed boards never repopulate the board-name row —
                    # drop it or it renders as a blank ghost row
                    del acc[name]
        for row_label, cells, dur in rows:
            row = acc.setdefault(row_label, [{}, None])
            row[0].update(cells)
            if dur is not None:
                row[1] = dur

    report_dir.mkdir(parents=True, exist_ok=True)
    jpath.write_text(json.dumps({'rows': [{'board': k, 'cells': c, 'duration': d}
                                          for k, (c, d) in acc.items()]}, indent=2) + '\n')

    md = render_matrix([(k, c, d) for k, (c, d) in acc.items()])
    (report_dir / REPORT_MD).write_text(md + '\n', encoding='utf-8')
    return md


def main() -> None:
    """
    Hardware test on specified boards
    """
    global verbose
    global test_only
    global board_test
    global build_dir
    global max_retry
    global skip_flash

    duration = time.time()

    parser = argparse.ArgumentParser()
    parser.add_argument('config_file', help='Configuration JSON file')
    parser.add_argument('-b', '--board', action='append', default=[], help='Boards to test, all if not specified')
    parser.add_argument('--flasher', action='append', default=[],
                        help='Only boards using these flashers, e.g. esptool '
                             '(for splitting one config across CI jobs)')
    parser.add_argument('--exclude-flasher', action='append', default=[],
                        help='Exclude boards using these flashers')
    parser.add_argument('-a', '--accumulate', action='store_true',
                        help='Merge results into the existing report instead of starting fresh '
                             '(re-runs; the .failed file starts with this)')
    parser.add_argument('-sf', '--skip-flash', action='store_true', help='Run tests without flashing firmware (use whatever is already on the board)')
    parser.add_argument('-t', '--test-only', action='append', default=[], help='Tests to run, all if not specified')
    parser.add_argument('-bt', '--board-test', action='append', default=[],
                        help='Per-board test list as BOARD:test1,test2 (overrides -t for that board); repeat for multiple boards')
    parser.add_argument('-B', '--build-dir', default='cmake-build', help='Build folder name (default: cmake-build)')
    parser.add_argument('--build', action='store_true', help='Build firmware for selected boards with cmake before running tests')
    parser.add_argument('-r', '--retry', type=int, default=3, help='Retry count for failed tests (default: 3)')
    parser.add_argument('-v', '--verbose', action='store_true', help='Verbose output')
    args = parser.parse_args()

    config_file = Path(args.config_file)
    boards = args.board
    verbose = args.verbose
    test_only = args.test_only
    for entry in args.board_test:
        bname, _, tnames = entry.partition(':')
        if not bname or not tnames:
            parser.error(f'invalid --board-test value: {entry!r} (expected BOARD:test1,test2)')
        board_test[bname] = [t for t in tnames.split(',') if t]
    build_dir = args.build_dir
    max_retry = args.retry
    skip_flash = args.skip_flash

    # if config file is not found, try to find it in the same directory as this script
    if not config_file.exists():
        config_file = Path(__file__).resolve().parent / config_file
    with config_file.open() as f:
        config = cast(HilConfig, json.load(f))

    if len(boards) == 0:
        config_boards = list(config['boards'])
    else:
        unknown = [b for b in boards if b not in {e['name'] for e in config['boards']}]
        if unknown:
            # exiting 0 with 'No tests were run.' would read as a green HIL run
            print(f'ERROR: board(s) not in {config_file.name}: {", ".join(unknown)}')
            sys.exit(1)
        config_boards = [e for e in config['boards'] if e['name'] in boards]
    config_boards = [e for e in config_boards if e['flasher']['name'] not in args.exclude_flasher
                     and (not args.flasher or e['flasher']['name'] in args.flasher)]

    build_err = 0
    if args.build:
        if build_dir != 'cmake-build':
            print(f'warning: --build writes into cmake-build/, but -B is {build_dir!r}; '
                  f'tests will not find the freshly built firmware')
        print('-' * 30)
        print(f'Build phase: {len(config_boards)} board(s)')
        print('-' * 30)
        for board in config_boards:
            _, nfail = build_board(board)
            build_err += nfail
        print('-' * 30)
        print(f'Build phase done: {build_err} failed')
        print('-' * 30)

    # HIL report sidecar (hil_report.json/.md) and the .failed re-run spec live in
    # report_dir (CI keys it by run id, so it persists across run attempts but is
    # private to one run). A full run starts fresh; a re-run (--accumulate / -bt,
    # i.e. the .failed file) merges so already-passed boards/tests are preserved.
    # Clear prior state up front on a fresh run so a crash mid-run can't leave a
    # stale report or re-run spec to be consumed by a retry.
    report_dir = Path(os.environ.get('HIL_REPORT_DIR', '.'))
    failed_fname = report_dir / (config_file.name + '.failed')
    fresh = not (args.accumulate or args.board_test)
    if fresh:
        report_dir.mkdir(parents=True, exist_ok=True)
        for f in (REPORT_JSON, REPORT_MD):
            (report_dir / f).unlink(missing_ok=True)
        failed_fname.unlink(missing_ok=True)

    seed = os.getenv('HIL_SHUFFLE_SEED') or str(int(time.time()))
    log_line(f'test-order shuffle seed: {seed} (HIL_SHUFFLE_SEED={seed} to replay); '
             f'flash/usbtest parallel per controller: {FLASH_PARALLEL}/{USBTEST_PARALLEL}; '
             f'enum timeout first/retry: {ENUM_TIMEOUT}/{ENUM_TIMEOUT_RETRY}s')

    hints = {}
    try:
        with CONTROLLER_CACHE.open() as f:
            loaded = json.load(f)
        # tolerate a hand-edited/torn cache: keep only the expected uid -> dict shape
        if isinstance(loaded, dict):
            hints = {k: v for k, v in loaded.items() if isinstance(v, dict)}
    except (OSError, ValueError):
        pass
    hints_by_uid = {uid: h['pci'] for uid, h in hints.items() if h.get('pci')}
    config_boards = schedule_boards(config_boards, hints_by_uid)
    log_line('dispatch order: ' + ', '.join(b['name'] for b in config_boards))

    mgr = Manager()
    cmap = mgr.dict()
    initargs = (Lock(), seed,
                [Semaphore(USBTEST_PARALLEL) for _ in range(CONTROLLER_SLOTS)],
                [Semaphore(FLASH_PARALLEL) for _ in range(CONTROLLER_SLOTS)],
                cmap, Lock(), hints_by_uid)
    with Pool(processes=os.cpu_count() or 1, initializer=init_worker, initargs=initargs) as pool:
        async_ret = pool.map_async(test_board, config_boards)
        try:
            mret = async_ret.get(timeout=POOL_TIMEOUT)
        except MpTimeoutError:
            pool.terminate()
            pool.join()
            raise RuntimeError(f'HIL worker pool timed out after {POOL_TIMEOUT}s')

        err_count = build_err + sum(e[1] for e in mret)
        # generate the re-run spec if anything failed: run ONLY the failed boards (-b),
        # each restricted to its own failed tests (-bt); a board with failures but no
        # test list (e.g. board-locked) re-runs entirely. --accumulate preserves the
        # already-passed cells in the report.
        parts = ['--accumulate']
        for name, err, fts, _, _ in mret:
            if err > 0:
                parts.append(f'-b {name}')
                if fts:
                    parts.append(f'-bt {name}:{",".join(fts)}')
        if len(parts) > 1:  # build-only failures have no boards to re-run
            report_dir.mkdir(parents=True, exist_ok=True)
            with failed_fname.open('w') as f:
                f.write(' '.join(parts))
        else:
            failed_fname.unlink(missing_ok=True)

    # refresh controller hints: pci resolved this run, plus board durations when the
    # full test list ran (a -t/-bt filtered run would understate the board's real cost)
    try:
        if PROFILE:
            # debug snapshot of the run's live uid->PCI / PCI->slot resolutions
            report_dir.mkdir(parents=True, exist_ok=True)
            with (report_dir / 'hil_profile_ctrl.json').open('w') as f:
                json.dump(dict(cmap), f, indent=1, sort_keys=True)
        uid_of = {b['name']: b['uid'] for b in config['boards']}
        for name, _, _, _, dur in mret:
            uid = uid_of.get(name)
            if uid is None:
                continue
            h = dict(hints.get(uid) or {})
            h['name'] = name  # informational: cache is keyed by uid
            h['pci'] = cmap.get(f'uid:{uid}') or h.get('pci')
            if dur > 0:  # test_board reports 0.0 for filtered (partial) runs
                h['duration'] = round(dur, 1)
            hints[uid] = h
        # merge-on-write: another HIL job (e.g. the esp split) may have finished since
        # our startup read - re-read and overlay only this run's boards so its entries
        # survive, then replace atomically so a concurrent reader never sees a torn file
        merged = {}
        try:
            with CONTROLLER_CACHE.open() as f:
                cur = json.load(f)
            if isinstance(cur, dict):
                merged = {k: v for k, v in cur.items() if isinstance(v, dict)}
        except (OSError, ValueError):
            pass
        merged.update({uid_of[n]: hints[uid_of[n]] for n, *_ in mret if n in uid_of})
        CONTROLLER_CACHE.parent.mkdir(parents=True, exist_ok=True)
        tmp = CONTROLLER_CACHE.with_suffix('.json.tmp')
        with tmp.open('w') as f:
            json.dump(merged, f, indent=1, sort_keys=True)
        tmp.replace(CONTROLLER_CACHE)
    except OSError as e:
        print(f'warning: cannot persist controller hints to {CONTROLLER_CACHE}: {e}')

    # board x test result matrix -> hil_report.md (accumulates across re-runs) + stdout
    report = accumulate_report(mret, report_dir, fresh)
    print()
    print(report)
    print(f'\nReport written to {(report_dir / REPORT_MD).resolve()}')

    duration = time.time() - duration
    print()
    print("-" * 30)
    print(f'Total failed: {err_count} in {duration:.1f}s')
    print("-" * 30)
    sys.exit(err_count)


if __name__ == '__main__':
    main()