#!/usr/bin/env python3 """Unix-socket client for pog422_uartd (C owns UART fd exclusively).""" from __future__ import annotations import errno import os import select import signal import socket import struct import subprocess import threading import time from dataclasses import dataclass from pathlib import Path from typing import Any, Callable, Optional POG422_IPC_MAGIC = 0x50342200 # UART/inject ABI (pog422_ipc.h). Not the aggregate EG_VERSION. POG422_EG_VERSION = "4.1.0.11" POG422_UARTD_VERSION = POG422_EG_VERSION POG422_INJECT_VERSION = POG422_EG_VERSION class UartdIpcDesyncError(ConnectionError): """IPC stream lost framing — client must reconnect (daemon may still be up).""" @dataclass class UartdRestartConfig: """How to respawn pog422_uartd after USB yank / daemon crash (EG 3.0.1.3).""" binary: str uart_port: str channel: str log_path: str min_restart_interval_s: float = 5.0 max_backoff_s: float = 60.0 port_wait_s: float = 20.0 socket_wait_s: float = 5.0 def resolve_uartd_binary() -> str: env = os.environ.get("POG422_UARTD_BIN", "").strip() if env and os.path.isfile(env) and os.access(env, os.X_OK): return env here = Path(__file__).resolve().parent for cand in ( here / "inject-shim" / "pog422_uartd", here.parent / "inject-shim" / "pog422_uartd", ): if cand.is_file() and os.access(cand, os.X_OK): return str(cand) return str(here / "inject-shim" / "pog422_uartd") def restart_config_from_env( *, sock_path: str, uart_port: str | None = None, channel: str | None = None, log_path: str | None = None, ) -> UartdRestartConfig | None: """Build restart config from hat_bridge exports + known socket/port.""" port = (uart_port or os.environ.get("POG422_UARTD_PORT", "")).strip() if not port: return None ch = (channel or os.environ.get("POG422_UARTD_CHANNEL", "") or "machine").strip() log = (log_path or os.environ.get("POG422_UARTD_LOG", "")).strip() if not log: log = f"/tmp/pog422_uartd_{os.getpid()}.log" binary = resolve_uartd_binary() if not os.path.isfile(binary): return None del sock_path # path lives on the client return UartdRestartConfig( binary=binary, uart_port=port, channel=ch, log_path=log, ) IPC_HS_ACTIVE = 1 IPC_HS_INACTIVE = 2 IPC_WINDOW = 3 IPC_DISCARD = 4 IPC_INJECT_DONE = 5 # EG 2.2.2.70: once bus evidence (echo / post_DE / forensics) proves the TX left # the driver, extend the inject wait once for the late DONE instead of failing a # delivered inject (pos5 2026-07-25 00:09 MRT: echo_len=42, DONE ~84 ms past the # 0.55 s gameplay budget). Worst-case gameplay pin: 0.55 + 1.0 s — still far # under the legacy 5 s hang ceiling this budget exists to prevent. INJECT_DONE_EXTEND_S = 1.0 IPC_SLV_IN = 6 IPC_GPIO7_ACK = 7 IPC_HAT_REGS = 8 IPC_HS_DIAG = 9 # optional — see pog422_uartd_hs_diag.h IPC_INJECT_FORENSICS = 10 IPC_UART_SUSPEND_ACK = 11 IPC_UART_RESUME_ACK = 12 IPC_INJECT_ECHO = 13 IPC_INJECT_COPPER = 14 IPC_VERSION = 15 IPC_POST_DE_RX = 16 IPC_HWDBG = 17 IPC_UART_PORT_LOST = 18 IPC_UART_PORT_REOPEN = 19 IPC_AUTO_CLOSE_DONE = 20 # EG 3.0.1.1 uartd hard-RT synth 0D IPC_X1_ACK_FRAME = 21 # EG 3.0.1.20 local x1 SOH..EOP before synth 0D # Capture events that must not be dropped while waiting for an RPC reply. _CAPTURE_IPC_TYPES = frozenset( { IPC_HS_ACTIVE, IPC_HS_INACTIVE, IPC_WINDOW, IPC_DISCARD, IPC_HS_DIAG, IPC_HWDBG, IPC_UART_PORT_LOST, IPC_UART_PORT_REOPEN, } ) # EG 3.0.0.9: inject evidence must not be dropped in GET_SLV_IN / other RPCs # (orphan DONE after timeout was silently discarded → false empty-echo MRT). _INJECT_IPC_TYPES = frozenset( { IPC_INJECT_DONE, IPC_INJECT_ECHO, IPC_INJECT_FORENSICS, IPC_INJECT_COPPER, IPC_POST_DE_RX, IPC_AUTO_CLOSE_DONE, IPC_X1_ACK_FRAME, } ) # Inject errnos that mean the UART device is gone — requeue, do not discard. UART_PORT_DEAD_ERRNOS = frozenset( {errno.ENODEV, errno.EIO, errno.ENXIO, errno.ENOENT, errno.EBADF} ) _UART_PORT_EVENT = struct.Struct("<128s128s") # old_path, new_path IPC_INJECT = 0x80 IPC_GET_SLV_IN = 0x81 IPC_SHUTDOWN = 0x82 IPC_GPIO7_ASSERT = 0x83 IPC_GPIO7_RELEASE = 0x84 IPC_GET_HAT_REGS = 0x85 IPC_UART_SUSPEND = 0x86 IPC_UART_RESUME = 0x87 IPC_GET_VERSION = 0x88 INJECT_ERR_HS_BUSY = errno.EBUSY _COPPER_PHASE_NAMES = { 0: "pre_de", 1: "de_on", 2: "pre_done", 3: "post_ioctl", 4: "pre_off", 5: "post_off", } _HDR = struct.Struct(" bool: return int(self.err) == 0 POG422_SLV_IN_UNKNOWN = 2 def _slv_in_state(raw: int) -> Optional[bool]: """Decode a tri-state SLV-IN sample (see POG422_SLV_IN_* in pog422_inject.h).""" value = int(raw) if value == POG422_SLV_IN_UNKNOWN: return None return bool(value) @dataclass class InjectResult: err: int tx_ns: int total_ns: int air_us: int slv_in_before: bool # DE-high samples are tri-state: None = no reading (GPIO1 event cache off). slv_in_after_on: Optional[bool] slv_in_after_body: Optional[bool] slv_in_after_drain: Optional[bool] slv_in_after_off: bool gpio1_edges: int gpio1_first_edge_us: int gpio1_last_edge_us: int forensics: Optional["InjectForensics"] = None echo_bytes: bytes = b"" post_de_rx_bytes: bytes = b"" x1_ack_frame: bytes = b"" # EG 3.0.1.20: local x1 before synth 0D copper: Optional["InjectCopper"] = None auto_close: Optional[AutoCloseResult] = None # EG 3.0.1.57 — inject stage forensics (0 if older uartd / short DONE) idle_wait_elapsed_us: int = 0 pre_bulk_us: int = 0 bulk_wall_us: int = 0 idle_wait_budget_ms: int = 0 require_idle: int = 0 idle_saw_busy: int = 0 # EG 3.0.1.58 — uartd worker wall + client IPC wait uartd_pre_burst_us: int = 0 uartd_burst_wall_us: int = 0 uartd_post_burst_us: int = 0 uartd_job_to_done_us: int = 0 inject_rpc_wait_us: int = 0 # Python: INJECT send → DONE received inject_rpc_scoop_us: int = 0 # Python: DONE → InjectResult return (luxury IPC) inject_capture_ipc_n: int = 0 # capture IPC drained while waiting for DONE # EG 3.0.2.4 — uartd x1 peel marks (µs; 0 if N/A / older uartd) peel_first_byte_us: int = 0 peel_hit_us: int = 0 peel_src: int = 0 # 0=none, 1=echo, 2=post_de, 3=hs (Pi HS Window2), 4=imm (UART-HS legacy) peel_x1_msg: int = 0 @dataclass class InjectCopperSample: t_us: int gpio_stat: int tx_fifo: int rx_fifo: int phase: int gpio7: bool gpio1: bool @property def phase_name(self) -> str: return _COPPER_PHASE_NAMES.get(self.phase, f"phase_{self.phase}") @dataclass class InjectCopper: gpio_mode: int uart_flow: int samples: list[InjectCopperSample] @dataclass class InjectForensics: inject_method: int wide_is_wire: bool driver_errno: int frame_len: int wide_len: int rx_fifo_before: int rx_fifo_after: int drain_wait_ns: int wide_bytes: bytes tx_fifo_before: int = 0 tx_fifo_after_bulk: int = 0 tx_fifo_after_drain: int = 0 usb_actual: int = 0 de_hold_us: int = 0 lsr: int = 0 # deprecated alias; use tx_empty_reason tx_empty_reason: int = 0 de_release_reason: int = 0 @property def de_release_reason_name(self) -> str: return { 0: "none", 1: "temt", 2: "timeout", 3: "error", 4: "deadline", }.get(int(self.de_release_reason), f"reason_{self.de_release_reason}") @dataclass class Gpio7Result: err: int slv_in: bool gpio7_tx: bool @dataclass class HatRegsResult: err: int gpio_mode: int uart_flow: int gpio_mode_ch0: int = 0 uart_flow_ch0: int = 0 @property def gpio_mode_rs485_auto(self) -> bool: return self.gpio_mode == 0x0339 @property def gpio_mode_manual_slv(self) -> bool: return self.gpio_mode == 0x0300 @property def gpio_mode_rs485_auto_ch0(self) -> bool: return self.gpio_mode_ch0 == 0x0339 class UartDaemonClient: """Talk to pog422_uartd over a Unix stream socket.""" def __init__(self, sock_path: str) -> None: self.sock_path = sock_path self._sock: Optional[socket.socket] = None self._lock = threading.Lock() self._mono_to_perf = time.perf_counter_ns() - time.monotonic_ns() self._slv_in_cache: bool = False # WINDOW/HS/DISCARD seen while inject()/RPC waiters hold the socket lock. # Emitting them into poll()/drain_events() after DONE fixes Class A drops. self._pending_capture: list[dict[str, Any]] = [] # Late INJECT_* after Python abandoned wait (or seen during GET_SLV_IN). self._pending_inject: list[tuple[int, bytes]] = [] # False after IPC_UART_PORT_LOST until IPC_UART_PORT_REOPEN. self._uart_attached: bool = True self.port_path: str = "" self.port_path_prev: str = "" # EG 3.0.1.3: auto-reinit after daemon/USB death (pos5 2026-07-28 CGS). self._restart_cfg: UartdRestartConfig | None = None self._restart_lock = threading.Lock() self._daemon_pid: int | None = None self._daemon_proc: subprocess.Popen[Any] | None = None self._last_restart_mono: float = 0.0 self._restart_failures: int = 0 self._last_ensure_detail: str = "" raw_pid = os.environ.get("POG422_UARTD_PID", "").strip() if raw_pid.isdigit(): self._daemon_pid = int(raw_pid) @property def uart_attached(self) -> bool: """True when uartd reports an open UART fd (not mid USB re-enumerate).""" return bool(self._uart_attached) def connect(self, *, timeout_s: float = 5.0) -> None: deadline = time.monotonic() + timeout_s last_err: Optional[OSError] = None while time.monotonic() < deadline: try: sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.connect(self.sock_path) sock.settimeout(2.0) self._sock = sock return except OSError as exc: last_err = exc time.sleep(0.05) raise ConnectionError(f"uartd not reachable at {self.sock_path}: {last_err}") def _queue_inject_ipc(self, typ: int, payload: bytes) -> bool: """Side-queue inject IPC seen outside inject(). Caller holds ``_lock``.""" if typ not in _INJECT_IPC_TYPES: return False # EG 3.0.1.20: x1 ack must reach eth-forward — never bury in discarded orphans. if typ == IPC_X1_ACK_FRAME: self._pending_capture.append(self._decode(typ, payload)) return True # EG 3.0.2.2: AUTO_CLOSE_DONE arrives after inject() returns (async 0D). # Capture path fires on_auto_close_done — do not stash in inject orphans # that next inject() discards. if typ == IPC_AUTO_CLOSE_DONE: self._pending_capture.append(self._decode(typ, payload)) return True self._pending_inject.append((int(typ), bytes(payload))) return True def _discard_pending_inject(self) -> None: """Drop stale orphan inject IPC. Caller holds ``_lock``.""" self._pending_inject.clear() def abort_inject_de(self) -> bool: """Best-effort GPIO7 release after inject timeout (uartd worker can act). EG 3.0.1.2: retry RELEASE — a single ACK miss left aborted_de=0 while DE may already be down (ipc_mu deadlock on DONE write). """ last_ok = False for _ in range(3): try: res = self.gpio7_release() if int(res.err) == 0: last_ok = True if not res.gpio7_tx: return True except (TimeoutError, OSError, ConnectionError, RuntimeError): pass time.sleep(0.015) return last_ok def drain_orphan_inject_ipc(self, *, wait_s: float = 0.15) -> int: """Read late INJECT_* / capture after timeout. Returns messages drained.""" with self._lock: if self._sock is None: return 0 n = 0 deadline = time.monotonic() + max(0.0, float(wait_s)) while time.monotonic() < deadline: remaining = deadline - time.monotonic() if remaining <= 0: break self._sock.settimeout(min(0.05, max(0.01, remaining))) try: magic, typ, plen = self._read_hdr() except (socket.timeout, TimeoutError): break except (OSError, ConnectionError): break if magic != POG422_IPC_MAGIC: break payload = self._read(plen) if plen else b"" n += 1 if self._queue_inject_ipc(typ, payload): continue if self._queue_capture_ipc(typ, payload): continue if self._sock is not None: self._sock.settimeout(2.0) # Orphans are stale relative to the abandoned inject — discard. self._discard_pending_inject() return n def get_versions(self) -> dict[str, str]: """Query live daemon + inject-shim version strings.""" with self._lock: if self._sock is None: raise RuntimeError("uartd not connected") self._send(IPC_GET_VERSION, b"") deadline = time.monotonic() + 2.0 while time.monotonic() < deadline: magic, typ, plen = self._read_hdr() if magic != POG422_IPC_MAGIC: raise RuntimeError(f"bad ipc magic {magic:#x}") payload = self._read(plen) if plen else b"" if typ == IPC_VERSION: if plen < _VERSION.size: raise RuntimeError(f"version reply too short plen={plen}") hw_debug = 0 if plen >= _VERSION_HWDBG.size: uartd_b, inject_b, eg_b, hw_debug = _VERSION_HWDBG.unpack( payload[: _VERSION_HWDBG.size] ) eg = eg_b.split(b"\0", 1)[0].decode("ascii", "replace") elif plen >= _VERSION_EG.size: uartd_b, inject_b, eg_b = _VERSION_EG.unpack( payload[: _VERSION_EG.size] ) eg = eg_b.split(b"\0", 1)[0].decode("ascii", "replace") else: uartd_b, inject_b = _VERSION.unpack(payload[: _VERSION.size]) eg = "" uartd = uartd_b.split(b"\0", 1)[0].decode("ascii", "replace") inject = inject_b.split(b"\0", 1)[0].decode("ascii", "replace") return { "uartd": uartd, "inject": inject, "eg": eg or uartd, "hw_debug": int(hw_debug), } # Ignore non-version chatter; queue capture so inject wait cannot drop it. self._queue_capture_ipc(typ, payload) continue raise RuntimeError("timeout waiting for uartd version") def close(self) -> None: with self._lock: if self._sock is not None: try: self._send(IPC_SHUTDOWN, b"") except OSError: pass try: self._sock.close() except OSError: pass self._sock = None def disconnect(self) -> None: """Drop the client socket without IPC_SHUTDOWN (daemon keeps running).""" with self._lock: if self._sock is not None: try: self._sock.close() except OSError: pass self._sock = None def reconnect(self, *, timeout_s: float = 5.0) -> None: """Re-open stream after desync / truncate — does not restart uartd.""" self.disconnect() self.connect(timeout_s=timeout_s) def configure_auto_restart(self, cfg: UartdRestartConfig | None) -> None: """Enable/disable daemon respawn after IPC death (field CGS guard).""" self._restart_cfg = cfg if cfg is not None and cfg.uart_port: self.port_path = cfg.uart_port def ipc_connected(self) -> bool: return self._sock is not None def _daemon_alive(self) -> bool: proc = self._daemon_proc if proc is not None: return proc.poll() is None pid = self._daemon_pid if pid is None or pid <= 0: return False try: os.kill(pid, 0) except OSError: return False # Zombie: exists but is not running — treat as dead. try: with open(f"/proc/{pid}/stat", encoding="utf-8") as fh: # pid (comm) state ... parts = fh.read().split() if len(parts) >= 3 and parts[2] == "Z": return False except OSError: return False return True def _kill_daemon(self) -> None: proc = self._daemon_proc if proc is not None and proc.poll() is None: try: proc.terminate() try: proc.wait(timeout=2.0) except subprocess.TimeoutExpired: proc.kill() proc.wait(timeout=1.0) except OSError: pass self._daemon_proc = None pid = self._daemon_pid if pid is not None and pid > 0: for sig in (signal.SIGTERM, signal.SIGKILL): try: os.kill(pid, sig) except OSError: break time.sleep(0.15) try: os.kill(pid, 0) except OSError: break self._daemon_pid = None # Reap any zombie we parented (exec'd shell → python owns uartd). try: os.waitpid(-1, os.WNOHANG) except ChildProcessError: pass def _wait_uart_port(self, port: str, wait_s: float) -> bool: deadline = time.monotonic() + max(0.5, wait_s) while time.monotonic() < deadline: if os.path.exists(port): return True time.sleep(0.1) return os.path.exists(port) def _spawn_daemon(self, cfg: UartdRestartConfig) -> None: try: os.makedirs(os.path.dirname(cfg.log_path) or ".", exist_ok=True) except OSError: pass try: if os.path.exists(self.sock_path): os.unlink(self.sock_path) except OSError: pass log_f = open(cfg.log_path, "a", encoding="utf-8", buffering=1) cmd = [ cfg.binary, "--port", cfg.uart_port, "--channel", cfg.channel, "--socket", self.sock_path, ] self._daemon_proc = subprocess.Popen( cmd, stdout=log_f, stderr=subprocess.STDOUT, close_fds=True, ) self._daemon_pid = int(self._daemon_proc.pid) deadline = time.monotonic() + max(0.5, cfg.socket_wait_s) while time.monotonic() < deadline: if self._daemon_proc.poll() is not None: raise ConnectionError( f"uartd exited during spawn rc={self._daemon_proc.returncode}" ) if os.path.exists(self.sock_path): return time.sleep(0.05) raise ConnectionError(f"uartd socket not ready: {self.sock_path}") def ensure_alive(self, *, reason: str = "") -> bool: """Reconnect IPC; if daemon is dead, respawn it. Returns True if usable. EG 3.0.1.3: pos5 2026-07-28 — XRUSB thrash killed uartd; bridge kept running and logged ``uartd not connected`` for ~30 min → CGS. Field seats must auto-heal without reboot. """ with self._restart_lock: if self._sock is not None and self._daemon_alive(): self._last_ensure_detail = "ok" return True # Soft path: daemon still up, only IPC desynced. if self._daemon_alive(): try: self.reconnect(timeout_s=2.0) self._restart_failures = 0 self._last_ensure_detail = f"reconnect:{reason or 'ipc'}" self._uart_attached = True return True except (OSError, ConnectionError, TimeoutError): pass cfg = self._restart_cfg if cfg is None: self._last_ensure_detail = "no_restart_cfg" return False now = time.monotonic() backoff = min( cfg.max_backoff_s, cfg.min_restart_interval_s * (2 ** min(self._restart_failures, 4)), ) if self._last_restart_mono and (now - self._last_restart_mono) < backoff: self._last_ensure_detail = f"backoff:{backoff:.1f}s" return False self._last_restart_mono = now self.disconnect() self._kill_daemon() if not self._wait_uart_port(cfg.uart_port, cfg.port_wait_s): self._restart_failures += 1 self._last_ensure_detail = f"no_port:{cfg.uart_port}" return False try: self._spawn_daemon(cfg) self.connect(timeout_s=cfg.socket_wait_s) except (OSError, ConnectionError, TimeoutError) as exc: self._restart_failures += 1 self._last_ensure_detail = f"spawn_fail:{exc}" return False self._restart_failures = 0 self._uart_attached = True self.port_path = cfg.uart_port self._last_ensure_detail = f"respawn:{reason or 'dead'}" return True def mono_to_perf_ns(self, mono_ns: int) -> int: return int(mono_ns) + self._mono_to_perf def _queue_capture_ipc(self, typ: int, payload: bytes) -> bool: """Side-queue capture IPC seen during RPC waits. Caller holds ``_lock``.""" if typ not in _CAPTURE_IPC_TYPES: return False self._pending_capture.append(self._decode(typ, payload)) return True def _take_pending_capture(self) -> list[dict[str, Any]]: """Pop queued capture events. Caller holds ``_lock``.""" if not self._pending_capture: return [] out = self._pending_capture self._pending_capture = [] return out def slv_in_cached(self) -> bool: """Last known SLV-IN — no IPC. Safe for hot-path diagnostics.""" return bool(self._slv_in_cache) def slv_in_active(self) -> bool: """Query SLV-IN via uartd. On IPC stall, return cache (never raise TimeoutError). Pos4 .43 soak: eth→UART called this for a log field while inject held the socket; ``TimeoutError`` killed the dual eth thread and froze the hop. EG 3.0.0.9: ≤50 ms budget (was 2 s) so a busy inject worker cannot pin the next hop; side-queue INJECT_* instead of dropping them. """ with self._lock: if self._sock is None: return bool(self._slv_in_cache) try: self._send(IPC_GET_SLV_IN, b"") deadline = time.monotonic() + 0.05 while time.monotonic() < deadline: remaining = deadline - time.monotonic() if remaining <= 0: break self._sock.settimeout(min(0.05, max(0.01, remaining))) try: magic, typ, plen = self._read_hdr() except (socket.timeout, TimeoutError): break if magic != POG422_IPC_MAGIC: raise RuntimeError(f"bad ipc magic {magic:#x}") payload = self._read(plen) if plen else b"" if typ == IPC_SLV_IN: if plen < 1: raise RuntimeError("slv_in reply empty") (active,) = struct.unpack(" HatRegsResult: """Read GPIO_MODE + UART_FLOW from uartd's open hat fd.""" with self._lock: if self._sock is None: raise RuntimeError("uartd not connected") self._send(IPC_GET_HAT_REGS, b"") deadline = time.monotonic() + 2.0 while time.monotonic() < deadline: magic, typ, plen = self._read_hdr() if magic != POG422_IPC_MAGIC: raise RuntimeError(f"bad ipc magic {magic:#x}") payload = self._read(plen) if plen else b"" if typ == IPC_HAT_REGS: if plen < _HAT_REGS.size: raise RuntimeError("hat_regs reply too short") err, gpio_mode, uart_flow, gpio_mode_ch0, uart_flow_ch0 = _HAT_REGS.unpack( payload[: _HAT_REGS.size] ) return HatRegsResult( err=int(err), gpio_mode=int(gpio_mode), uart_flow=int(uart_flow), gpio_mode_ch0=int(gpio_mode_ch0), uart_flow_ch0=int(uart_flow_ch0), ) if self._queue_capture_ipc(typ, payload): continue if self._queue_inject_ipc(typ, payload): continue raise RuntimeError("timeout waiting for hat_regs reply") def gpio7_assert(self) -> Gpio7Result: """Assert SLV-OUT (GPIO7) — driver 1N manual mode + uartd C path.""" return self._gpio7_op(IPC_GPIO7_ASSERT) def gpio7_release(self) -> Gpio7Result: """Release SLV-OUT (GPIO7) — driver 1N manual mode + uartd C path.""" return self._gpio7_op(IPC_GPIO7_RELEASE) def _gpio7_op(self, typ: int) -> Gpio7Result: with self._lock: if self._sock is None: raise RuntimeError("uartd not connected") self._send(typ, b"") deadline = time.monotonic() + 2.0 while time.monotonic() < deadline: magic, reply_typ, plen = self._read_hdr() if magic != POG422_IPC_MAGIC: raise RuntimeError(f"bad ipc magic {magic:#x}") payload = self._read(plen) if plen else b"" if reply_typ == IPC_GPIO7_ACK: if plen < _GPIO7_ACK.size: raise RuntimeError("gpio7_ack too short") err, slv_in, gpio7_tx = _GPIO7_ACK.unpack(payload[: _GPIO7_ACK.size]) self._slv_in_cache = bool(slv_in) return Gpio7Result(err=int(err), slv_in=bool(slv_in), gpio7_tx=bool(gpio7_tx)) if self._queue_capture_ipc(reply_typ, payload): continue if self._queue_inject_ipc(reply_typ, payload): continue raise RuntimeError("timeout waiting for gpio7_ack") def inject( self, frame: bytes, *, pre_tx_us: float, post_hold_us: float, flush_rx_before_tx: bool = False, require_idle: bool = True, idle_wait_ms: float = 80.0, on_bus_rx: Callable[[bytes, bytes], None] | None = None, on_x1_ack_frame: Callable[[bytes], None] | None = None, wait_s: float | None = None, auto_close: bool = False, auto_close_tid_tail: bytes | None = None, auto_close_local_tid: bytes | None = None, ) -> InjectResult: with self._lock: if self._sock is None: raise RuntimeError("uartd not connected") # Discard orphans from a prior abandoned wait before starting fresh. # (AUTO_CLOSE_DONE is capture-queued, not discarded here — EG 3.0.2.2.) self._discard_pending_inject() # = pre, post, flush, require_idle, idle_wait_ms, frame_len hdr = struct.pack( " None: nonlocal deadline, done_extend_armed if done_extend_armed: return done_extend_armed = True deadline = max(deadline, time.monotonic() + INJECT_DONE_EXTEND_S) def _side_queue_auto_close_done(raw: bytes) -> None: """Late / wrong-inject AC → capture so poll fires on_auto_close.""" self._pending_capture.append( self._decode(IPC_AUTO_CLOSE_DONE, raw) ) def _apply_inject_ipc(typ: int, raw: bytes) -> bool: """Handle inject luxury IPC. Returns True if consumed.""" nonlocal forensics, copper, echo_bytes, post_de_rx_bytes nonlocal x1_ack_frame, auto_close_res if typ == IPC_INJECT_FORENSICS: forensics = self._decode_inject_forensics(raw) _arm_done_extend() return True if typ == IPC_INJECT_COPPER: copper = self._decode_inject_copper(raw) return True if typ == IPC_INJECT_ECHO: echo_bytes = raw _arm_done_extend() # EG 3.0.1.1: do not call on_bus_rx mid-wait (backpressures C). return True if typ == IPC_POST_DE_RX: post_de_rx_bytes = raw _arm_done_extend() return True if typ == IPC_X1_ACK_FRAME: # EG 3.0.1.20: eth-forward during C HS-idle / 0D TX window. # EG 3.0.1.21: only merge wire-len ack (≥12 B) into post_de. x1_ack_frame = raw if raw and len(raw) >= 12 and raw[0] == 0x05 and raw[-1] == 0x0D: if not post_de_rx_bytes: post_de_rx_bytes = raw _arm_done_extend() if on_x1_ack_frame is not None and raw: try: on_x1_ack_frame(raw) except Exception: pass return True if typ == IPC_AUTO_CLOSE_DONE: # EG 4.0.0.63: empty peel no longer emits pre-DONE NO_X1. # Sync miss → pending (no AC on inject result); expire miss # and success AC arrive post-DONE via capture / on_auto_close. # Older uartd may still emit pre-DONE miss — bind if seen. if len(raw) < _AUTO_CLOSE_DONE.size: return True ( ac_err, x1_msg, counter, harvest, od_tx_ns, od_total_ns, ) = _AUTO_CLOSE_DONE.unpack(raw[: _AUTO_CLOSE_DONE.size]) ac = AutoCloseResult( err=int(ac_err), x1_msg=int(x1_msg), counter=int(counter), harvest=int(harvest), od_tx_ns=int(od_tx_ns), od_total_ns=int(od_total_ns), ) if auto_close and not inject_done_seen: # Pre-DONE AC for *this* job (legacy miss / rare race). auto_close_res = ac _arm_done_extend() elif auto_close and inject_done_seen and auto_close_res is None: # Same job, AC already in kernel buffer at DONE scoop. auto_close_res = ac else: # Wrong inject / late after Python returned earlier job. _side_queue_auto_close_done(raw) return True return False try: while True: remaining = deadline - time.monotonic() if remaining <= 0: break self._sock.settimeout(min(2.0, max(0.05, remaining))) try: magic, typ, plen = self._read_hdr() except (socket.timeout, TimeoutError) as exc: if "truncated" in str(exc): raise continue self._sock.settimeout(2.0) if magic != POG422_IPC_MAGIC: raise RuntimeError( f"bad ipc magic {magic:#x} waiting for inject_done" ) raw = self._read(plen) if plen else b"" if _apply_inject_ipc(typ, raw): continue if typ != IPC_INJECT_DONE: capture_ipc_n += 1 self._queue_capture_ipc(typ, raw) continue if plen < _INJECT_DONE.size: raise RuntimeError( f"inject_done too short type={typ} plen={plen} " f"need={_INJECT_DONE.size}" ) done_recv_mono = time.monotonic() inject_done_seen = True inject_rpc_wait_us = int( max(0.0, (done_recv_mono - wait_start) * 1_000_000.0) ) ( err, tx_ns, total_ns, air_us, slv_b, slv_on, slv_body, slv_drain, slv_off, gpio1_edges, _pad, gpio1_first_us, gpio1_last_us, ) = _INJECT_DONE.unpack(raw[: _INJECT_DONE.size]) idle_wait_elapsed_us = 0 pre_bulk_us = 0 bulk_wall_us = 0 idle_wait_budget_ms = 0 require_idle_flag = 0 idle_saw_busy = 0 uartd_pre_burst_us = 0 uartd_burst_wall_us = 0 uartd_post_burst_us = 0 uartd_job_to_done_us = 0 peel_first_byte_us = 0 peel_hit_us = 0 peel_src = 0 peel_x1_msg = 0 stage_off = _INJECT_DONE.size if plen >= stage_off + _INJECT_DONE_STAGE.size: ( idle_wait_elapsed_us, pre_bulk_us, bulk_wall_us, idle_wait_budget_ms, require_idle_flag, idle_saw_busy, ) = _INJECT_DONE_STAGE.unpack( raw[stage_off : stage_off + _INJECT_DONE_STAGE.size] ) stage_off += _INJECT_DONE_STAGE.size if plen >= stage_off + _INJECT_DONE_UARTD.size: ( uartd_pre_burst_us, uartd_burst_wall_us, uartd_post_burst_us, uartd_job_to_done_us, ) = _INJECT_DONE_UARTD.unpack( raw[stage_off : stage_off + _INJECT_DONE_UARTD.size] ) stage_off += _INJECT_DONE_UARTD.size if plen >= stage_off + _INJECT_DONE_PEEL.size: ( peel_first_byte_us, peel_hit_us, peel_src, peel_x1_msg, _peel_pad, ) = _INJECT_DONE_PEEL.unpack( raw[stage_off : stage_off + _INJECT_DONE_PEEL.size] ) # EG 3.0.2.1: after DONE, drain luxury/capture WITHOUT the # pre-.2.1 50–120 ms quiet-socket scoop (soak 0121: scoop p50 # ~83 ms under _tx_lock while copper inject already finished). # EG 3.0.2.2 / 4.0.0.63: do NOT block for AUTO_CLOSE_DONE. # Empty peel is pending (no miss AC pre-DONE); success/expire # AC arrive via capture / on_auto_close. def _scoop_one(block_s: float) -> bool: """Read one IPC frame. Return False if socket empty/timeout.""" nonlocal capture_ipc_n, forensics, copper, echo_bytes nonlocal post_de_rx_bytes, x1_ack_frame, auto_close_res if self._sock is None: return False self._sock.settimeout(block_s) try: magic2, typ2, plen2 = self._read_hdr() except (socket.timeout, TimeoutError, BlockingIOError): return False except (OSError, ConnectionError): return False if magic2 != POG422_IPC_MAGIC: return False raw2 = self._read(plen2) if plen2 else b"" if _apply_inject_ipc(typ2, raw2): return True capture_ipc_n += 1 self._queue_capture_ipc(typ2, raw2) return True # 1) Non-blocking drain of anything already in the kernel buffer # (uartd sends ECHO/FORENSICS/COPPER immediately after DONE). while _scoop_one(0.0): pass # 2) Brief luxury arm only if forensics never arrived (race vs # concurrent capture IPC) — never sit for a quiet 50 ms. if forensics is None: lux_deadline = time.monotonic() + 0.005 while forensics is None and time.monotonic() < lux_deadline: if not _scoop_one(0.001): continue inject_rpc_scoop_us = int( max(0.0, (time.monotonic() - done_recv_mono) * 1_000_000.0) ) # Optional post-return harvest callback (logging only). if on_bus_rx is not None and (echo_bytes or post_de_rx_bytes): try: on_bus_rx(echo_bytes, post_de_rx_bytes) except Exception: pass return InjectResult( err=int(err), tx_ns=int(tx_ns), total_ns=int(total_ns), air_us=int(air_us), slv_in_before=bool(slv_b), slv_in_after_on=_slv_in_state(slv_on), slv_in_after_body=_slv_in_state(slv_body), slv_in_after_drain=_slv_in_state(slv_drain), slv_in_after_off=bool(slv_off), gpio1_edges=int(gpio1_edges), gpio1_first_edge_us=int(gpio1_first_us), gpio1_last_edge_us=int(gpio1_last_us), forensics=forensics, echo_bytes=echo_bytes, post_de_rx_bytes=post_de_rx_bytes, x1_ack_frame=x1_ack_frame, copper=copper, auto_close=auto_close_res, idle_wait_elapsed_us=int(idle_wait_elapsed_us), pre_bulk_us=int(pre_bulk_us), bulk_wall_us=int(bulk_wall_us), idle_wait_budget_ms=int(idle_wait_budget_ms), require_idle=int(require_idle_flag), idle_saw_busy=int(idle_saw_busy), uartd_pre_burst_us=int(uartd_pre_burst_us), uartd_burst_wall_us=int(uartd_burst_wall_us), uartd_post_burst_us=int(uartd_post_burst_us), uartd_job_to_done_us=int(uartd_job_to_done_us), inject_rpc_wait_us=int(inject_rpc_wait_us), inject_rpc_scoop_us=int(inject_rpc_scoop_us), inject_capture_ipc_n=int(capture_ipc_n), peel_first_byte_us=int(peel_first_byte_us), peel_hit_us=int(peel_hit_us), peel_src=int(peel_src), peel_x1_msg=int(peel_x1_msg), ) finally: if self._sock is not None: self._sock.settimeout(2.0) elapsed = time.monotonic() - wait_start raise TimeoutError( f"uartd inject timed out after {elapsed:.2f}s " f"(budget={wait_budget:.2f}s done_extend=" f"{'armed' if done_extend_armed else 'no'} " f"echo_len={len(echo_bytes)} post_de_len={len(post_de_rx_bytes)})" ) def _uart_gate(self, req_type: int, ack_type: int) -> None: with self._lock: if self._sock is None: raise RuntimeError("uartd not connected") self._send(req_type, b"") deadline = time.monotonic() + 5.0 while time.monotonic() < deadline: magic, typ, plen = self._read_hdr() if magic != POG422_IPC_MAGIC: raise RuntimeError(f"bad ipc magic {magic:#x}") payload = self._read(plen) if plen else b"" if typ == ack_type: if plen < _GATE_ACK.size: raise RuntimeError(f"uart gate ack too short type={typ}") (err,) = _GATE_ACK.unpack(payload[: _GATE_ACK.size]) if err: raise OSError(err, os.strerror(err) if err > 0 else "uart gate failed") return if self._queue_capture_ipc(typ, payload): continue if typ in (IPC_UART_SUSPEND_ACK, IPC_UART_RESUME_ACK): continue raise RuntimeError("timeout waiting for uart gate ack") def suspend_uart(self) -> None: """Release UART fd so bridge can open pyserial for inject.""" self._uart_gate(IPC_UART_SUSPEND, IPC_UART_SUSPEND_ACK) def resume_uart(self) -> None: """Re-open UART fd after pyserial inject.""" self._uart_gate(IPC_UART_RESUME, IPC_UART_RESUME_ACK) def poll(self, *, block: bool = False) -> list[dict[str, Any]]: """Return capture events from daemon (non-blocking unless block=True).""" with self._lock: if self._sock is None: # Still flush any side-queued events from a prior inject wait. return self._take_pending_capture() out: list[dict[str, Any]] = self._take_pending_capture() if block: self._sock.settimeout(0.05) else: self._sock.settimeout(0.0) try: while True: try: magic, typ, plen = self._read_hdr() except BlockingIOError: break except socket.timeout: break except TimeoutError as exc: # Mid-header truncate — stream framing unknown. self._disconnect_locked() raise UartdIpcDesyncError(str(exc)) from exc if magic != POG422_IPC_MAGIC: self._disconnect_locked() raise UartdIpcDesyncError(f"bad ipc magic {magic:#x}") # Header committed: wait for payload (non-blocking empty # recv used to raise EAGAIN → listen retry → next read # treated payload as magic → RuntimeError crash). if plen: self._sock.settimeout(2.0) try: payload = self._read(plen) except (TimeoutError, ConnectionError, socket.timeout) as exc: self._disconnect_locked() raise UartdIpcDesyncError(str(exc)) from exc finally: if self._sock is not None: self._sock.settimeout(0.05 if block else 0.0) else: payload = b"" out.append(self._decode(typ, payload)) finally: if self._sock is not None: self._sock.settimeout(2.0) return out def _disconnect_locked(self) -> None: """Close socket; caller must already hold ``_lock``.""" if self._sock is None: return try: self._sock.close() except OSError: pass self._sock = None def drain_events(self) -> list[dict[str, Any]]: return self.poll(block=False) def _decode(self, typ: int, payload: bytes) -> dict[str, Any]: if typ == IPC_HS_ACTIVE: (mono,) = struct.unpack("= _AUTO_CLOSE_DONE.size: ( ac_err, x1_msg, counter, harvest, od_tx_ns, od_total_ns, ) = _AUTO_CLOSE_DONE.unpack(payload[: _AUTO_CLOSE_DONE.size]) return { "ev": "auto_close_done", "err": int(ac_err), "x1_msg": int(x1_msg), "counter": int(counter), "harvest": int(harvest), "od_tx_ns": int(od_tx_ns), "od_total_ns": int(od_total_ns), "ok": int(ac_err) == 0, } return {"ev": "auto_close_done", "err": -1, "ok": False} if typ == IPC_HS_DIAG: ( mono, phase, slv_in, cap_burst, _reserved, fifo_avail, hw_fifo_avail, cap_len, drain_bytes, probe_off, probe_len, probe, driver_rx_chars, ) = _HS_DIAG.unpack(payload[: _HS_DIAG.size]) probe_hex = ",".join(f"{b:02X}" for b in probe[:probe_len]) phase_name = ("hs_rise", "hs_fall_pre_drain", "hs_fall_post_drain")[phase] if phase < 3 else f"phase{phase}" return { "ev": "hs_fifo_diag", "mono_ns": mono, "phase": phase_name, "slv_in": bool(slv_in), "cap_burst_on_fall": bool(cap_burst), "fifo_avail": fifo_avail, "hw_fifo_avail": hw_fifo_avail, "cap_len": cap_len, "drain_bytes": drain_bytes, "driver_rx_chars": driver_rx_chars, "probe_off": probe_off, "probe_hex": probe_hex, } if typ == IPC_WINDOW: ( ws, fb, lb, wlen, ) = struct.unpack("= _UART_PORT_EVENT.size: old_b, new_b = _UART_PORT_EVENT.unpack(payload[: _UART_PORT_EVENT.size]) elif len(payload) >= 128: old_b = payload[:128] new_b = payload[128:256] if len(payload) >= 256 else b"" old_path = old_b.split(b"\0", 1)[0].decode("ascii", "replace") new_path = new_b.split(b"\0", 1)[0].decode("ascii", "replace") if typ == IPC_UART_PORT_LOST: self._uart_attached = False self.port_path_prev = old_path or self.port_path return { "ev": "uart_port_lost", "old_path": old_path, "new_path": "", } self._uart_attached = True self.port_path_prev = old_path or self.port_path_prev self.port_path = new_path or self.port_path return { "ev": "uart_port_reopen", "old_path": old_path, "new_path": new_path, } return {"ev": "unknown", "type": typ, "len": len(payload)} def _decode_inject_forensics(self, payload: bytes) -> InjectForensics: if len(payload) < _INJECT_FORENSICS_HDR_LEGACY.size: raise RuntimeError( f"inject_forensics too short plen={len(payload)} need={_INJECT_FORENSICS_HDR_LEGACY.size}" ) tx_fifo_before = 0 tx_fifo_after_bulk = 0 tx_fifo_after_drain = 0 usb_actual = 0 de_hold_us = 0 tx_empty_reason = 0 de_release_reason = 0 if len(payload) >= _INJECT_FORENSICS_HDR.size: ( inject_method, wide_is_wire, driver_errno, frame_len, wide_len, rx_fifo_before, rx_fifo_after, drain_wait_ns, tx_fifo_before, tx_fifo_after_bulk, tx_fifo_after_drain, usb_actual, ) = _INJECT_FORENSICS_HDR.unpack(payload[: _INJECT_FORENSICS_HDR.size]) hdr_size = _INJECT_FORENSICS_HDR.size rest = payload[hdr_size:] # EG 2.2: de_* fields sit between usb_actual and wide[] if len(rest) >= wide_len + _INJECT_FORENSICS_DE.size: de_hold_us, tx_empty_reason, de_release_reason = _INJECT_FORENSICS_DE.unpack( rest[: _INJECT_FORENSICS_DE.size] ) hdr_size += _INJECT_FORENSICS_DE.size else: ( inject_method, wide_is_wire, driver_errno, frame_len, wide_len, rx_fifo_before, rx_fifo_after, drain_wait_ns, ) = _INJECT_FORENSICS_HDR_LEGACY.unpack(payload[: _INJECT_FORENSICS_HDR_LEGACY.size]) hdr_size = _INJECT_FORENSICS_HDR_LEGACY.size wide_bytes = payload[hdr_size : hdr_size + wide_len] if len(wide_bytes) != wide_len: raise RuntimeError( f"inject_forensics wide truncated want={wide_len} got={len(wide_bytes)}" ) return InjectForensics( inject_method=int(inject_method), wide_is_wire=bool(wide_is_wire), driver_errno=int(driver_errno), frame_len=int(frame_len), wide_len=int(wide_len), rx_fifo_before=int(rx_fifo_before), rx_fifo_after=int(rx_fifo_after), drain_wait_ns=int(drain_wait_ns), wide_bytes=wide_bytes, tx_fifo_before=int(tx_fifo_before), tx_fifo_after_bulk=int(tx_fifo_after_bulk), tx_fifo_after_drain=int(tx_fifo_after_drain), usb_actual=int(usb_actual), de_hold_us=int(de_hold_us), lsr=int(tx_empty_reason), # legacy field name tx_empty_reason=int(tx_empty_reason), de_release_reason=int(de_release_reason), ) def _decode_inject_copper(self, payload: bytes) -> InjectCopper: if len(payload) < _INJECT_COPPER_HDR.size: raise RuntimeError( f"inject_copper too short plen={len(payload)} need={_INJECT_COPPER_HDR.size}" ) gpio_mode, uart_flow, n = _INJECT_COPPER_HDR.unpack(payload[: _INJECT_COPPER_HDR.size]) off = _INJECT_COPPER_HDR.size samples: list[InjectCopperSample] = [] for _ in range(int(n)): chunk = payload[off : off + _INJECT_COPPER_SAMPLE.size] if len(chunk) < _INJECT_COPPER_SAMPLE.size: raise RuntimeError("inject_copper sample truncated") t_us, gpio_stat, tx_fifo, rx_fifo, phase, gpio7, gpio1, _pad = ( _INJECT_COPPER_SAMPLE.unpack(chunk) ) samples.append( InjectCopperSample( t_us=int(t_us), gpio_stat=int(gpio_stat), tx_fifo=int(tx_fifo), rx_fifo=int(rx_fifo), phase=int(phase), gpio7=bool(gpio7), gpio1=bool(gpio1), ) ) off += _INJECT_COPPER_SAMPLE.size return InjectCopper( gpio_mode=int(gpio_mode), uart_flow=int(uart_flow), samples=samples, ) def _send(self, typ: int, payload: bytes) -> None: assert self._sock is not None hdr = _HDR.pack(POG422_IPC_MAGIC, typ, len(payload)) self._sock.sendall(hdr + payload) def _read_hdr(self) -> tuple[int, int, int]: assert self._sock is not None raw = self._recv_exact(12) return _HDR.unpack(raw) def _read(self, n: int) -> bytes: return self._recv_exact(n) def _recv_exact(self, n: int) -> bytes: """Read exactly *n* bytes. Non-blocking drain (``poll(block=False)`` / timeout=0) used to raise ``BlockingIOError`` mid-frame after the IPC header arrived but before the payload finished — that killed overnight listen (EAGAIN / errno 11). Empty-socket EAGAIN still propagates when *got==0*; partial frames wait. """ assert self._sock is not None chunks: list[bytes] = [] got = 0 deadline = time.monotonic() + 2.0 while got < n: try: part = self._sock.recv(n - got) except BlockingIOError: if got == 0: raise remain = max(0.0, deadline - time.monotonic()) if remain <= 0: raise TimeoutError( f"uartd ipc truncated got={got} need={n}" ) from None select.select([self._sock], [], [], remain) continue except socket.timeout: if got == 0: raise if time.monotonic() >= deadline: raise TimeoutError( f"uartd ipc truncated got={got} need={n}" ) from None continue if not part: raise ConnectionError("uartd closed socket") chunks.append(part) got += len(part) return b"".join(chunks) class DaemonHandshake: """SLV-IN via uartd (replaces HandshakeMonitor when C owns fd).""" def __init__(self, client: UartDaemonClient, channel) -> None: self.client = client self.channel = channel def is_active(self) -> bool: return self.client.slv_in_active() def is_active_cached(self) -> bool: return self.client.slv_in_cached() def default_socket_path(fixture_pos: int, channel: str | None = None) -> str: """Unix socket for pog422_uartd. CHANNEL=both uses per-channel sockets.""" override = os.environ.get("POG422_UARTD_SOCK", "").strip() if override and not channel: return override ch = (channel or "").strip().lower() if ch in ("machine", "ois"): env_key = f"POG422_UARTD_SOCK_{ch.upper()}" ch_override = os.environ.get(env_key, "").strip() if ch_override: return ch_override return f"/tmp/pog422_uart_{fixture_pos}_{ch}.sock" if override: return override return f"/tmp/pog422_uart_{fixture_pos}.sock" def env_enabled(default: bool = True) -> bool: raw = os.environ.get("POG422_UARTD", "").strip().lower() if raw in ("0", "false", "no", "off"): return False if raw in ("1", "true", "yes", "on"): return True return default def env_inject_enabled(*, default: bool = True) -> bool: """EG 2.1 default: uartd INJECT_TX. Python MARK/SPACE is unsupported.""" raw = os.environ.get("POG422_UARTD_INJECT", "").strip().lower() if raw in ("0", "false", "no", "off"): return False if raw in ("1", "true", "yes", "on"): return True return default