| #!/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, 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/<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, 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 (<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'], |
| 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-<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() |
