diff options
| author | Ha Thach <[email protected]> | 2026-07-29 17:29:59 +0700 |
|---|---|---|
| committer | GitHub <[email protected]> | 2026-07-29 17:29:59 +0700 |
| commit | e88fc441ddcaaa5abd4f4673ef2bf29499522dc0 (patch) | |
| tree | 798ba5be8bff5b9433dfeab1d4b2a054fe28f679 /test/hil/hil_pool_check.py | |
| parent | 538ec3e3320e41231c7e0bcef50a608db394de59 (diff) | |
hil: split hil_test.py into hil_lock/hil_flash, add pool_check, update rig probes (#3794)
test/hil: add board-pool health check, split hil_test into focused modules (#3794)
Add test/hil/hil_pool_check.py: per-board rig health scan — probe presence,
light-example flash (dfu_runtime; device_info + serial check for host-only
boards), uid re-enumeration, safe recovery (probe authorized-toggle, board
reset), verified board_test re-park, USB topology report, and a markdown
summary table. Missing firmware is built on the spot (tools/build.py, idf.py
for espressif, one get_deps retry); row statuses: ok, flash-failed, failed,
locked. Board locks are always respected, never bypassed.
Refactor hil_test.py into hil_lock.py (flock protocol, controller permits,
hold/release/status CLI; replaces board_lock.py) and hil_flash.py (flashers,
find_firmware, run_cmd). Update WCH probe uids and the board roster in
tinyusb.json; add the hil-pool-check skill.
Diffstat (limited to 'test/hil/hil_pool_check.py')
| -rw-r--r-- | test/hil/hil_pool_check.py | 1013 |
1 files changed, 1013 insertions, 0 deletions
diff --git a/test/hil/hil_pool_check.py b/test/hil/hil_pool_check.py new file mode 100644 index 000000000..63284213e --- /dev/null +++ b/test/hil/hil_pool_check.py @@ -0,0 +1,1013 @@ +#!/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 + <bus>-<port>[.<port>...] 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): + 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 base + path to flash. 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) + 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/<example>/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/<ex>/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) + 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 (<variant>/<example>/): 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']) + _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') + 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') 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-<board>"}' + 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() |
