#!/usr/bin/env python3
"""Ground-truth HS edge logger — Pi GPIO tap on the POG-COM SLV_HS net.

Why this exists: both USB observation paths are blind below a few ms. The XR21
GPIO interrupt endpoint is hard-quantized to 8 ms (`bInterval`), and the uartd
GET_REG poll floors at ~2.75 ms with a 1.67 ms transport cost. Neither can prove
whether HS windows shorter than ~3 ms exist. A Pi GPIO edge interrupt timestamps
in the kernel at tens of microseconds, so it can.

Wiring and rationale: see HS_GPIO_TAP.md. Tap the `SLV_HS` net (THVD1420
receiver output, R6/LED6 pads) — never `SLV-IN`, which sits on the TXS0108E
level shifter that the live bridge depends on.

HARDWARE WARNING: `SLV_HS` is in the HAT's **5 V** domain (VCC = +5 V; only VDD
is 3.3 V — that is why the TXS0108E is there). Pi GPIO is 3.3 V only and RP1 is
not 5 V tolerant. Level-shift with a 74LVC1G17 or a 1.8k/3.3k divider first; a
plain series resistor is not protection. See HS_GPIO_TAP.md.

Timestamps: kernel `timestamp_ns` on CLOCK_MONOTONIC, the same base uartd uses
(`clock_gettime(CLOCK_MONOTONIC)`) and the XR driver uses (`ktime_get_ns()`).
All three sources are therefore directly comparable — that is the whole point.

Because the kernel stamps the edge, userspace scheduling delay cannot corrupt a
timestamp; it only delays delivery. `recv_ns` is recorded per edge so drain lag
is measurable rather than assumed (the uartd kernel ring hid a 130 ms artifact
this way).

Usage:
  python3 hs_gpio_logger.py --line 9 --output-dir ./data
  python3 hs_gpio_logger.py --self-test
"""

from __future__ import annotations

import argparse
import errno
import json
import os
import signal
import struct
import sys
import time
from pathlib import Path
from typing import Any, Optional

# fcntl/select.poll are Unix-only; keep them optional so --self-test runs on the PC.
try:
    import fcntl
except ImportError:  # pragma: no cover - Windows dev host
    fcntl = None  # type: ignore[assignment]

try:
    from pog422_core import EG_VERSION
except Exception:  # analyzer/logger must run on hosts without the full tree
    EG_VERSION = "unknown"

TRACE_VERSION = 1

GPIO_MAX_NAME_SIZE = 32
GPIO_V2_LINES_MAX = 64
GPIO_V2_LINE_NUM_ATTRS_MAX = 10

SIZE_CHIPINFO = 68
SIZE_LINE_ATTRIBUTE = 16
SIZE_LINE_CONFIG_ATTRIBUTE = 24
SIZE_LINE_CONFIG = 272
SIZE_LINE_REQUEST = 592
SIZE_LINE_EVENT = 48
SIZE_LINE_VALUES = 16

FLAG_ACTIVE_LOW = 1 << 1
FLAG_INPUT = 1 << 2
FLAG_EDGE_RISING = 1 << 4
FLAG_EDGE_FALLING = 1 << 5
FLAG_BIAS_DISABLED = 1 << 10
FLAG_EVENT_CLOCK_REALTIME = 1 << 11

EVENT_RISING_EDGE = 1
EVENT_FALLING_EDGE = 2

_IOC_NRBITS, _IOC_TYPEBITS, _IOC_SIZEBITS = 8, 8, 14
_IOC_NRSHIFT = 0
_IOC_TYPESHIFT = _IOC_NRSHIFT + _IOC_NRBITS
_IOC_SIZESHIFT = _IOC_TYPESHIFT + _IOC_TYPEBITS
_IOC_DIRSHIFT = _IOC_SIZESHIFT + _IOC_SIZEBITS
_IOC_WRITE, _IOC_READ = 1, 2


def _ioc(direction: int, typ: int, nr: int, size: int) -> int:
    """Encode an ioctl number, wrapped to signed so fcntl.ioctl accepts it."""
    value = (
        (direction << _IOC_DIRSHIFT)
        | (size << _IOC_SIZESHIFT)
        | (typ << _IOC_TYPESHIFT)
        | (nr << _IOC_NRSHIFT)
    )
    if value >= 1 << 31:
        value -= 1 << 32
    return value


