path: root/test/hil/helper/hil_util.py

#!/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 and imports nothing local -- 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)

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   # bound on one attribute read of a possibly-wedged device
SYSFS_STUCK_MAX = 4      # stranded readers tolerated before read_sysfs goes blind
_sysfs_stuck = 0         # each costs a thread + an fd for the life of the process
_sysfs_stuck_lock = threading.Lock()
_sysfs_blind_logged = False


class _SysfsUnknown:
    """Sentinel: the read did not answer. NOT "the attribute is absent" -- reading it as
    absence turns a healthy board into a firmware regression in the report."""
    __slots__ = ()

    def __bool__(self) -> bool:
        return False

    def __repr__(self) -> str:
        return 'SYSFS_UNKNOWN'


SYSFS_UNKNOWN = _SysfsUnknown()


def sysfs_blind() -> bool:
    """True once this process has stranded SYSFS_STUCK_MAX readers: every later read
    answers SYSFS_UNKNOWN, so nothing it reports about a device is a fact any more."""
    return _sysfs_stuck >= SYSFS_STUCK_MAX


def sysfs_blind_note() -> str:
    """Suffix for a failure message, so a blind worker's verdict never reads as hardware."""
    return (f' (this worker is blind: {SYSFS_STUCK_MAX} sysfs reads stranded on a wedged '
            f'device, so the check could not see the bus)') if sysfs_blind() else ''


def read_sysfs(path: str, grace: float = SYSFS_READ_GRACE) -> str | None | _SysfsUnknown:
    """Read a sysfs attribute with a WALL-CLOCK bound.

    The value, None when the attribute is genuinely unreadable (OSError), or SYSFS_UNKNOWN
    when the read did not answer -- it timed out, or this process is already blind. Callers
    MUST keep those apart: absence is a fact, unknown is not.

    usb_string_attr (serial/product/manufacturer) is served under the device lock a wedged
    usbfs ioctl holds, so a plain open().read() blocks for as long as the wedge lasts, on
    exactly the board an incident is about. The reader sleeps INTERRUPTIBLY (every read
    takes usb_lock_device_interruptible, v6.12.96 sysfs.c:124-139 -- uninterruptible is the
    ioctl holder, not us), so it dies with a SIGKILLed worker; what it costs meanwhile is a
    thread and an fd for this process's life, because on sysfs the open() SUCCEEDS and only
    the read blocks. Measured: 20 blocking reads leave 20 live threads.

    Hence the cap: callers rescan (hil_lock's controller_of re-reads every unresolved
    device on EVERY permit), and hitting RLIMIT_NOFILE or the thread ceiling raises inside
    the worker and loses every board's result -- worse than the hang this prevents.
    """
    if sysfs_blind():
        return SYSFS_UNKNOWN
    # Known-stranded? Re-reading costs another permanent thread+fd and a blindness credit
    # to learn what we already know. Lives HERE, not at the call sites: a call-site memo
    # has to be remembered by every new scanner, and twice it was not.
    was = _sysfs_stranded.get(path, _STRAND_MISS)
    if was is not _STRAND_MISS:
        if was is None:
            return SYSFS_UNKNOWN          # stranded, inode unknown: never re-read it
        try:
            if os.stat(path).st_ino == was:
                return SYSFS_UNKNOWN          # same node, still wedged
        except OSError:
            pass                              # gone: fall through, the read reports it
        _sysfs_stranded.pop(path, None)       # replaced or gone -> re-read it
    out: dict = {}

    def _read():
        try:
            with open(path) as f:
                out['v'] = f.read().strip()
        except (OSError, ValueError):
            pass      # no such attribute, or not text: unreadable, and that IS a fact

    t = threading.Thread(target=_read, daemon=True)
    t.start()
    t.join(grace)
    # `out` FIRST, not is_alive() alone: a reader can deposit its value and still be alive
    # for a moment afterwards, and counting that as a strand memoises a healthy attribute as
    # unreadable and spends one of four blindness credits. bounded_open has always checked
    # its box for the same reason.
    if t.is_alive() and 'v' not in out:
        # Count the PATH once, not once per reader. hil_pool_check runs -j4 by default,
        # which equals SYSFS_STUCK_MAX, so four threads hitting ONE wedged device used to
        # spend the entire blindness budget between them -- latching blind on the single
        # wedge the tool was run to find. The strand is real for each thread, but the
        # DEVICE is what the cap is about.
        # Under the SAME lock as the counter: check-then-act here is a race, and
        # hil_pool_check runs a ThreadPoolExecutor of exactly SYSFS_STUCK_MAX workers in
        # ONE process, so four threads on one wedged path could each see `first` before any
        # of them recorded it -- spending the whole blindness budget on a single device,
        # which is what this memo exists to prevent. note_sysfs_strand takes the lock
        # itself, so call it after releasing.
        with _sysfs_stuck_lock:
            first = path not in _sysfs_stranded
            if first:
                try:
                    # stat, never the thread's own open(): stat does not call ->show(), so
                    # it cannot block on the device lock the reader is stuck behind
                    _sysfs_stranded[path] = os.stat(path).st_ino
                except OSError:
                    _sysfs_stranded[path] = None   # unstattable, but still known-stranded
        if first:
            note_sysfs_strand()
        return SYSFS_UNKNOWN
    return out.get('v')


