summaryrefslogtreecommitdiff
path: root/test/hil/hil_pool_check.py
diff options
context:
space:
mode:
authorHa Thach <[email protected]>2026-07-29 17:29:59 +0700
committerGitHub <[email protected]>2026-07-29 17:29:59 +0700
commite88fc441ddcaaa5abd4f4673ef2bf29499522dc0 (patch)
tree798ba5be8bff5b9433dfeab1d4b2a054fe28f679 /test/hil/hil_pool_check.py
parent538ec3e3320e41231c7e0bcef50a608db394de59 (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.py1013
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()