GPIO_GET_CHIPINFO_IOCTL = _ioc(_IOC_READ, 0xB4, 0x01, SIZE_CHIPINFO)
GPIO_V2_GET_LINE_IOCTL = _ioc(_IOC_READ | _IOC_WRITE, 0xB4, 0x07, SIZE_LINE_REQUEST)
GPIO_V2_LINE_GET_VALUES_IOCTL = _ioc(_IOC_READ | _IOC_WRITE, 0xB4, 0x0E, SIZE_LINE_VALUES)

# Header GPIO controller labels: RP1 is Pi 5 / CM5, bcm2711 is Pi 4 / CM4.
PINCTRL_LABELS = ("pinctrl-rp1", "pinctrl-bcm2711", "pinctrl-bcm2835", "pinctrl-bcm2712")


def mono_ns() -> int:
    return time.clock_gettime_ns(time.CLOCK_MONOTONIC)


def realtime_ns() -> int:
    return time.clock_gettime_ns(time.CLOCK_REALTIME)


class GpioChipInfo:
    __slots__ = ("path", "name", "label", "lines")

    def __init__(self, path: str, name: str, label: str, lines: int) -> None:
        self.path = path
        self.name = name
        self.label = label
        self.lines = lines

    def __repr__(self) -> str:
        return f"GpioChipInfo({self.path!r}, label={self.label!r}, lines={self.lines})"


def read_chip_info(path: str) -> Optional[GpioChipInfo]:
    try:
        fd = os.open(path, os.O_RDONLY | os.O_CLOEXEC)
    except OSError:
        return None
    try:
        buf = bytearray(SIZE_CHIPINFO)
        fcntl.ioctl(fd, GPIO_GET_CHIPINFO_IOCTL, buf, True)
    except OSError:
        return None
    finally:
        os.close(fd)
    name = buf[0:GPIO_MAX_NAME_SIZE].split(b"\x00", 1)[0].decode("ascii", "replace")
    label = buf[GPIO_MAX_NAME_SIZE : 2 * GPIO_MAX_NAME_SIZE]
    label = label.split(b"\x00", 1)[0].decode("ascii", "replace")
    (lines,) = struct.unpack_from("<I", buf, 2 * GPIO_MAX_NAME_SIZE)
    return GpioChipInfo(path, name, label, lines)


def list_chips() -> list[GpioChipInfo]:
    out: list[GpioChipInfo] = []
    dev = Path("/dev")
    if not dev.is_dir():
        return out
    for entry in sorted(dev.glob("gpiochip*")):
        info = read_chip_info(str(entry))
        if info is not None:
            out.append(info)
    return out


def find_header_chip(explicit: Optional[str] = None) -> GpioChipInfo:
    """Locate the SoC header GPIO controller.

    Pi 5 / CM5 renumbered the RP1 chip between kernel releases, so match on the
    controller label instead of a hardcoded /dev/gpiochipN.
    """
    if explicit:
        info = read_chip_info(explicit)
        if info is None:
            raise SystemExit(f"FATAL: cannot read chip info from {explicit}")
        return info
    chips = list_chips()
    if not chips:
        raise SystemExit("FATAL: no /dev/gpiochip* found (not Linux, or no GPIO access)")
    for label in PINCTRL_LABELS:
        for chip in chips:
            if chip.label == label:
                return chip
    # Fall back to the first controller exposing at least the 28 header lines.
    for chip in chips:
        if chip.lines >= 28:
            return chip
    listing = ", ".join(f"{c.path}:{c.label}({c.lines})" for c in chips)
    raise SystemExit(f"FATAL: no header GPIO controller among: {listing}")


def build_line_request(
    line: int,
    consumer: str,
    *,
    event_buffer_size: int,
    bias_disabled: bool,
) -> bytearray:
    flags = FLAG_INPUT | FLAG_EDGE_RISING | FLAG_EDGE_FALLING
    if bias_disabled:
        flags |= FLAG_BIAS_DISABLED
    buf = bytearray(SIZE_LINE_REQUEST)
    struct.pack_into("<I", buf, 0, line)
    name = consumer.encode("ascii", "replace")[: GPIO_MAX_NAME_SIZE - 1]
    off = GPIO_V2_LINES_MAX * 4
    buf[off : off + len(name)] = name
    cfg = off + GPIO_MAX_NAME_SIZE
    struct.pack_into("<Q", buf, cfg, flags)
    struct.pack_into("<I", buf, cfg + 8, 0)
    tail = cfg + SIZE_LINE_CONFIG
    struct.pack_into("<I", buf, tail, 1)
    struct.pack_into("<I", buf, tail + 4, event_buffer_size)
    return buf