def note_sysfs_strand() -> None:
    """Record ONE stranded sysfs reader. Shared by read_sysfs and bounded_open so both
    account against a single counter -- the report caveat keys off it."""
    global _sysfs_stuck, _sysfs_blind_logged
    with _sysfs_stuck_lock:
        _sysfs_stuck += 1
        announce = sysfs_blind() and not _sysfs_blind_logged
        _sysfs_blind_logged = _sysfs_blind_logged or announce
    if announce:
        # once per process, on stderr: a worker's stdout is compacted into one report
        # row, where this would be lost among the test output
        print(f'warning: {SYSFS_STUCK_MAX} sysfs reads stranded on a wedged device; '
              f'this process is now blind and answers SYSFS_UNKNOWN for every '
              f'attribute -- its verdicts about device presence are not evidence',
              file=sys.stderr, flush=True)


# path -> the inode it had when its read stranded. A stranded attribute stays
# stranded until the DEVICE is replaced, and a re-enumeration destroys the kernfs
# node and makes a new one -- so a changed inode is the all-clear. Keyed by path
# alone it would outlive the wedge: a busport does not change when a board comes
# back on the same port, so the HUNG reflash this branch performs would recover a
# board the harness could then never see again.
_sysfs_stranded: dict = {}
# A stranded path whose inode could not be read is stored as None, so a plain .get() cannot
# tell 'known stranded, inode unknown' from 'never seen' -- and treating the first as the
# second re-reads it, stranding another permanent thread and fd every call. Distinct miss
# sentinel, so None keeps its own meaning.
_STRAND_MISS = object()


def usb_scan(vid_pid=None, serial=None, vid=None) -> tuple[list, bool]:
    """Enumerated USB devices matching the filters, and whether anything is unknown.

    Returns ([{busport, dir, vid, pid, serial}], unknown). `unknown` True means a bounded
    read did not answer, so absence is NOT proven -- the same contract as read_sysfs.

    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:124-139).
    * 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 served under the lock a wedged ioctl holds, and a
      path that already stranded is never re-read (each strand costs a thread and an fd
      for this process's life).
    """
    out = []
    unknown = False
    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'))
        if sn is SYSFS_UNKNOWN:
            unknown = True    # read_sysfs memoises it; a repeat scan costs nothing
            continue
        if sn is None:
            continue          # no serial attribute: a fact
        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, unknown


def bounded_open(path: str, flags: int, timeout: float = SYSFS_READ_GRACE):
    """os.open() with a wall-clock bound.

    The fd, None when the open genuinely FAILED (OSError: EBUSY, ENOENT, EACCES), or
    SYSFS_UNKNOWN when it did not answer -- the same three-valued contract as read_sysfs,
    and for the same reason: folding a fact into an unknown made an ordinary EBUSY read as
    a wedged device and sent the operator hunting hardware that is healthy.

    An open CAN block on a wedged device -- not on O_NONBLOCK, which usblp_open never
    consults, but on usb_autopm_get_interface(), a runtime-PM resume that does I/O
    (v6.12.96 drivers/usb/class/usblp.c). It holds usblp_mutex while it waits, and that
    mutex is driver-GLOBAL, so one wedged printer blocks opens of every usblp node.

    Unlike read_sysfs the stranded thread cleans up after itself: if we have given up it
    closes the fd it eventually got, so only the thread leaks. Both sides take `handoff`
    -- "store or close" and "abandon and drain" are a check-then-act pair that can
    interleave into an fd stored after the box was drained, which would leak it into a
    node that allows a SINGLE opener (usblp_open returns -EBUSY when usblp->used).
    """
    # Same short-circuit as read_sysfs: once blind, another stranded thread buys nothing
    # and the cap exists precisely to stop them accumulating.
    if sysfs_blind():
        return SYSFS_UNKNOWN
    # Known-stranded? Re-opening costs another thread, another fd and another blindness
    # credit to learn what we already know -- and the printer test re-opens ONE lp node on
    # every retry. Same memo and same inode check as read_sysfs.
    was = _sysfs_stranded.get(path, _STRAND_MISS)
    if was is not _STRAND_MISS:
        if was is None:
            return SYSFS_UNKNOWN          # stranded, inode unknown: never re-read it
        try:
            if os.stat(path).st_ino == was:
                return SYSFS_UNKNOWN
        except OSError:
            pass
        _sysfs_stranded.pop(path, None)
    box: dict = {}
    done, abandoned = threading.Event(), threading.Event()
    handoff = threading.Lock()

    def _open():
        try:
            fd = os.open(path, flags)
        except OSError:
            done.set()
            return
        with handoff:
            stored = not abandoned.is_set()
            if stored:
                box['fd'] = fd
        if not stored:
            try:
                os.close(fd)
            except OSError:
                pass
        done.set()

    threading.Thread(target=_open, daemon=True).start()
    if not done.wait(timeout):
        with handoff:
            abandoned.set()
            fd = box.pop('fd', None)  # completed in the gap between timeout and flag
        if fd is not None:
            # It DID open, just after our deadline -- the thread finished, so nothing is
            # stranded. Report unknown (we already gave up on it) but do not spend a
            # blindness credit, and do not call a merely-slow node wedged.
            try:
                os.close(fd)
            except OSError:
                pass
            return SYSFS_UNKNOWN
        # counted like a stranded read_sysfs: the thread and (eventually) its fd are gone
        # for the life of the process, and the cap exists to stop that reaching the
        # thread/fd ceiling -- an exception there escapes the worker and loses every board.
        # Memoised by inode so a retry of the same node does not pay again.
        # same lock as read_sysfs, same reason
        with _sysfs_stuck_lock:
            first = path not in _sysfs_stranded
            if first:
                try:
                    _sysfs_stranded[path] = os.stat(path).st_ino
                except OSError:
                    _sysfs_stranded[path] = None
        if first:
            note_sysfs_strand()
        return SYSFS_UNKNOWN
    return box.get('fd')


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=5)
            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()


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=10)
        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]