"""Shared pog422_uartd capture — HS edges + burst windows for jun11-good bridge.""" from __future__ import annotations from collections.abc import Callable from typing import TYPE_CHECKING, Any if TYPE_CHECKING: from pog422_core import BusChannel, EventLog, HandshakeCaptureState from pog422_uart_client import UartDaemonClient HsInactiveFieldsFn = Callable[[int | None, int], dict[str, Any]] def hs_window_us(window_start_perf_ns: int | None, inactive_perf_ns: int) -> float | None: if window_start_perf_ns is None: return None return (inactive_perf_ns - window_start_perf_ns) / 1000.0 def default_hs_inactive_fields( window_start_perf_ns: int | None, inactive_perf_ns: int, ) -> dict[str, Any]: fields: dict[str, Any] = {} ws = hs_window_us(window_start_perf_ns, inactive_perf_ns) if ws is not None: fields["hs_window_us"] = ws return fields class UartCaptureHandler: """Apply pog422_uartd IPC capture events to HandshakeCaptureState + EventLog.""" def __init__( self, uartd: UartDaemonClient, channel: BusChannel, event_log: EventLog, state: HandshakeCaptureState, *, hs_inactive_fields: HsInactiveFieldsFn | None = None, on_auto_close_done: Callable[[dict[str, Any]], None] | None = None, on_x1_ack_frame: Callable[[bytes], None] | None = None, ) -> None: self.uartd = uartd self.channel = channel self.event_log = event_log self.state = state self._hs_inactive_fields = hs_inactive_fields or default_hs_inactive_fields self.on_auto_close_done = on_auto_close_done self.on_x1_ack_frame = on_x1_ack_frame self.last_hs_inactive_perf_ns: int | None = None self.last_hs_active_perf_ns: int | None = None self.last_uart_byte_perf_ns: int | None = None self.hs_rise_counter: int = 0 def drain(self) -> None: for ev in self.uartd.drain_events(): self.apply(ev) def apply(self, ev: dict[str, Any]) -> None: kind = ev.get("ev") if kind == "x1_ack_frame": # EG 3.0.1.20: late pending auto-close x1 (outside inject wait). data = ev.get("data") or b"" if data and self.on_x1_ack_frame is not None: try: self.on_x1_ack_frame(bytes(data)) except Exception: pass return if kind == "auto_close_done": if self.on_auto_close_done is not None: try: self.on_auto_close_done(ev) except Exception: pass return if kind == "hs_active": was_active = self.state.handshake_active mono = int(ev["mono_ns"]) perf = self.uartd.mono_to_perf_ns(mono) self.state.buffer.clear() self.state.window_start_perf_ns = perf self.state.window_start_mono_ns = mono self.state.first_byte_perf_ns = None self.state.last_byte_perf_ns = None self.state.handshake_active = True if not was_active: self.hs_rise_counter += 1 self.last_hs_active_perf_ns = perf self.event_log.emit( "hs_active", gpio=self.channel.gpio_name, silk=self.channel.legacy_silk, signal="SLV-HS", slv_in=True, ) return if kind == "hs_inactive": mono = int(ev["mono_ns"]) perf = self.uartd.mono_to_perf_ns(mono) self.state.handshake_active = False self.last_hs_inactive_perf_ns = perf hs_fields = self._hs_inactive_fields( self.state.window_start_perf_ns, perf, ) hs_fields.update( self.event_log.inject_overlap_fields( self.state.window_start_perf_ns, perf, ) ) self.event_log.emit( "hs_inactive", gpio=self.channel.gpio_name, silk=self.channel.legacy_silk, signal="SLV-HS", slv_in=False, window_bytes=len(self.state.buffer), **hs_fields, ) if not self.state.buffer: self.state.window_start_perf_ns = None self.state.window_start_mono_ns = None self.state.first_byte_perf_ns = None self.state.last_byte_perf_ns = None return if kind == "window": from pog422_core import ( SOH, _clear_post_x9_grace_state, filter_handoff_capture_to_x1_only, is_gameplay_ack, post_x9_grace_active, ) import time data = ev.get("data") or b"" ws_mono = int(ev["window_start_mono_ns"]) fb_mono = int(ev.get("first_byte_mono_ns") or 0) lb_mono = int(ev.get("last_byte_mono_ns") or 0) ws_perf = self.uartd.mono_to_perf_ns(ws_mono) fb_perf = self.uartd.mono_to_perf_ns(fb_mono) if fb_mono else None lb_perf = self.uartd.mono_to_perf_ns(lb_mono) if lb_mono else None now_ns = time.perf_counter_ns() grace_was_active = post_x9_grace_active(self.state, now_ns=now_ns) # Glue orphan post-DE SOH held by bridge when HS burst lacks leading MARK. if self.state.post_x9_grace_soh_hold: held = bytes(self.state.post_x9_grace_soh_hold) self.state.post_x9_grace_soh_hold.clear() if not data or data[0] != SOH: data = held + data self.event_log.emit( "post_de_soh_glued_window", held_bytes=len(held), window_bytes=len(data), ) if self.state.handoff_capture_x1_only: self.state.handoff_capture_x1_only = False filtered, detail = filter_handoff_capture_to_x1_only(bytearray(data)) if filtered is None: self.event_log.emit( "post_x9_handoff_capture_reject", window_bytes=len(data), reason=detail, first_byte=f"{data[0]:02X}" if data else None, ) self.state.bytes_discarded_inactive += len(data) _clear_post_x9_grace_state(self.state) self.state.buffer.clear() self.state.window_start_perf_ns = None self.state.window_start_mono_ns = None self.state.first_byte_perf_ns = None self.state.last_byte_perf_ns = None return if len(filtered) != len(data): self.event_log.emit( "post_x9_handoff_capture_filter", window_bytes=len(data), kept_bytes=len(filtered), reason=detail, ) data = bytes(filtered) if ( grace_was_active and len(data) >= 2 and data[0] == SOH and is_gameplay_ack(data[1]) ): self.event_log.emit( "post_x9_inject_grace_absorb", fifo_bytes=len(data), msg=f"{data[1]:02X}", engine="uartd", ) _clear_post_x9_grace_state(self.state) if data and fb_perf is not None: self.event_log.emit("uart_first_byte", byte=f"{data[0]:02X}") if lb_perf is not None: self.last_uart_byte_perf_ns = lb_perf self.state.bytes_collected += len(data) self.event_log.finalize_handshake_window_for_channel( self.channel, data, window_start_perf_ns=ws_perf, window_start_mono_ns=ws_mono, first_byte_perf_ns=fb_perf, last_byte_perf_ns=lb_perf, hs_inactive_perf_ns=self.last_hs_inactive_perf_ns, ) self.state.buffer.clear() self.state.window_start_perf_ns = None self.state.window_start_mono_ns = None self.state.first_byte_perf_ns = None self.state.last_byte_perf_ns = None return if kind == "bytes_discarded_inactive": count = int(ev.get("count") or 0) self.state.bytes_discarded_inactive += count self.event_log.emit( "bytes_discarded_inactive", count=count, first_byte=ev.get("first_byte", "00"), ) return if kind == "hs_fifo_diag": self.event_log.emit( "hs_fifo_diag", phase=ev.get("phase"), slv_in=ev.get("slv_in"), cap_burst_on_fall=ev.get("cap_burst_on_fall"), fifo_avail=ev.get("fifo_avail"), hw_fifo_avail=ev.get("hw_fifo_avail"), cap_len=ev.get("cap_len"), drain_bytes=ev.get("drain_bytes"), driver_rx_chars=ev.get("driver_rx_chars"), probe_off=ev.get("probe_off"), probe_hex=ev.get("probe_hex"), ) return if kind == "hw_debug": samples = ev.get("samples") or [] self.event_log.write_hw_debug( samples, n=int(ev.get("n") or 0), dropped=int(ev.get("dropped") or 0), ) return if kind == "uart_port_lost": self.event_log.emit( "uart_port_lost", channel=self.channel.key, old_path=ev.get("old_path") or "", ) return if kind == "uart_port_reopen": self.event_log.emit( "uart_port_reopen", channel=self.channel.key, old_path=ev.get("old_path") or "", new_path=ev.get("new_path") or "", ) return def resolve_uartd_enabled(args, *, default: bool = True) -> bool: """CLI/env: POG422_UARTD=1 enables (EG 2.1 default); --no-uartd disables.""" if getattr(args, "no_uartd", False): return False if getattr(args, "uartd", False): return True try: from pog422_uart_client import env_enabled as uartd_env_enabled except ImportError: return False return uartd_env_enabled(default=default) def resolve_uartd_inject_enabled(args, *, default: bool = True) -> bool: """CLI/env: POG422_UARTD_INJECT=1 → uartd INJECT_TX (EG 2.1 default).""" if getattr(args, "no_uartd_inject", False): return False if getattr(args, "uartd_inject", False): return True try: from pog422_uart_client import env_inject_enabled except ImportError: return default return env_inject_enabled(default=default)