def request_line(
    chip: GpioChipInfo,
    line: int,
    consumer: str,
    *,
    event_buffer_size: int = 1024,
    bias_disabled: bool = True,
) -> int:
    if line >= chip.lines:
        raise SystemExit(f"FATAL: line {line} out of range for {chip.path} ({chip.lines} lines)")
    fd = os.open(chip.path, os.O_RDONLY | os.O_CLOEXEC)
    try:
        buf = build_line_request(
            line, consumer, event_buffer_size=event_buffer_size, bias_disabled=bias_disabled
        )
        try:
            fcntl.ioctl(fd, GPIO_V2_GET_LINE_IOCTL, buf, True)
        except OSError as exc:
            if exc.errno == errno.EBUSY:
                raise SystemExit(
                    f"FATAL: GPIO line {line} on {chip.path} is already claimed.\n"
                    f"  Check with: pinctrl get {line}   (or: gpioinfo)\n"
                    f"  If SPI0 owns it, disable SPI in /boot/firmware/config.txt."
                ) from exc
            if exc.errno in (errno.EACCES, errno.EPERM):
                raise SystemExit(
                    f"FATAL: permission denied on {chip.path}.\n"
                    f"  Add the user to the gpio group, or run with sudo."
                ) from exc
            raise SystemExit(f"FATAL: GPIO_V2_GET_LINE failed on line {line}: {exc}") from exc
        (req_fd,) = struct.unpack_from("<i", buf, SIZE_LINE_REQUEST - 4)
        if req_fd < 0:
            raise SystemExit(f"FATAL: kernel returned invalid request fd {req_fd}")
        return req_fd
    finally:
        os.close(fd)


def read_line_value(req_fd: int) -> Optional[int]:
    buf = bytearray(SIZE_LINE_VALUES)
    struct.pack_into("<QQ", buf, 0, 0, 1)
    try:
        fcntl.ioctl(req_fd, GPIO_V2_LINE_GET_VALUES_IOCTL, buf, True)
    except OSError:
        return None
    (bits, _mask) = struct.unpack_from("<QQ", buf, 0)
    return int(bits & 1)


class EdgeWriter:
    """Line-buffered JSONL sink. Flushes on an interval so a Ctrl-C keeps data.

    Also detects CLEAR_CAPTURE unlinking the path out from under the open fd
    (common when the bridge starts after this logger) and reopens the path so
    subsequent edges land in a file rclone can see.
    """

    def __init__(self, path: Path, flush_interval_s: float = 1.0) -> None:
        self.path = path
        self.flush_interval_s = flush_interval_s
        path.parent.mkdir(parents=True, exist_ok=True)
        self._fp = path.open("a", encoding="utf-8")
        self._last_flush = mono_ns()
        self._reopens = 0

    def _fd_matches_path(self) -> bool:
        try:
            st_path = self.path.stat()
            st_fd = os.fstat(self._fp.fileno())
        except OSError:
            return False
        return st_path.st_ino == st_fd.st_ino and st_path.st_dev == st_fd.st_dev

    def _reopen_if_unlinked(self) -> None:
        if self._fd_matches_path():
            return
        self._reopens += 1
        print(
            f"WARNING: {self.path} was replaced/unlinked under the open fd "
            f"(CLEAR_CAPTURE wipe?). Reopening so edges stay visible to sync "
            f"(reopen #{self._reopens}).",
            flush=True,
        )
        try:
            self._fp.flush()
        except Exception:
            pass
        try:
            self._fp.close()
        except Exception:
            pass
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self._fp = self.path.open("a", encoding="utf-8")
        self._last_flush = mono_ns()
        # Marker so analysis can see a mid-session reopen.
        self._fp.write(
            json.dumps(
                {
                    "v": TRACE_VERSION,
                    "src": "hs_gpio",
                    "type": "reopen",
                    "mono_ns": mono_ns(),
                    "reopen": self._reopens,
                    "note": "path_unlinked_under_fd_likely_clear_capture",
                    "path": str(self.path),
                },
                separators=(",", ":"),
            )
            + "\n"
        )
        self._fp.flush()

    def write(self, rec: dict[str, Any]) -> None:
        self._reopen_if_unlinked()
        rec.setdefault("v", TRACE_VERSION)
        rec.setdefault("src", "hs_gpio")
        self._fp.write(json.dumps(rec, separators=(",", ":")) + "\n")
        now = mono_ns()
        if now - self._last_flush >= int(self.flush_interval_s * 1e9):
            self.flush()

    def flush(self) -> None:
        self._reopen_if_unlinked()
        self._fp.flush()
        self._last_flush = mono_ns()

    def close(self) -> None:
        try:
            self.flush()
        finally:
            self._fp.close()


