#!/usr/bin/env python3 """Quick HIL pool health check. For every board in the rig's HIL config: is the flash probe on the USB bus, does a light example flash, and does the board's USB device (uid) come back up? Missing firmware is BUILT on the spot (tools/build.py, idf.py for espressif; one get_deps retry) — never skipped; --no-build opts out. Applies only per-device-safe recovery (probe authorized-toggle, board reset/re-flash) and prints a markdown summary table. Row statuses: ok (flashed and verified; under --scan-only: probe present — the scan checks presence only), flash-failed (firmware delivery failed: probe missing, build failed, flasher error, silent flash no-op, park not verified), failed (the check ran but did not verify: flashed with no enumeration/serial, or the check itself errored), locked (board flock held by another process — reported, never waited on or bypassed). Config is picked by hostname unless given: ci -> tinyusb.json, tusb (hifiphile rig) -> hfp.json, anything else is a dev PC -> local.json. Lives in test/hil/ beside hil_lock.py and hil_flash.py, which it imports; board recovery uses the repo's .claude/skills/usb-kernel-recover/scripts/usb_recover.sh. """ import argparse import io import json import glob import os import re import shlex import shutil import socket import subprocess import sys import threading import time from concurrent.futures import ThreadPoolExecutor from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(Path(__file__).resolve().parent)) # for import-as-module callers import hil_lock import hil_flash USB_RECOVER = REPO_ROOT / '.claude' / 'skills' / 'usb-kernel-recover' / 'scripts' / 'usb_recover.sh' SEEN_CACHE = Path.home() / '.cache' / 'tinyusb-hil' / 'pool_seen.json' CONFIG_BY_HOST = {'ci': 'tinyusb.json', 'tusb': 'hfp.json'} # anything else: dev PC -> local.json # light-example preference; first built wins DEVICE_CANDIDATES = ['device/dfu_runtime', 'device/cdc_msc', 'device/cdc_msc_freertos', 'device/hid_composite_freertos', 'device/cdc_dual_ports'] HOST_CANDIDATES = ['host/device_info', 'host/cdc_msc_hid', 'host/msc_file_explorer_freertos'] ENUM_WAIT = 12 # s, uid wait after flash ENUM_WAIT_RETRY = 8 # s, uid wait after a recovery reset/re-flash SERIAL_WAIT = 6 # s, host-board serial-output wait print_mutex = threading.Lock() t0 = time.monotonic() def say(msg: str) -> None: with print_mutex: print(f'[{time.monotonic() - t0:6.1f}s] {msg}', file=sys.__stdout__, flush=True) def scan_usb() -> dict: """busport -> {'serial', 'vidpid', 'ino'} for every enumerated USB device. Only -[....] dirs match (root hubs, named 'usbN' with no dash, are excluded: their fabricated PCI-address 'serial' and slow autosuspend-wake read cost 6-7s/scan on this rig). Keyed by busport, not serial: a serial can be shared by two different devices (e.g. an Espressif USB-Serial-JTAG bridge and the cafe TinyUSB device it flashes derive both from the same MAC) — collapsing them into one dict slot would silently drop whichever lost the race.""" found = {} for f in glob.glob('/sys/bus/usb/devices/*-*/serial'): d = os.path.dirname(f) busport = os.path.basename(d) try: sn = open(f).read().strip().lower() vidpid = f'{open(d + "/idVendor").read().strip()}:{open(d + "/idProduct").read().strip()}' found[busport] = {'serial': sn, 'vidpid': vidpid, 'ino': os.stat(d + '/').st_ino} except OSError: continue return found def find_usb(uid: str, devs: dict | None = None): """Locate a flasher probe by uid, excluding VID cafe (TinyUSB DUT firmware): a probe's uid can coincidentally equal its DUT's (Espressif USB-Serial-JTAG bridges derive both from the same MAC), and the DUT is never the probe. J-Link zero-pads numeric serials (681295394 -> 000681295394): an all-digit uid matches an all-digit serial only when that serial equals the uid zero-padded to the serial's own length (leading zeros only) — never when the zero-stripped uid is empty, so a placeholder serial (metro_m4_express's probe legitimately reports '123456') can't be mistaken for an unrelated device.""" devs = devs if devs is not None else scan_usb() u = uid.lower() candidates = [(bp, dev) for bp, dev in devs.items() if not dev['vidpid'].startswith('cafe:')] for bp, dev in candidates: if dev['serial'] == u: return bp, dev['vidpid'], dev['ino'] stripped = u.lstrip('0') if u.isdigit() and stripped: for bp, dev in candidates: s = dev['serial'] if s.isdigit() and s == stripped.zfill(len(s)): return bp, dev['vidpid'], dev['ino'] return None def find_device(uid: str, pid: str | None): """Board-online check: TinyUSB device (idVendor cafe) with this uid, optionally PID-pinned. VID cafe keeps an Espressif USB-Serial-JTAG (303a) sharing the MAC serial from false-passing.""" for busport, dev in scan_usb().items(): if (dev['serial'] == uid.lower() and dev['vidpid'].startswith('cafe:') and (pid is None or dev['vidpid'].endswith(pid))): return busport, dev['vidpid'], dev['ino'] return None def wait_device(uid: str, pid: str | None, old_ino, budget: float): """Wait for the board's device with a NEW sysfs inode (flash resets the MCU, so a genuine flash must re-enumerate; the inode is the re-enumeration marker).""" deadline = time.monotonic() + budget while time.monotonic() < deadline: hit = find_device(uid, pid) if hit and hit[2] != old_ino: return hit time.sleep(0.5) return None def lock_board(name: str): """Nonblocking flock per hil_lock.py protocol. Returns handle, or a str with the holder's info when the board is locked elsewhere. Board locks are ALWAYS respected: a held board is reported as locked and skipped — never waited on, and there is deliberately no bypass here.""" os.makedirs(hil_lock.BOARD_LOCK_DIR, exist_ok=True) try: fh = hil_lock.flock_nb(name) except OSError: # NB: conflates a held flock with open() failures (EACCES/EROFS/ENOSPC) — # benign while everything on the rig runs as one uid; a cross-uid setup # would need flock_nb to distinguish the two info = hil_lock.read_record(name) return json.dumps(info) if info else 'unknown holder' if not hil_lock.write_record(fh, 'pool_check'): # an invisible lock (flock held, no record) is worse than no lock: status # can't show us and release can't recognize the protected holder — bail out hil_lock.clear_record(fh) fh.close() return 'ERROR: holder record write failed (lock dir unwritable?)' return fh def unlock_board(fh) -> None: hil_lock.clear_record(fh) fh.close() def can_recover() -> bool: if not USB_RECOVER.is_file(): return False try: r = subprocess.run(['sudo', '-n', 'true'], capture_output=True) except OSError: # sudo not installed (bare dev PC/container): recovery off, not fatal return False return r.returncode == 0 def recover_probe(uid: str, busport: str) -> bool: """Soft-replug an enumerated-but-wedged probe: deauthorize+reauthorize (no VBUS cut, touches only this device). Success = the probe re-enumerated (new sysfs inode), not the helper's exit code (observed to flake while the toggle worked). J-Links respond with a full disconnect and can stay off the bus for >8 s.""" pre = find_usb(uid) try: # bounded: the sysfs authorized store can block in D state on a wedged # device, and this runs while the board's (release-protected) flock is held subprocess.run(['sudo', '-n', str(USB_RECOVER), 'authorized', busport], capture_output=True, text=True, timeout=30) except subprocess.TimeoutExpired: return False deadline = time.monotonic() + 20 while time.monotonic() < deadline: post = find_usb(uid) if post and (pre is None or post[2] != pre[2]): return True time.sleep(0.5) return False def resolve_variant(board: dict, example: str, note: list | None = None) -> str: """Build-dir variant name for `example`: the first of the board's variants with already-built firmware, falling back to the board name. Notes the pick when it differs from the board name (e.g. nanoch32v203's build dir is variant 'nanoch32v203-fsdev', not the board name).""" name = board['name'] for v in board.get('variant') or [{'name': name}]: vn = v['name'] if hil_flash.find_firmware(vn, example, flasher=board['flasher']['name']): if vn != name and note is not None and f'variant: {vn}' not in note: note.append(f'variant: {vn}') return vn return name def pick_example(board: dict, note: list, build_missing: bool = True): """(example, kind, variant, fw) with built firmware for this board; kind is 'device' (uid check) or 'host' (serial-output check); variant is the resolved build-dir variant that has it (see resolve_variant); fw is the firmware path to flash, extension included. When nothing is built and build_missing is set (the default — never skip a board for lack of a build), the preferred candidate is built on the spot via ensure_fw.""" tests = board.get('tests', {}) only = tests.get('only', []) skip = set(tests.get('skip', [])) # config's known-broken examples: never pick one is_device = tests.get('device') or any(t.startswith('device/') for t in only) if is_device: cand = DEVICE_CANDIDATES + [t for t in only if t.startswith('device/') and t != 'device/usbtest'] kind = 'device' else: cand = HOST_CANDIDATES + [t for t in only if t.startswith('host/')] kind = 'host' for ex in dict.fromkeys(cand): if ex in skip: continue variant = resolve_variant(board, ex, note) fw = hil_flash.find_firmware(variant, ex, flasher=board['flasher']['name']) if fw: return ex, kind, variant, fw if not build_missing: return None, kind, None, None # nothing built anywhere: build the preferred candidate (an only-list board # must get one of its own examples — dfu_runtime etc. may not even configure) pref = [c for c in dict.fromkeys(cand) if c not in skip and (not only or c in only)] if not pref: return None, kind, None, None variant = (board.get('variant') or [{'name': board['name']}])[0]['name'] for ex in pref[:2]: # the second candidate covers a preferred example that fails to build fw = ensure_fw(board, variant, ex, note) if fw: return ex, kind, variant, fw return None, kind, None, None _pid_cache: dict[str, str | None] = {} def get_expected_pid(example: str) -> str | None: """USB_PID for `example`'s device descriptor (examples//src/ usb_descriptors.c, '#define USB_PID 0x....'), lowercased and without the 0x prefix to match sysfs idProduct. Cached per example; None (also cached) when the file or define isn't there — host examples have no usb_descriptors.c, and the caller must stay quiet rather than false-warn.""" if example not in _pid_cache: pid = None try: text = (REPO_ROOT / 'examples' / example / 'src' / 'usb_descriptors.c').read_text() # optional parens as in tools/check_example_pids.py's parser m = re.search(r'#define\s+USB_PID\s+\(?\s*(0x[0-9a-fA-F]+)', text) if m: pid = m.group(1)[2:].lower() except OSError: pass _pid_cache[example] = pid return _pid_cache[example] def call_flasher(fn, *fn_args) -> tuple[int, str]: """Run a hil_flash flash_*/reset_* backend, normalizing raises to a failure: several backends raise instead of returning nonzero (get_serial_dev RuntimeError when a bridge's /dev/serial/by-id node vanishes, config.env FileNotFoundError, .jlink script OSError) and an exception must not skip the caller's retry/recovery ladder. Returns (returncode, error line).""" try: ret = fn(*fn_args) if ret.returncode == 0: return 0, '' err = flash_error_line(hil_flash.cmd_stdout_text(ret.stdout)) return ret.returncode, err or f'rc={ret.returncode}' except Exception as e: return -1, repr(e)[:90] def flash(board: dict, fw, allow_recovery: bool, probe_port: str, note: list) -> bool: """Flash the resolved firmware with one retry; on repeated failure soft-replug the probe and always make one final flash attempt afterward, regardless of whether the replug is confirmed — some probes (WCH-Link, ST-Link, CP210x, picoprobe) leave their sysfs kobject intact across an authorized toggle instead of dropping off the bus. Returns True on success. `fw` comes from pick_example: a re-resolve here would use the global search policy and miss a firmware ensure_fw just built into cmake-build/ under an exclusive -B.""" fn = getattr(hil_flash, f'flash_{board["flasher"]["name"].lower()}') for attempt in range(3): if attempt == 2: if not (allow_recovery and probe_port): return False cur = find_usb(board['flasher']['uid']) if cur is None: # probe gone from the bus: its old busport may now hold an UNRELATED # device (bus renumbering) and the helper only checks occupancy, so # toggling would deauthorize an innocent fixture — skip the toggle note.append('probe vanished before toggle') else: say(f'{board["name"]:26} recovery: replugging probe {cur[0]} (authorized toggle)') if recover_probe(board['flasher']['uid'], cur[0]): note.append('probe replugged') time.sleep(2) # udev recreates /dev/serial/by-id symlinks after re-enumeration else: note.append('probe toggle unconfirmed') rc, err = call_flasher(fn, board, str(fw)) if rc == 0: return True if rc == 127: # flasher binary missing: retries/probe recovery can't fix env note.append(f'flasher tool missing ({err}) — esptool needs the ESP-IDF env (get-idf)' if board['flasher']['name'].lower() == 'esptool' else f'flasher tool missing: {err}') return False if attempt == 0: say(f'{board["name"]:26} flash retry: {err}') else: note.append(f'flash: {err}') return False def flash_error_line(out: str) -> str: """Most informative line of a failed flash's output: last error-looking line, else the last non-empty one.""" lines = [l.strip() for l in out.splitlines() if l.strip()] for l in reversed(lines): if any(k in l.lower() for k in ('error', 'fail', 'unknown', 'cannot', 'timeout', 'no valid', 'not found', 'unable')): return l[:90] return lines[-1][:90] if lines else '' def check_host_serial(board: dict, do_reset: bool = True, want_hello: bool = False) -> bytes | None: """Host-only boards never enumerate their uid (their USB port is the host side); aliveness = output on the flasher's UART bridge after a reset. A probe byte is written each poll so an echo-only firmware (board_test) also answers. Returns the first output chunk (b'' when silent, None when the port is absent/drops) so the caller can also judge WHAT answered — see boardtest_output(). do_reset=False listens to the firmware as-is: used right after a flash whose own reset already started it — a second openocd/JLink session back-to-back on the same probe can fail transiently and leave the target halted.""" import serial try: port = hil_flash.get_serial_dev(board['flasher']['uid'], None, None, 0) ser = serial.Serial(port, baudrate=115200, timeout=0.3, write_timeout=1) except Exception as e: say(f'{board["name"]:26} no flasher serial port: {e}') return None try: # flush BEFORE issuing the reset: pyserial's open-time flush is long past, # so this drops the pre-reset CDC backlog (which must not count as life) # while keeping the board's post-reset boot banner, which prints while the # reset tool is still tearing down and would be eaten by a post-reset flush ser.reset_input_buffer() if do_reset: getattr(hil_flash, f'reset_{board["flasher"]["name"].lower()}')(board) # collect the WHOLE window and judge content, not the first chunk: the # probe's CDC bridge has its own FIFO, so stale pre-flash output (e.g. # board_test hellos) can arrive after our host-side flush and must not # decide the verdict alone. Early-exit once non-board_test output proves # a real example is talking. data = b'' deadline = time.monotonic() + SERIAL_WAIT while time.monotonic() < deadline: try: ser.write(b'U') data += ser.read(256) except serial.SerialTimeoutException: pass except serial.SerialException: return None # port dropped mid-poll (bridge re-enumerating) # early-exit on the caller's positive signal: fresh board_test hello # (park verification) vs any non-board_test output (example liveness); # stale bridge-FIFO backlog of the OTHER kind must not end the window if want_hello: if b'Hello from TinyUSB' in data: return data elif data and not boardtest_output(data): return data return data finally: ser.close() def boardtest_output(data: bytes) -> bool: """True when (non-empty) serial output is recognizably ONLY board_test's: its periodic HELLO_STR and echoes of our b'U' pokes, nothing else. Any residue beyond that (an example banner, log lines) proves other firmware is talking, however much stale board_test backlog surrounds it. Used as a negative identity marker — after flashing a host example, board_test-only chatter means the flash silently didn't take (the host analog of the PID check).""" residue = data.replace(b'Hello from TinyUSB', b'') for junk in (b'U', b'\r', b'\n'): residue = residue.replace(junk, b'') return len(residue) == 0 def build_example(board: dict, variant: str, example: str) -> int: """Build one example for this board: tools/build.py (same invocation shape as hil_test.build_board: -T target, -D per build.args, variant defines/flags, --build-name), or idf.py directly for espressif (tools/build.py's esp branch ignores -T and builds everything; variant flags travel as -DCFLAGS_CLI, the same channel tools/build.py uses). Bounded and process-group-killed via run_cmd; 600 s: a first configure+build of an SDK-heavy family (pico, nrf, esp) exceeds the old 300. Builds normally run pre-lock (pick_example / the pre-park ensure), so a board flock is not held here except on rare recovery paths. Per-build compile parallelism is capped at cpu/-j so -j concurrent builds cannot swamp sibling workers' verification windows. Returns the build's returncode (127 = ESP-IDF env missing).""" name = board['name'] variants = board.get('variant') or [{'name': name}] vcfg = next((v for v in variants if v['name'] == variant), variants[0]) if board['flasher']['name'].lower() == 'esptool': if not shutil.which('idf.py'): return 127 # ESP-IDF env not sourced in this shell # -B keyed off the VARIANT so ensure_fw's post-build lookup finds it cmd = ['idf.py', '-C', f'examples/{example}', '-B', f'cmake-build/cmake-build-{vcfg["name"]}/{example}', '-G', 'Ninja', f'-DBOARD={name}', 'build'] for d in board.get('build', {}).get('args', []) + vcfg.get('defines', []): cmd.insert(-1, f'-D{d}') if vcfg.get('flags'): cmd.insert(-1, f'-DCFLAGS_CLI={vcfg["flags"]}') # the IDF component manager writes examples//dependencies.lock in the # SOURCE tree (idf.py -B relocates only the build dir), so concurrent esp # builds of one example for different targets corrupt each other's solve with _esp_lock, _build_sem: return hil_flash.run_cmd(shlex.join(cmd), cwd=str(hil_flash.TINYUSB_ROOT), timeout=600).returncode cmd = [sys.executable, str(hil_flash.TINYUSB_ROOT / 'tools' / 'build.py'), '-b', name, '-T', Path(example).name, '-j', str(max(1, (os.cpu_count() or _jobs) // _jobs))] for d in board.get('build', {}).get('args', []): cmd += ['-D', d] if vcfg['name'] != name: cmd += ['--build-name', vcfg['name']] for d in vcfg.get('defines', []): cmd += ['-D', d] for tok in vcfg.get('flags', '').split(): cmd += [f'--cflag={tok}'] with _build_sem: return hil_flash.run_cmd(shlex.join(cmd), cwd=str(hil_flash.TINYUSB_ROOT), timeout=600).returncode _deps_lock = threading.Lock() # one get_deps at a time (it also drains _build_sem) _esp_lock = threading.Lock() # idf.py mutates source-tree dependencies.lock per example _no_build = False # --no-build: ensure_fw never invokes a build _jobs = 4 # mirrors -j; set in main before the pool starts _build_sem = threading.BoundedSemaphore(4) # build slots; get_deps drains ALL (exclusive) _builds: dict = {} # (variant, example) -> (fw|None, reason): one attempt per run def ensure_fw(board: dict, variant: str, example: str, note: list): """Firmware for `example`, building it when absent — never skip a board for lack of a build (--no-build opts out). One retry with deps fetched and the CMake caches dropped when the first build fails (fresh checkouts lack the family deps; a cache configured in a broken env poisons every later attempt). Returns the firmware path, or None with the failure noted. Call BEFORE taking the board lock: builds are long. One build attempt per (variant, example) per run, success or failure — memoized in _builds, so a repeat call (park, under the held flock) resolves instantly even when an exclusive -B hides the fresh cmake-build/ artifact from the global search.""" fw = hil_flash.find_firmware(variant, example, flasher=board['flasher']['name']) if fw: return fw key, base = (variant, example), Path(example).name if key in _builds: return _builds[key][0] if _no_build: _builds[key] = (None, 'disabled') note.append(f'build skipped (--no-build): {base}') return None rc = build_example(board, variant, example) if rc == 127 and board['flasher']['name'].lower() == 'esptool': _builds[key] = (None, 'no-env') note.append(f'cannot build {base}: ESP-IDF env missing (get-idf)') return None if rc == 124: # hung build: a deps/cache retry cannot cure it, don't double the stall _builds[key] = (None, 'timeout') note.append(f'build timeout: {base}') return None if rc != 0: # retry once with deps fetched and the CMake caches dropped (cache only — # a tree wipe would destroy every other example's firmware). get_deps # git-resets already-present shared deps (lib/fatfs's ffconf.h dance), so # it must exclude every in-flight build, not just other get_deps calls: # it drains ALL build slots before running. with _deps_lock: for _ in range(_jobs): _build_sem.acquire() try: r = hil_flash.run_cmd(shlex.join([sys.executable, str(hil_flash.TINYUSB_ROOT / 'tools' / 'get_deps.py'), '-b', board['name']]), cwd=str(hil_flash.TINYUSB_ROOT), timeout=600) finally: for _ in range(_jobs): _build_sem.release() if r.returncode != 0: note.append('get_deps failed') bd = hil_flash.TINYUSB_ROOT / 'cmake-build' / f'cmake-build-{variant}' # esp configures one level deeper (//): wipe both layouts for d in (bd, bd / example): shutil.rmtree(d / 'CMakeFiles', ignore_errors=True) (d / 'CMakeCache.txt').unlink(missing_ok=True) rc = build_example(board, variant, example) if rc != 0: _builds[key] = (None, 'fail') note.append(f'build failed: {base}') return None # tools/build.py and the idf.py invocation above always write to cmake-build/: # look there too even when an explicit -B narrowed the global search — this is # OUR fresh build, not a stale-candidate fallback fw = hil_flash.find_firmware(variant, example, roots=[hil_flash.build_dir, 'cmake-build'], flasher=board['flasher']['name']) _builds[key] = (fw, 'ok' if fw else 'no-fw') note.append(f'built {base}' if fw else f'build produced no firmware: {base}') return fw def ensure_board_test(board: dict, variant: str, note: list): """board_test firmware for parking, building it if absent (via ensure_fw). Espressif included — tools/build.py builds board_test for that family too; the build just needs the ESP-IDF env (127 → noted, park is then skipped).""" fw = hil_flash.find_firmware(variant, 'device/board_test', flasher=board['flasher']['name']) if fw: return fw variants = board.get('variant') or [{'name': board['name']}] if not any(v['name'] == variant for v in variants): variant = variants[0]['name'] return ensure_fw(board, variant, 'device/board_test', note) def verdict(row: dict, ok: bool) -> str: """Row status for a verification result, preserving a 'flash-failed' a deeper layer already recorded (silent flash no-op, board_test delivery failure).""" return 'ok' if ok else ('flash-failed' if row['status'] == 'flash-failed' else 'failed') def host_alive(board: dict, note: list, row: dict, flashed_example: bool = False) -> bool: """Serial aliveness with recovery: silent -> (build and) flash board_test (it hellos every second and echoes) -> recheck. Also cures a silent flash no-op that left the board crashed. With flashed_example=True (a host example was just flashed), board_test-shaped output FAILS the check: the parked image still talking means the example flash silently didn't take — the host analog of the device path's PID check. Side effect: delivery-class failures (silent no-op, board_test build/flash failure) set row['status'] = 'flash-failed' so verdict() preserves the cause; the caller derives the final status from the return value via verdict().""" data = check_host_serial(board) if data: if flashed_example and boardtest_output(data): note.append('board_test output after example flash: silent flash no-op') row['status'] = 'flash-failed' return False return True variant = resolve_variant(board, 'device/board_test', note) fw = ensure_board_test(board, variant, note) if fw is None: note.append('serial silent; board_test unavailable') row['status'] = 'flash-failed' return False say(f'{board["name"]:26} recovery: serial silent, flashing board_test') rc, err = call_flasher(getattr(hil_flash, f'flash_{board["flasher"]["name"].lower()}'), board, str(fw)) if rc != 0: note.append(f'serial silent; board_test flash failed: {err}') row['status'] = 'flash-failed' return False if not check_host_serial(board): return False if flashed_example: # board_test talking proves the BOARD is alive, but the just-flashed # example never produced serial — that verification still fails note.append('example silent; board alive via board_test reflash') return False note.append('recovered via board_test reflash') return True def device_recover_and_check(board: dict, example: str, variant: str, old_ino, note: list, row: dict, seen: dict) -> bool: """Wait for the flashed board's uid to re-enumerate; on timeout, try one board reset (skipped for flashers with no hardware reset — see hil_flash.RESET_NOOP, it would just burn the wait) and wait again. The PID policy is deliberately asymmetric. Pre-reset, the re-enumeration was caused by the flash itself, so a PID mismatch most likely means the build dir is stale (the flash DID write what find_firmware found) — warn, don't fail — UNLESS the firmware was built this very run: then 'stale build' is impossible and the mismatch can only be a silent flash no-op, which fails. Post-reset, the re-enumeration proves nothing about the flash (the reset alone explains it), so a mismatch is treated as a silent flash no-op and fails; an unknown expected PID scores ok with a 'pid unverified' note in both paths.""" name = board['name'] expected_pid = get_expected_pid(example) built_this_run = _builds.get((variant, example), (None, ''))[1] == 'ok' def seen_hit(hit): seen[board['uid']] = {'name': name, 'busport': hit[0], 'when': time.strftime('%Y-%m-%d %H:%M')} hit = wait_device(board['uid'], None, old_ino, ENUM_WAIT) if hit: if expected_pid is not None and not hit[1].endswith(expected_pid): if built_this_run: row['device'] = f'❌ {hit[1]}' note.append(f'pid {hit[1]}, this run built {expected_pid}: silent flash no-op') row['status'] = 'flash-failed' return False note.append(f'⚠ pid {hit[1]}, source says {expected_pid}: stale build or silent flash no-op') elif expected_pid is None: note.append('pid unverified') row['device'] = f'✅ {hit[1]}' seen_hit(hit) return True flasher_name = board['flasher']['name'].lower() if flasher_name in hil_flash.RESET_NOOP: note.append(f'no hardware reset available for {flasher_name}') row['device'] = '❌ not enumerated' return False say(f'{name:26} recovery: uid not up, resetting board') rc, err = call_flasher(getattr(hil_flash, f'reset_{flasher_name}'), board) if rc != 0: note.append(f'reset failed: {err}') hit = wait_device(board['uid'], None, old_ino, ENUM_WAIT_RETRY) if not hit: row['device'] = '❌ not enumerated' note.append('reset did not help') return False if expected_pid is None: row['device'] = f'✅ {hit[1]}' note.append('reset recovered (pid unverified)') seen_hit(hit) return True if hit[1].endswith(expected_pid): row['device'] = f'✅ {hit[1]}' note.append('reset recovered') seen_hit(hit) return True row['device'] = f'❌ {hit[1]}' note.append(f'reset recovered wrong pid, expected {expected_pid}: silent flash no-op') row['status'] = 'flash-failed' return False def check_board(board: dict, args, allow_recovery: bool, seen: dict) -> dict: name = board['name'] row = {'name': name, 'probe': '❌ missing', 'flash': '–', 'device': '–', 'note': [], 'status': 'failed'} note = row['note'] probe = find_usb(board['flasher']['uid']) if probe: row['probe'] = f'✅ {probe[0]}' seen[board['flasher']['uid']] = {'name': f'{name} probe', 'busport': probe[0], 'when': time.strftime('%Y-%m-%d %H:%M')} else: last = seen.get(board['flasher']['uid']) note.append(f'probe last seen {last["busport"]} {last["when"]}' if last else 'probe never seen by pool_check') say(f'{name:26} probe MISSING ({board["flasher"]["name"]} {board["flasher"]["uid"]})') # existing firmware only here; a missing build is built on the spot further # down (after a lock peek), except in scan/no-build modes — and never for a # missing probe (nothing could be flashed anyway) example, kind, variant, fw = pick_example(board, note, build_missing=False) if kind == 'host': note.append('host-only board') if args.scan_only: hit = find_device(board['uid'], None) # report the BOARD's usb state, not just the probe's: the enumerated device # (with busport), off-bus (normal when parked in board_test), or n/a for # host-only boards whose uid never enumerates if hit: row['device'] = f'✅ {hit[1]} @{hit[0]}' elif kind == 'host': row['device'] = '– n/a (host-only)' else: row['device'] = '⚫ off bus (parked?)' # scan verifies probe presence only: that check DID run, so probe present # is ok; a missing probe means no firmware could be delivered → flash-failed row['status'] = 'ok' if probe else 'flash-failed' if probe: say(f'{name:26} probe ✅ {probe[0]}' + (f' device {hit[1]}' if hit else '')) return row if not probe: row['status'] = 'flash-failed' return row bt_variant = resolve_variant(board, 'device/board_test', note) need_example = example is None and not args.no_build # board_test is also host_alive's recovery image, so host boards pre-build it # even under --no-park; --no-build gates EVERY build, board_test included need_bt = (not args.no_build and (not args.no_park or kind == 'host') and hil_flash.find_firmware(bt_variant, 'device/board_test', flasher=board['flasher']['name']) is None) if need_example or need_bt: # builds are long and run BEFORE locking (park must never hold the flock # through a build); peek the lock first so minutes of building are not # wasted on — or a rebuilt tree swapped under — a board CI holds right now peek = lock_board(name) if isinstance(peek, str): if peek.startswith('ERROR:'): # environment failure, not a held lock row['flash'] = '❌ lock' row['status'] = 'failed' else: row['flash'] = '🔒 locked' row['status'] = 'locked' note.append(peek) say(f'{name:26} locked: {peek}') return row unlock_board(peek) if need_example: example, kind, variant, fw = pick_example(board, note, build_missing=True) if need_bt and (example is not None or kind == 'host'): # skip the park-image build when the example build already failed on a # device board: the row returns before any flash/park could use it ensure_board_test(board, bt_variant, note) if example is None: if not any(n.startswith(('build failed', 'build timeout', 'build produced', 'build skipped', 'cannot build')) for n in note): note.append('no firmware built') if kind != 'host': row['status'] = 'flash-failed' say(f'{name:26} probe ✅ {probe[0]} (no firmware to flash)') return row # host-only board: aliveness is still checkable without flashing — reset and # listen to whatever firmware is on it (the parked board_test echoes and # prints a periodic hello on the flasher UART) lk = lock_board(name) if isinstance(lk, str): if lk.startswith('ERROR:'): # environment failure, not a held lock row['flash'] = '❌ lock' row['status'] = 'failed' else: row['flash'] = '🔒 locked' row['status'] = 'locked' note.append(lk) say(f'{name:26} locked: {lk}') return row try: if example is None: # host-only without firmware: UART-only aliveness check ok = host_alive(board, note, row) row['device'] = '✅ serial out' if ok else '❌ no serial out' row['status'] = verdict(row, ok) say(f'{name:26} – {row["device"]} (existing firmware)') return row pre = find_device(board['uid'], None) old_ino = pre[2] if pre else None try: if not flash(board, fw, allow_recovery, probe[0], note): row['flash'] = f'❌ {Path(example).name}' row['status'] = 'flash-failed' say(f'{name:26} flash FAILED ({example})') return row row['flash'] = f'✅ {Path(example).name}' if kind == 'host': ok = host_alive(board, note, row, flashed_example=True) row['device'] = '✅ serial out' if ok else '❌ no serial out' else: ok = device_recover_and_check(board, example, variant, old_ino, note, row, seen) row['status'] = verdict(row, ok) say(f'{name:26} {row["flash"]} {row["device"]}') return row finally: # teardown for EVERY path that attempted a flash (a failed programmer op # can still have erased/half-written the target): re-park while the # board lock is still held if not args.no_park: park_board(board, kind, row, note) finally: unlock_board(lk) def park_board(board: dict, kind: str, row: dict, note: list) -> None: """Re-park with board_test, building it if absent (ensure_board_test), and VERIFY it took: board_test never enumerates USB, so a device board's cafe device must drop off the bus, and a host board must answer with board_test's own output — a rc=0 park that changed nothing (silent no-op) must not pass. A board left unparked marks an ok row flash-failed (never downgrading a 'failed' verify verdict — that is the more diagnostic signal), with one exception: an espressif board without the ESP-IDF env cannot build board_test — noted, not a board fault.""" # capture BEFORE the park flash: uid-disappearance only verifies the park if # the device was on the bus to begin with (a fast park drops it immediately) on_bus_before = kind != 'host' and find_device(board['uid'], None) is not None variant = resolve_variant(board, 'device/board_test', note) fw = ensure_board_test(board, variant, note) if fw is None: if any(n.startswith('cannot build board_test') for n in note): note.append('park skipped (no ESP-IDF env)') else: # --no-build disables builds, not parking (--no-park is that opt-out): # a board left running a USB-active image is unparked either way note.append('unparked: board_test not built (--no-build)' if any(n.startswith('build skipped (--no-build): board_test') for n in note) else 'unparked: board_test unavailable (build failed/timed out)') if row['status'] == 'ok': row['status'] = 'flash-failed' return rc, err = call_flasher(getattr(hil_flash, f'flash_{board["flasher"]["name"].lower()}'), board, str(fw)) if rc != 0: note.append(f'park flash failed: {err}') if row['status'] == 'ok': row['status'] = 'flash-failed' return if kind == 'host': # no second reset (the park flash's own reset already started board_test); # POSITIVE marker: its hello must appear — stale example output may still # drain from the probe bridge's FIFO alongside it and is not disqualifying data = check_host_serial(board, do_reset=False, want_hello=True) if not (data and b'Hello from TinyUSB' in data): note.append('park unverified: no board_test output') if row['status'] == 'ok': row['status'] = 'flash-failed' return if not on_bus_before: # board never enumerated this run: uid-disappearance can't distinguish a # verified park from a silent no-op — say so instead of passing vacuously note.append('park unverified (device already off bus)') return deadline = time.monotonic() + 6 while time.monotonic() < deadline: if find_device(board['uid'], None) is None: return time.sleep(0.5) note.append('park unverified: device still enumerated') if row['status'] == 'ok': row['status'] = 'flash-failed' def check_board_safe(board: dict, args, allow_recovery: bool, seen: dict) -> dict: """Isolate one board's exceptions: a crashing worker must not discard every other board's row, the table, the topology, and the seen-cache write.""" try: return check_board(board, args, allow_recovery, seen) except Exception as e: name = board.get('name', '?') say(f'{name:26} INTERNAL ERROR: {e!r}') return {'name': name, 'probe': '–', 'flash': '–', 'device': '❌ error', 'note': [repr(e)[:120]], 'status': 'failed'} def controller_summary() -> list[str]: """USB topology: controller (PCI addr, vendor) -> bus -> root-port subtree device counts (hubs included, interfaces/root hubs not). Bus numbers renumber every boot; PCI addresses and root-port numbers are stable.""" vendor_names = {'0x1022': 'AMD', '0x1912': 'Renesas', '0x8086': 'Intel', '0x1b21': 'ASMedia'} ctrl = {} for root in glob.glob('/sys/bus/usb/devices/usb*'): bus = int(os.path.basename(root)[3:]) m = re.findall(r'[0-9a-f]{4}:[0-9a-f]{2}:[0-9a-f]{2}\.[0-9a-f]', os.path.realpath(root)) pci = m[-1] if m else '?' c = ctrl.setdefault(pci, {'vendor': '?', 'buses': {}}) subtrees = {} for d in glob.glob(f'/sys/bus/usb/devices/{bus}-*'): b = os.path.basename(d) if ':' in b: continue subtrees[b.split('.')[0]] = subtrees.get(b.split('.')[0], 0) + 1 c['buses'][bus] = subtrees try: vid = open(f'/sys/bus/pci/devices/{pci}/vendor').read().strip() c['vendor'] = vendor_names.get(vid, vid) except OSError: pass lines = [] for pci, c in sorted(ctrl.items()): lines.append(f'{pci} ({c["vendor"]})') for bus, subtrees in sorted(c['buses'].items()): detail = ' '.join(f'{k}: {n} dev' for k, n in sorted(subtrees.items(), key=lambda i: int(i[0].split('-')[1]))) lines.append(f' bus {bus}: {sum(subtrees.values())} devices' + (f' {detail}' if detail else '')) return lines def main() -> None: # toolchain/flasher CLIs live in the user bin dirs (arm-none-eabi-gcc + esptool # in ~/.local/bin, STM32_Programmer_CLI in ~/bin) which non-login shells may # lack — same PATH shim hil_ci.sh applies on the remote side for d in (Path.home() / 'bin', Path.home() / '.local' / 'bin'): if d.is_dir() and str(d) not in os.environ.get('PATH', '').split(os.pathsep): os.environ['PATH'] = f'{d}{os.pathsep}{os.environ.get("PATH", "")}' parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument('config', nargs='?', help='HIL config json (default: by hostname)') parser.add_argument('-b', '--board', action='append', default=[], help='only these boards') parser.add_argument('-B', '--build-dir', default=None, help='firmware parent dir, searched EXCLUSIVELY when given ' '(default: examples, plus cmake-build as fallback)') parser.add_argument('--scan-only', action='store_true', help='USB presence scan only: no locks, no flashing') parser.add_argument('--no-build', action='store_true', help='do not build missing firmware (default: build the light example on the spot)') parser.add_argument('--no-park', action='store_true', help='leave the light example running (default: park with board_test)') # no cross-process flash budget with a concurrent hil_test.py run yet (would need # a file-lock budget in hil_lock; hil_test uses in-process semaphores) — keep modest parser.add_argument('-j', '--jobs', type=int, default=4) parser.add_argument('-v', '--verbose', action='store_true') args = parser.parse_args() global _no_build, _jobs, _build_sem _no_build = args.no_build _jobs = max(1, args.jobs) _build_sem = threading.BoundedSemaphore(_jobs) host = socket.gethostname() cfg_name = args.config or CONFIG_BY_HOST.get(host, 'local.json') cfg_path = Path(cfg_name) if not cfg_path.exists(): cfg_path = REPO_ROOT / 'test' / 'hil' / cfg_name if not cfg_path.exists(): sys.exit(f'config not found: {cfg_name} (host {host}; dev PCs need test/hil/local.json)') with cfg_path.open() as f: config = json.load(f) boards = list(config['boards']) # boards-skip (parked hardware) is not scanned by default if args.board: boards += config.get('boards-skip', []) # explicitly named parked boards are fair game unknown = set(args.board) - {b['name'] for b in boards} if unknown: sys.exit(f'board(s) not in {cfg_path.name}: {", ".join(sorted(unknown))}') boards = [b for b in boards if b['name'] in args.board] hil_flash.build_dir = args.build_dir or 'examples' hil_flash.verbose = args.verbose if args.build_dir is None: # default mode: search both standard layouts (cmake-build/ from tools/build.py # + ESP-IDF, examples/ from manual builds). An EXPLICIT -B is exclusive — the # caller named an artifact tree, so a miss must report, not silently flash an # older build from elsewhere. hil_test's -B is likewise untouched by this. hil_flash.EXTRA_BUILD_DIRS = ['cmake-build', 'examples'] allow_recovery = not args.scan_only and can_recover() seen = {} try: loaded = json.loads(SEEN_CACHE.read_text()) if isinstance(loaded, dict): # tolerate a torn/hand-edited cache seen = {k: v for k, v in loaded.items() if isinstance(v, dict)} except (OSError, ValueError): pass roots = ' + '.join(dict.fromkeys([hil_flash.build_dir, *hil_flash.EXTRA_BUILD_DIRS])) say(f'pool check: host {host}, config {cfg_path.name}, {len(boards)} boards, ' f'{"scan-only" if args.scan_only else f"flash via {{{roots}}}/cmake-build-"}' f'{"" if allow_recovery or args.scan_only else ", recovery unavailable (no sudo -n / usb_recover.sh)"}') if args.verbose: rows = [check_board_safe(b, args, allow_recovery, seen) for b in boards] else: with io.StringIO() as spool, ThreadPoolExecutor(max_workers=args.jobs) as pool: sys.stdout = spool # silence hil_flash's COMMAND FAILED dumps; say() uses __stdout__ try: rows = list(pool.map(lambda b: check_board_safe(b, args, allow_recovery, seen), boards)) finally: sys.stdout = sys.__stdout__ try: SEEN_CACHE.parent.mkdir(parents=True, exist_ok=True) tmp = SEEN_CACHE.with_suffix('.json.tmp') tmp.write_text(json.dumps(seen, indent=1, sort_keys=True) + '\n') tmp.replace(SEEN_CACHE) # atomic: a killed run can't tear the cache except OSError: pass status_mark = {'ok': '✅ ok', 'flash-failed': '❌ flash-failed', 'failed': '❌ failed', 'locked': '🔒 locked'} headers = ['Board', 'Probe', 'Flash', 'Device', 'Status', 'Note'] cells = [[r['name'], r['probe'], r['flash'], r['device'], status_mark.get(r['status'], r['status']), '; '.join(r['note'])] for r in rows] widths = [max(len(h), *(len(c[i]) for c in cells)) if cells else len(h) for i, h in enumerate(headers)] line = lambda vals: '| ' + ' | '.join(v.ljust(w) for v, w in zip(vals, widths)) + ' |' print() print(line(headers)) print('|' + '|'.join('-' * (w + 2) for w in widths) + '|') for c in cells: print(line(c)) print('\nUSB topology (controller → root-port subtree):') for line in controller_summary(): print(f' {line}') counts = {'ok': 0, 'flash-failed': 0, 'failed': 0, 'locked': 0} for r in rows: counts[r.get('status', 'failed')] += 1 print(f'\n{counts["ok"]} ok · {counts["flash-failed"]} flash-failed · {counts["failed"]} failed ' f'· {counts["locked"]} locked · in {time.monotonic() - t0:.0f}s') sys.exit(min(counts['flash-failed'] + counts['failed'], 125)) if __name__ == '__main__': main()