| #!/usr/bin/env python3 |
| # SPDX-License-Identifier: MIT |
| # Bottom layer of the HIL harness: the bounded command runner plus the shared helpers and |
| # data every other module needs. Stays stdlib-only; its one local dependency is |
| # tools/rtt.py (the RTT console, loaded by path below) -- everything |
| # else imports this, including the unit tests on GitHub's bare runner; never import them |
| # from here. Callers set the module global `verbose`. |
| |
| from __future__ import annotations |
| |
| import glob |
| import os |
| import signal |
| import subprocess |
| import unicodedata |
| import threading |
| import sys |
| from pathlib import Path |
| from typing import Any |
| |
| |
| # ------------------------------------------------------------- |
| # HIL example test lists, shared by hil_test.py (runner) and ci_select.py (PR-diff |
| # selector). Run order is shuffled per board (see test_board); every example carries a |
| # unique hardcoded idProduct (see its usb_descriptors.c). |
| # ------------------------------------------------------------- |
| |
| # device tests |
| device_tests = [ |
| 'device/cdc_dual_ports', |
| 'device/cdc_msc', |
| 'device/dfu', |
| 'device/cdc_msc_throughput', |
| 'device/audio_test_freertos', |
| 'device/dfu_runtime', |
| 'device/cdc_msc_freertos', |
| 'device/hid_boot_interface', |
| 'device/msc_dual_lun', |
| 'device/hid_generic_inout', |
| 'device/printer_to_cdc', |
| 'device/midi_test', |
| 'device/mtp', |
| 'device/usbtest', # cafe:4010, unique PID; runs the Linux testusb tier-4 battery via usbtest.py |
| # 'device/net_lwip_webserver', # disabled for PR #3605: USB net iface enum is flaky on the CI HIL host |
| ] |
| |
| dual_tests = [ |
| 'dual/host_info_to_device_cdc', |
| ] |
| |
| host_test = [ |
| 'host/cdc_msc_hid', |
| 'host/msc_file_explorer', |
| 'host/msc_file_explorer_freertos', |
| 'host/device_info', |
| ] |
| |
| verbose = False |
| |
| def pos_int_env(name: str, default: int) -> int: |
| # One parsing policy for every HIL_* knob: a bare int() crashes every run at import |
| # on a malformed value, and 0/negative silently removes the bound the knob enforces. |
| try: |
| v = int(os.getenv(name, str(default))) |
| except ValueError: |
| print(f'warning: {name} is not an integer; using {default}', |
| file=sys.stderr, flush=True) |
| return default |
| if v <= 0: |
| print(f'warning: {name}={v} is not usable; using {default}', |
| file=sys.stderr, flush=True) |
| return default |
| return v |
| |
| |
| def pos_float_env(name: str, default: float) -> float: |
| try: |
| v = float(os.getenv(name, str(default))) |
| except ValueError: |
| print(f'warning: {name} is not a number; using {default}', |
| file=sys.stderr, flush=True) |
| return default |
| # float() accepts 'inf'/'nan': an infinite serial timeout is an unbounded read, the |
| # very thing these knobs exist to prevent, and nan fails every comparison silently |
| if not (v > 0 and v < float('inf')): |
| print(f'warning: {name}={v} is not usable; using {default}', |
| file=sys.stderr, flush=True) |
| return default |
| return v |
| |
| |
| CMD_TIMEOUT = pos_int_env('HIL_CMD_TIMEOUT', 180) |
| # Post-SIGKILL reap, spent ON TOP of a run_cmd timeout whenever the child has to be killed. |
| # A caller budgeting several bounded steps must add one of these PER STEP, or its own outer |
| # bound fires mid-step -- for a flasher, orphaning it on the probe. |
| REAP_GRACE = 10 |
| |
| TINYUSB_ROOT = Path(__file__).resolve().parents[3] # test/hil/helper/ -> repo root |
| |
| |
| def display_width(s: str) -> int: |
| """Terminal COLUMNS, not characters. |
| |
| The status marks the reports use -- ✅ ❌ ⚪ ⚠ 🔒 -- are one Python character and TWO |
| columns wide. Measuring with len() pads every cell containing one a column short, so |
| the pipes drift out of line against the header rule for the whole table. |
| """ |
| return sum(2 if unicodedata.east_asian_width(c) in 'WF' else 1 for c in s) |
| |
| |
| def pad(s: str, width: int, center: bool = False) -> str: |
| """str.ljust/center, measured in display columns. See display_width.""" |
| room = max(0, width - display_width(s)) |
| if not center: |
| return s + ' ' * room |
| left = room // 2 |
| return ' ' * left + s + ' ' * (room - left) |
| |
| |
| def cmd_stdout_text(out: Any) -> str: |
| if out is None: |
| return '' |
| if isinstance(out, bytes): |
| return out.decode('utf-8', errors='ignore') |
| return str(out) |
| |
| |
| def _banner_body(out: Any, err: Any) -> str: |
| # split_stderr callers keep the diagnostic in stderr — a banner of stdout alone |
| # would be blank exactly when something went wrong |
| body = cmd_stdout_text(out) |
| err_text = cmd_stdout_text(err) |
| if err_text: |
| body = f'{body}\n{err_text}' if body else err_text |
| return body |
| |
| |
| # Shared with compact_output's stripper in hil_test: duplicated literals let the two |
| # layers drift and reintroduce literal marker noise mid-row in the GitHub log. |
| GROUP_MARK, ENDGROUP_MARK = '::group::', '::endgroup::' |
| |
| |
| def strip_workflow_markers(line: str) -> str: |
| # run_cmd only ever emits markers at line start; mid-line is not a real case. |
| return line.removeprefix(GROUP_MARK).removeprefix(ENDGROUP_MARK) |
| |
| |
| def _ci_log_groups() -> bool: |
| # GitHub folds ::group::/::endgroup:: only at line start of the JOB's real stdout; a |
| # pool worker's capture is compacted into one row line, where they render literally. |
| return bool(os.getenv('CI')) and sys.stdout is sys.__stdout__ |
| |
| |
| def _print_banner(title: str, out: Any, err: Any) -> None: |
| print() |
| if _ci_log_groups(): |
| print(f'{GROUP_MARK}{title}') |
| print(_banner_body(out, err)) |
| print(ENDGROUP_MARK) |
| else: |
| print(title) |
| print(_banner_body(out, err)) |
| |
| |
| SYSFS_READ_GRACE = 2.0 # default bound on one attribute read; see read_sysfs |
| |
| # path -> the kernfs inode the node had when its bounded read gave up. Keyed by INODE, not |
| # by path alone: a busport does not change when a board returns to the same physical port, |
| # so a path-only blacklist outlives the wedge -- hil_pool_check resets or reflashes the |
| # board, wait_device polls that busport for the new inode, and the scan it polls through |
| # would never look at the device again. A re-enumeration destroys the kernfs node and makes |
| # a new one, so a CHANGED inode is the all-clear. os.stat is safe on a wedged device: it |
| # does not call ->show(), so it cannot block on the lock the reader is stuck behind. |
| _stranded: dict = {} |
| _strand_hits: dict = {} # path -> how many times it has stranded, ever |
| _refused: set = set() # paths answered None WITHOUT reading, once past _STRAND_MAX |
| _strand_lock = threading.Lock() |
| _ever_stranded = False |
| |
| # Each strand costs a thread AND an fd for the life of the process -- on sysfs the open() |
| # SUCCEEDS and only the read blocks. Two ceilings, because they bound different things: |
| # |
| # _PATH_STRAND_MAX -- a device that FLAPS while still wedged re-enumerates, clears the |
| # inode memo, and strands again. Per path, so one sick board cannot leak without bound. |
| # After this many it stays memoised whatever its inode says. |
| # _STRAND_MAX -- a whole-process backstop against RLIMIT_NOFILE or the thread ceiling, |
| # which would raise inside a worker and lose every board's result. Counted PER PATH, not |
| # per reader: hil_pool_check runs four poll threads over one bus, and counting each |
| # reader let four threads on ONE wedged device spend four credits between them. With |
| # per-path counting a 27-board rig cannot approach this. |
| _PATH_STRAND_MAX = 4 |
| _STRAND_MAX = 64 |
| |
| |
| def sysfs_stranded() -> bool: |
| """True once any bounded read has given up, and it STAYS true. |
| |
| A sticky, process-wide fact, so it answers exactly one question: "could anything in |
| this process's output be the tool losing sight of healthy hardware?" -- which is what |
| hil_pool_check's footer needs. It canNOT answer "is THIS device unreadable" for a |
| caller deciding what a single missing device means; use path_stranded() for that. |
| """ |
| return _ever_stranded |
| |
| |
| def strand_note() -> str: |
| """Suffix for an absence claim, so "not found" never reads as proven absence. |
| |
| Lives here because every caller that can say "not found" needs the same sentence, and |
| the one that had to re-invent it got missed: a wedged-but-enumerated printer was |
| reported as an enumeration failure, sending a maintainer after firmware. |
| """ |
| return (' (a bounded sysfs read gave up, so "not found" here means "could not tell"' |
| ' -- see the usb-kernel-recover skill)') if sysfs_stranded() else '' |
| |
| |
| def path_stranded(path: str) -> bool: |
| """Whether THIS attribute is currently memoised as unreadable. |
| |
| The per-device question sysfs_stranded() cannot answer. usbtest uses it to tell a DUT |
| whose `serial` is held under device_lock from one that genuinely left the bus, because |
| the difference decides whether it performs driver-registry writes that take the |
| UNINTERRUPTIBLE device_lock. |
| """ |
| with _strand_lock: |
| return path in _stranded or path in _refused |
| |
| |
| def read_sysfs(path: str, timeout: float = SYSFS_READ_GRACE) -> str | None: |
| """A sysfs attribute's value, or None when it did not answer. |
| |
| BOUNDED BY DEFAULT, and it has to be. `serial` is served by usb_string_attr, which |
| takes usb_lock_device_interruptible (v6.12.96 sysfs.c:141-143) -- the same lock a |
| wedged usbfs ioctl holds. Every OTHER attribute the harness reads (idVendor, idProduct, |
| bcdDevice, busnum, devnum, speed) is a lock-free sysfs_emit from a cached field and |
| cannot block. |
| |
| "Only the wedged board's own worker pays" is FALSE, which is why the bound is not |
| opt-in: usb_scan reads `serial` on every device matching the VID to find the one it |
| wants, so resolving MY board touches every peer's locked attribute. hil_lock's |
| controller_of does that from controller_permit, on essentially every board -- one |
| wedged DUT would stall every worker, not one. hil_pool_check has no guard at all. |
| |
| A give-up reads as None, the same as unreadable: there is no third value and no |
| per-attribute blindness. The memo is keyed by inode so the cost stays on the device |
| that is actually wedged; path_stranded() tells a caller which device that was. |
| """ |
| with _strand_lock: |
| was = _stranded.get(path) |
| stuck_for_good = _strand_hits.get(path, 0) >= _PATH_STRAND_MAX |
| budget_spent = len(_stranded) >= _STRAND_MAX |
| if was is not None: |
| try: |
| if os.stat(path).st_ino == was: |
| return None # same kernfs node, still wedged |
| except OSError: |
| pass # gone: let the read below report it |
| if stuck_for_good: |
| return None # flapped too many times; see _PATH_STRAND_MAX |
| with _strand_lock: |
| _stranded.pop(path, None) # a different inode is the all-clear |
| elif budget_spent: |
| # see _STRAND_MAX. Recorded, not just returned: usbtest fails CLOSED on |
| # path_stranded() before the lock-taking cleanup, and a path we declined to read |
| # is exactly the case it must not be told is readable-and-absent. |
| with _strand_lock: |
| _refused.add(path) |
| return None |
| |
| # BEFORE the read, not after: a node that re-enumerates DURING the grace would |
| # otherwise have its brand-new HEALTHY inode recorded as the wedged one, and only a |
| # second re-enumeration could ever clear it. If it cannot be stat'd there is no key to |
| # memoise against, so the path is simply re-read next time -- the open fails fast. |
| try: |
| ino = os.stat(path).st_ino |
| except OSError: |
| ino = None |
| out: dict = {} |
| |
| def _read(): |
| try: |
| with open(path) as f: |
| out['v'] = f.read().strip() |
| except (OSError, ValueError): |
| pass |
| |
| t = threading.Thread(target=_read, daemon=True) |
| t.start() |
| t.join(timeout) |
| # `out` FIRST: a reader can deposit its value and still be alive for a moment |
| # afterwards, and counting that as a strand blacklists a healthy attribute forever |
| if 'v' in out: |
| # a path that answered is not refused any more: _refused feeds path_stranded(), |
| # and a stale entry makes usbtest read a LATER genuine disconnect as "cannot tell" |
| with _strand_lock: |
| _refused.discard(path) |
| if t.is_alive() and 'v' not in out: |
| global _ever_stranded |
| announce = False |
| if ino is None: |
| # the pre-read stat lost a race the open then won -- the node was replaced |
| # between them. Re-stat now: the reader is blocked on whatever node exists, |
| # so this is the key it is stuck on. Without a key nothing is memoised and |
| # every later poll starts another permanent thread and fd for this path. |
| try: |
| ino = os.stat(path).st_ino |
| except OSError: |
| pass |
| with _strand_lock: |
| _ever_stranded = True |
| if ino is not None: |
| first = path not in _stranded # count the PATH once, not each reader |
| _stranded[path] = ino |
| if first: |
| _strand_hits[path] = _strand_hits.get(path, 0) + 1 |
| announce = len(_stranded) == _STRAND_MAX |
| else: |
| _refused.add(path) # unkeyable: at least do not vouch for it |
| if announce: |
| print(f'warning: {_STRAND_MAX} devices have unreadable sysfs attributes; ' |
| f'refusing to start more bounded readers, so later reads answer None ' |
| f'without looking. Find the wedged device (usb-kernel-recover skill).', |
| file=sys.stderr, flush=True) |
| return None |
| return out.get('v') |
| |
| |
| def usb_scan(vid_pid=None, serial=None, vid=None, timeout=SYSFS_READ_GRACE) -> list: |
| """Enumerated USB devices matching the filters: [{busport, dir, vid, pid, serial}]. |
| |
| Three rules, one implementation for every caller: |
| |
| * Root hubs excluded (glob `*-*`): no DUT is one, and scans including them measured |
| seconds slower (observation, no mechanism -- the "autosuspend wake" explanation was |
| wrong; usb_string_attr reads a cached string, sysfs.c:141-143). |
| * idVendor/idProduct first: lock-free `sysfs_emit` from udev->descriptor |
| (sysfs.c:688-705), so they rule out nearly every device for free. |
| * `serial` LAST and BOUNDED: it is the only attribute here served under the device |
| lock, so it is the only one that can block. Filtering on the lock-free pair first |
| keeps most devices out of it, but a scan for ONE board still reads the serial of |
| every peer that shares its VID -- so the bound is what stops one wedged DUT from |
| stalling every caller (see read_sysfs). |
| """ |
| out = [] |
| for d in glob.glob('/sys/bus/usb/devices/*-*'): |
| # Interfaces are '<busport>:<cfg>.<ifnum>' (e.g. 2-4:1.0) -- they CONTAIN the |
| # colon, they do not end with it, so the original endswith() never fired and every |
| # scan opened idVendor/idProduct on all of them (measured: 31 of 44 matches). |
| if ':' in os.path.basename(d): |
| continue |
| try: |
| with open(os.path.join(d, 'idVendor')) as f: |
| dev_vid = f.read().strip() |
| with open(os.path.join(d, 'idProduct')) as f: |
| dev_pid = f.read().strip() |
| except OSError: |
| continue # vanished mid-walk, or not a device dir: a fact, not unknown |
| if vid_pid is not None and (dev_vid, dev_pid) != tuple(vid_pid): |
| continue # ruled out for free, without touching the locked attribute |
| if vid is not None and dev_vid != vid: |
| continue # same, for callers that know the VID but not the PID |
| sn = read_sysfs(os.path.join(d, 'serial'), timeout) |
| if sn is None: |
| continue # no serial attribute |
| if serial is not None and sn.lower() != serial.lower(): |
| continue |
| out.append({'busport': os.path.basename(d), 'dir': d, |
| 'vid': dev_vid, 'pid': dev_pid, 'serial': sn}) |
| return out |
| |
| |
| def _close_pipes(p: subprocess.Popen) -> None: |
| """Close OUR ends of an abandoned child's pipes. Never raises.""" |
| for pipe in (p.stdout, p.stderr, p.stdin): |
| try: |
| if pipe is not None: |
| pipe.close() |
| except OSError: |
| pass |
| |
| |
| def run_alongside(argv: list, work, timeout: int) -> subprocess.CompletedProcess: |
| """Run `argv` alongside `work()`, which runs in THIS thread, then reap it -- bounded. |
| |
| The read-while-we-write shape run_cmd cannot express: the caller needs the child |
| RUNNING while it does something else. Everything else about the contract is run_cmd's |
| -- own session, killpg, bounded reap, our pipe ends closed, rc 124 on the kill. |
| |
| A PROCESS, not a thread: an abandoned thread keeps the fd, and usblp_open returns |
| -EBUSY while usblp->used (v6.12.96 usblp.c), so every later open in this long-lived |
| worker would read as a wedged device. A killed process takes its fd with it. |
| |
| stdout is captured as BYTES and kept CLEAN -- a caller byte-compares it against the |
| payload it sent, so a single stderr byte (a PYTHONWARNINGS chirp, a sitecustomize |
| print, a .pth deprecation from a venv) would read as USB data corruption. stderr gets |
| its own pipe; communicate() drains both, so the split cannot deadlock. |
| `work` runs even if the child dies immediately -- the caller's own asserts decide. |
| """ |
| p = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, |
| start_new_session=True) |
| |
| def _reap() -> subprocess.CompletedProcess: |
| try: |
| out, err = p.communicate(timeout=timeout) |
| return subprocess.CompletedProcess(argv, p.returncode, out, err) |
| except subprocess.TimeoutExpired: |
| try: |
| os.killpg(p.pid, signal.SIGKILL) |
| except OSError: |
| p.kill() |
| try: |
| out, err = p.communicate(timeout=REAP_GRACE) |
| except subprocess.TimeoutExpired: |
| # Outlasted SIGKILL: uninterruptible, still holding whatever it opened. |
| # Abandoned like any other stray -- but as a real child in its own |
| # session, so the containment sweep FINDS it (child_procs walks the ppid |
| # tree) and the report names it. That is the whole difference from a |
| # blocked thread, which no sweep can see and no signal can reach. |
| out, err = b'', b'' |
| _close_pipes(p) # our own fds must not leak either |
| return subprocess.CompletedProcess(argv, 124, out, err) |
| |
| try: |
| work() |
| except BaseException: |
| # Reap first so the child never outlives us, then let the caller's error through. |
| # A `return` inside a `finally` would SWALLOW it -- an assert in `work` would |
| # vanish and the caller would compare data it never finished sending. |
| _reap() |
| raise |
| return _reap() |
| |
| |
| # The RTT console implementation lives in tools/rtt.py (importable classes + CLI, |
| # stdlib-only, harness-critical — see its module docstring). Loaded by file path so |
| # no sys.path entry for tools/ can shadow other imports; re-exported here so the |
| # harness keeps addressing hil_util.JlinkRtt. |
| import importlib.util as _ilu |
| |
| _rtt_path = TINYUSB_ROOT / 'tools' / 'rtt.py' |
| if not _rtt_path.exists(): |
| # name the real cause: a bare FileNotFoundError out of an exec_module here reads |
| # as a harness bug, when the actual problem is an incompletely staged tree |
| raise ImportError(f'{_rtt_path} is missing — the RTT console lives there and the ' |
| f'harness depends on it; stage it alongside test/hil (hil_ci.sh does)') |
| _rtt_spec = _ilu.spec_from_file_location('tinyusb_tools_rtt', _rtt_path) |
| _rtt = _ilu.module_from_spec(_rtt_spec) |
| sys.modules[_rtt_spec.name] = _rtt # registered: RttError must be picklable across the fork Pool |
| _rtt_spec.loader.exec_module(_rtt) |
| JlinkRtt = _rtt.JlinkRtt |
| OpenocdRtt = _rtt.OpenocdRtt |
| RttError = _rtt.RttError |
| RTT_BANNER_RE = _rtt.RTT_BANNER_RE |
| strip_banner = _rtt.strip_banner |
| |
| |
| def _cmd_label(cmd) -> str: |
| """A one-line name for a banner. An argv whose payload is a `python3 -c` program would |
| otherwise dump the whole body into the CI log, where run_cmd's banners are already the |
| noisiest thing in a failing row.""" |
| if isinstance(cmd, str): |
| return cmd |
| parts = [a if len(a) <= 60 else f'<{len(a)}-char program>' for a in cmd] |
| return ' '.join(parts) |
| |
| |
| def run_cmd(cmd: str | list, cwd: str | None = None, timeout: int | None = None, |
| binary: bool = False, split_stderr: bool = False, |
| quiet: bool = False) -> subprocess.CompletedProcess: |
| """Bounded subprocess: own session, killpg on expiry, rc 124 when it had to be killed. |
| |
| `cmd` is a shell STRING or an argv LIST. argv exists for a program that cannot survive |
| a trip through the shell -- a multi-line `python3 -c` body -- which is how the harness |
| runs a library call that no in-process bound can contain. A daemon thread cannot bound |
| a C call that holds the GIL, so for those the child process IS the bound. |
| """ |
| if timeout is None: |
| timeout = CMD_TIMEOUT |
| # binary: raw bytes (text mode's errors='replace' mangles non-UTF-8 file content). |
| # split_stderr: keep stderr out of stdout, for callers that parse stdout. quiet: no |
| # COMMAND FAILED banner, for retry loops that report failures themselves (timeouts |
| # still print: a killed child is always noteworthy). |
| popen_kwargs = { |
| 'cwd': cwd, |
| # a list goes straight to execve; only a string needs a shell to parse it |
| 'shell': isinstance(cmd, str), |
| 'stdout': subprocess.PIPE, |
| 'stderr': subprocess.PIPE if split_stderr else subprocess.STDOUT, |
| } |
| if not binary: |
| popen_kwargs.update({'text': True, 'encoding': 'utf-8', 'errors': 'replace'}) |
| # C-level setsid, same process-group semantics as preexec_fn=os.setsid but safe when |
| # called from threads (pool_check runs flashes from a thread pool) |
| popen_kwargs['start_new_session'] = True |
| |
| p = subprocess.Popen(cmd, **popen_kwargs) |
| try: |
| out, err = p.communicate(timeout=timeout) |
| r = subprocess.CompletedProcess(args=cmd, returncode=p.returncode, stdout=out, stderr=err) |
| except subprocess.TimeoutExpired as ex: |
| try: |
| os.killpg(p.pid, signal.SIGKILL) |
| except OSError: |
| # ProcessLookupError: already gone. PermissionError: an all-root group refuses |
| # the group kill -- letting either escape would skip the bounded reap, the pipe |
| # close and the rc-124 return this handler exists for. |
| pass |
| try: |
| out, err = p.communicate(timeout=REAP_GRACE) |
| except subprocess.TimeoutExpired: |
| # Something in the group outlived SIGKILL: D state (truly unkillable), or |
| # root-owned because sudo FORKS rather than execs, so the wrapper dies and its |
| # root child does not. Abandon it and let the report name it; the harness never |
| # sudo-kills its way out. Our ends of its pipes must not leak, though: a pool |
| # worker lives for the whole run, so every wedged command would cost it two fds. |
| out, err = None, None |
| _close_pipes(p) |
| # prefer the post-kill buffers (supersets of the exception's), falling back to ex.* |
| # when the child was unkillable. TimeoutExpired carries BYTES even for a text-mode |
| # Popen, so the fallbacks must be decoded or a text-mode caller gets bytes exactly |
| # when the child wedged in D state. |
| def _typed(v): |
| if not binary and isinstance(v, bytes): |
| return v.decode('utf-8', errors='replace') |
| return v |
| |
| timeout_out = _typed(out or ex.stdout) or (b'' if binary else '') |
| # ...and never None: with split_stderr the SUCCESS path always yields a str/bytes, |
| # so a caller that does `r.stderr.strip()` works everywhere except the timeout -- |
| # the one path it was written for. Without split_stderr stderr stays None, as on |
| # the success path (it was merged into stdout). |
| timeout_err = _typed(err if err is not None else ex.stderr) |
| if split_stderr and timeout_err is None: |
| timeout_err = b'' if binary else '' |
| _print_banner(f'COMMAND TIMEOUT ({timeout}s): {_cmd_label(cmd)}', timeout_out, timeout_err) |
| return subprocess.CompletedProcess(args=cmd, returncode=124, stdout=timeout_out, stderr=timeout_err) |
| except BaseException: |
| # BaseException, not Exception (as in CPython's own subprocess.run): |
| # KeyboardInterrupt is the case that matters, and start_new_session put the child in |
| # its OWN group, so it never got the terminal's SIGINT -- without this, Ctrl-C |
| # leaves the flasher or testusb holding the probe and its usbfs node. Kill and |
| # close, never wait: this path must not add a hang of its own. |
| try: |
| os.killpg(p.pid, signal.SIGKILL) |
| except OSError: |
| pass |
| _close_pipes(p) |
| raise |
| |
| if r.returncode != 0 and not quiet: |
| _print_banner(f'COMMAND FAILED: {_cmd_label(cmd)}', r.stdout, r.stderr) |
| elif verbose: |
| print(cmd) |
| print(cmd_stdout_text(r.stdout)) |
| return r |
| |
| |
| # get usb serial by id |
| def get_serial_dev(id, vendor_str, product_str, ifnum): |
| if vendor_str and product_str: |
| # known vendor and product |
| vendor_str = vendor_str.replace(' ', '_') |
| product_str = product_str.replace(' ', '_') |
| return f'/dev/serial/by-id/usb-{vendor_str}_{product_str}_{id}-if{ifnum:02d}' |
| else: |
| # just use id: mostly for cp210x/ftdi flasher |
| pattern = f'/dev/serial/by-id/usb-*_{id}-if*' |
| port_list = glob.glob(pattern) |
| if len(port_list) == 0: |
| raise RuntimeError(f'No serial device found for {pattern}') |
| return port_list[0] |