_STOP = False


def _on_signal(signum: int, _frame: Any) -> None:
    global _STOP
    _STOP = True


def percentile(values: list[int], pct: float) -> Optional[float]:
    if not values:
        return None
    s = sorted(values)
    idx = int(round((len(s) - 1) * pct))
    return float(s[idx])


def run(args: argparse.Namespace) -> int:
    import select

    chip = find_header_chip(args.chip)
    consumer = args.consumer
    req_fd = request_line(
        chip,
        args.line,
        consumer,
        event_buffer_size=args.event_buffer,
        bias_disabled=not args.keep_bias,
    )

    active_low = not args.active_high
    level = read_line_value(req_fd)
    initial_hs: Optional[int] = None
    if level is not None:
        initial_hs = (1 - level) if active_low else level

    out_dir = Path(args.output_dir)
    out_path = out_dir / args.filename
    writer = EdgeWriter(out_path)

    t_mono, t_real = mono_ns(), realtime_ns()
    writer.write(
        {
            "type": "hdr",
            "mono_ns": t_mono,
            "realtime_ns": t_real,
            "eg": EG_VERSION,
            "chip": chip.name,
            "chip_path": chip.path,
            "chip_label": chip.label,
            "chip_lines": chip.lines,
            "line": args.line,
            "consumer": consumer,
            "event_clock": "monotonic",
            "active_low": active_low,
            "initial_level": level,
            "initial_hs": initial_hs,
            "fixture_pos": args.fixture_pos,
            "signal": args.signal_name,
            "note": "kernel_stamped_edges_clock_monotonic_same_base_as_uartd",
            "path": str(out_path),
        }
    )
    writer.flush()

    print(f"hs_gpio_logger: {chip.path} ({chip.label}) line {args.line} -> {out_path}")
    print(f"  signal={args.signal_name} active_low={active_low} initial_hs={initial_hs}")
    print(f"  EG={EG_VERSION} event_clock=monotonic buffer={args.event_buffer}")
    print("  Ctrl-C to stop.")

    signal.signal(signal.SIGINT, _on_signal)
    signal.signal(signal.SIGTERM, _on_signal)

    poller = select.poll()
    poller.register(req_fd, select.POLLIN)

    edges = 0
    lost_total = 0
    gaps = 0
    expected_seqno: Optional[int] = None
    drain: list[int] = []
    last_status = mono_ns()
    status_every_ns = int(args.status_interval_s * 1e9)

    try:
        while not _STOP:
            if not poller.poll(500):
                now = mono_ns()
                if now - last_status >= status_every_ns:
                    _emit_status(writer, edges, gaps, lost_total, drain)
                    last_status = now
                continue
            try:
                raw = os.read(req_fd, SIZE_LINE_EVENT * 64)
            except OSError as exc:
                if exc.errno in (errno.EAGAIN, errno.EINTR):
                    continue
                raise
            recv = mono_ns()
            for off in range(0, len(raw) - SIZE_LINE_EVENT + 1, SIZE_LINE_EVENT):
                (ts_ns,) = struct.unpack_from("<Q", raw, off)
                ev_id, line_off, seqno, line_seqno = struct.unpack_from("<IIII", raw, off + 8)
                if expected_seqno is not None and seqno != expected_seqno:
                    lost = seqno - expected_seqno
                    if lost > 0:
                        gaps += 1
                        lost_total += lost
                        writer.write(
                            {
                                "type": "seq_gap",
                                "mono_ns": recv,
                                "expected_seqno": expected_seqno,
                                "got_seqno": seqno,
                                "lost": lost,
                                "note": "kernel_event_buffer_overflow_widths_across_gap_invalid",
                            }
                        )
                expected_seqno = seqno + 1
                rising = ev_id == EVENT_RISING_EDGE
                hs = (0 if rising else 1) if active_low else (1 if rising else 0)
                writer.write(
                    {
                        "type": "hs_edge",
                        "ts_ns": ts_ns,
                        "recv_ns": recv,
                        "hs": hs,
                        "edge": "rise" if rising else "fall",
                        "line": line_off,
                        "seqno": seqno,
                        "line_seqno": line_seqno,
                    }
                )
                edges += 1
                lag = recv - ts_ns
                if 0 <= lag < 5_000_000_000:
                    drain.append(lag)
                    if len(drain) > 20000:
                        del drain[:10000]
            now = mono_ns()
            if now - last_status >= status_every_ns:
                _emit_status(writer, edges, gaps, lost_total, drain)
                last_status = now
    finally:
        _emit_status(writer, edges, gaps, lost_total, drain, final=True)
        writer.close()
        os.close(req_fd)

    print(f"\nhs_gpio_logger: {edges} edges, {gaps} seq gaps ({lost_total} lost) -> {out_path}")
    if gaps:
        print("  WARNING: event buffer overflowed; widths spanning a gap are invalid.")
    return 0


def _emit_status(
    writer: EdgeWriter,
    edges: int,
    gaps: int,
    lost: int,
    drain: list[int],
    *,
    final: bool = False,
) -> None:
    p50 = percentile(drain, 0.50)
    p95 = percentile(drain, 0.95)
    rec: dict[str, Any] = {
        "type": "stat",
        "mono_ns": mono_ns(),
        "edges": edges,
        "seq_gaps": gaps,
        "lost": lost,
        "drain_p50_us": round(p50 / 1000.0, 2) if p50 is not None else None,
        "drain_p95_us": round(p95 / 1000.0, 2) if p95 is not None else None,
    }
    if final:
        rec["final"] = True
    writer.write(rec)
    writer.flush()
    if not final:
        d = f" drain_p50={rec['drain_p50_us']}us" if p50 is not None else ""
        print(f"  edges={edges} gaps={gaps} lost={lost}{d}", flush=True)


def self_test() -> int:
    """Validate uAPI struct sizes and ioctl encodings without touching hardware."""
    errors: list[str] = []

    if SIZE_LINE_ATTRIBUTE != 16:
        errors.append("gpio_v2_line_attribute must be 16 bytes")
    if SIZE_LINE_CONFIG_ATTRIBUTE != SIZE_LINE_ATTRIBUTE + 8:
        errors.append("gpio_v2_line_config_attribute must be attr + u64 mask")
    expect_cfg = 8 + 4 + 20 + SIZE_LINE_CONFIG_ATTRIBUTE * GPIO_V2_LINE_NUM_ATTRS_MAX
    if SIZE_LINE_CONFIG != expect_cfg:
        errors.append(f"gpio_v2_line_config {SIZE_LINE_CONFIG} != {expect_cfg}")
    expect_req = (
        GPIO_V2_LINES_MAX * 4 + GPIO_MAX_NAME_SIZE + SIZE_LINE_CONFIG + 4 + 4 + 20 + 4
    )
    if SIZE_LINE_REQUEST != expect_req:
        errors.append(f"gpio_v2_line_request {SIZE_LINE_REQUEST} != {expect_req}")
    if SIZE_LINE_REQUEST % 8 != 0:
        errors.append("gpio_v2_line_request must be 8-byte aligned")
    expect_ev = 8 + 4 * 4 + 4 * 6
    if SIZE_LINE_EVENT != expect_ev:
        errors.append(f"gpio_v2_line_event {SIZE_LINE_EVENT} != {expect_ev}")

    if GPIO_GET_CHIPINFO_IOCTL != _signed32(0x8044B401):
        errors.append(f"GPIO_GET_CHIPINFO_IOCTL={GPIO_GET_CHIPINFO_IOCTL:#x}")
    if GPIO_V2_GET_LINE_IOCTL != _signed32(0xC250B407):
        errors.append(f"GPIO_V2_GET_LINE_IOCTL={GPIO_V2_GET_LINE_IOCTL:#x}")
    if GPIO_V2_LINE_GET_VALUES_IOCTL != _signed32(0xC010B40E):
        errors.append(f"GPIO_V2_LINE_GET_VALUES_IOCTL={GPIO_V2_LINE_GET_VALUES_IOCTL:#x}")

    buf = build_line_request(9, "pog422-hs", event_buffer_size=1024, bias_disabled=True)
    if len(buf) != SIZE_LINE_REQUEST:
        errors.append("build_line_request produced wrong length")
    (line0,) = struct.unpack_from("<I", buf, 0)
    if line0 != 9:
        errors.append(f"offsets[0]={line0} != 9")
    cfg = GPIO_V2_LINES_MAX * 4 + GPIO_MAX_NAME_SIZE
    (flags,) = struct.unpack_from("<Q", buf, cfg)
    want = FLAG_INPUT | FLAG_EDGE_RISING | FLAG_EDGE_FALLING | FLAG_BIAS_DISABLED
    if flags != want:
        errors.append(f"config.flags={flags:#x} != {want:#x}")
    tail = cfg + SIZE_LINE_CONFIG
    (num_lines,) = struct.unpack_from("<I", buf, tail)
    (ebs,) = struct.unpack_from("<I", buf, tail + 4)
    if num_lines != 1 or ebs != 1024:
        errors.append(f"num_lines={num_lines} event_buffer_size={ebs}")
    consumer = buf[GPIO_V2_LINES_MAX * 4 : GPIO_V2_LINES_MAX * 4 + 9].decode()
    if consumer != "pog422-hs":
        errors.append(f"consumer={consumer!r}")

    for al, rising, want_hs in ((True, True, 0), (True, False, 1), (False, True, 1), (False, False, 0)):
        hs = (0 if rising else 1) if al else (1 if rising else 0)
        if hs != want_hs:
            errors.append(f"polarity active_low={al} rising={rising} -> {hs} != {want_hs}")

    if errors:
        for e in errors:
            print(f"FAIL: {e}")
        return 1
    print("hs_gpio_logger self-test OK")
    print(f"  gpio_v2_line_request = {SIZE_LINE_REQUEST} bytes")
    print(f"  gpio_v2_line_event   = {SIZE_LINE_EVENT} bytes")
    print(f"  GPIO_V2_GET_LINE_IOCTL = {GPIO_V2_GET_LINE_IOCTL & 0xFFFFFFFF:#010x}")
    return 0


def _signed32(value: int) -> int:
    return value - (1 << 32) if value >= 1 << 31 else value


def list_chips_cmd() -> int:
    chips = list_chips()
    if not chips:
        print("no /dev/gpiochip* found")
        return 1
    for c in chips:
        print(f"{c.path:20s} name={c.name:12s} label={c.label:20s} lines={c.lines}")
    try:
        hdr = find_header_chip(None)
        print(f"\nheader controller: {hdr.path} (label {hdr.label})")
    except SystemExit as exc:
        print(f"\n{exc}")
    return 0


def main(argv: Optional[list[str]] = None) -> int:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--line", type=int, default=9, help="BCM GPIO line offset (default 9 = J19 pin 7)")
    ap.add_argument("--chip", default=None, help="explicit /dev/gpiochipN (default: match pinctrl label)")
    ap.add_argument("--output-dir", default=os.environ.get("OUTPUT_DIR", "./data"))
    ap.add_argument("--filename", default="hs_gpio_machine.jsonl")
    ap.add_argument("--consumer", default="pog422-hs")
    ap.add_argument("--signal-name", default="SLV_HS")
    ap.add_argument("--fixture-pos", type=int, default=int(os.environ.get("FIXTURE_POS", "0") or 0))
    ap.add_argument(
        "--active-high",
        action="store_true",
        help="SLV_HS is active-low as wired (sinks LED6); use this only if you buffer/invert it",
    )
    ap.add_argument("--keep-bias", action="store_true", help="do not request BIAS_DISABLED")
    ap.add_argument("--event-buffer", type=int, default=1024)
    ap.add_argument("--status-interval-s", type=float, default=30.0)
    ap.add_argument("--list-chips", action="store_true")
    ap.add_argument("--self-test", action="store_true")
    args = ap.parse_args(argv)

    if args.self_test:
        return self_test()
    if args.list_chips:
        return list_chips_cmd()
    if not sys.platform.startswith("linux"):
        raise SystemExit("FATAL: GPIO capture requires Linux (use --self-test on the PC)")
    return run(args)


if __name__ == "__main__":
    raise SystemExit(main())