#!/usr/bin/env python3
"""POG422 listen-only — passive bus witness + progressive CSV for field collectors.

Default capture uses pog422_uartd (same C HS/RX path as bridge, including HW_DEBUG
→ hw_debug.jsonl). No Ethernet bridge / inject — witness only. Early-win fireworks
still run here (decode UART → fireworks-request.json). That is not bus drive.

Typical use:
  Position 2 (Master trunk) or 1 — FIXTURE_POS=N bash run_listen_jun11_good.sh
  Position 1 field collector — FIXTURE_PROD=1 bash run_listen_prod.sh

Outputs (debug):
  data/packets.raw, data/events.jsonl, data/fixture_stats.json, data/hw_debug.jsonl, data/wide_rx.bin
Outputs (prod / FIXTURE_PROD=1):
  logs/errors.jsonl only + progressives.csv

Usage:
  ./hat_fixture_listen.sh
  FIXTURE_POS=2 bash run_listen_jun11_good.sh
  POG422_UARTD=0 …  # legacy pyserial register-poll (not recommended)
"""

from __future__ import annotations

import argparse
import os
import sys
import time

import serial

from pathlib import Path

from current_game_ipc import (
    CurrentGamePublisher,
    clear_preboot_progressive_runtime,
    current_game_ipc_mode,
    map_wire_game_to_index,
    maybe_arm_listen_df_mismatch,
    resolve_current_game_path,
)
from early_win_listen import attach_early_win_listen_hook
from pog422_core import (
    EventLog,
    FIXTURE_PARSER_BUILD,
    FixtureIdentity,
    HandshakeCaptureState,
    ProgressiveCsvWriter,
    RawCaptureState,
    add_common_cli,
    channel_uses_raw_hs_window,
    clear_capture_files,
    handshake_capture_tick,
    is_gameplay_request,
    is_gameplay_response,
    is_progressive_broadcast,
    open_serial,
    probe_hat_session,
    raw_capture_tick,
    resolve_bridge_channels,
    resolve_channel,
    resolve_output_dir,
    resolve_serial_port,
)
from pog422_gpio_hs import hs_in_bcm, resolve_hs_path
from pog422_uart_capture import UartCaptureHandler, resolve_uartd_enabled

try:
    from pog422_uart_client import (
        POG422_UARTD_VERSION,
        UartDaemonClient,
        default_socket_path,
    )
except ImportError:
    POG422_UARTD_VERSION = ""  # type: ignore[misc, assignment]
    UartDaemonClient = None  # type: ignore[misc, assignment]
    default_socket_path = lambda pos, channel=None: f"/tmp/pog422_uart_{pos}.sock"  # type: ignore[misc, assignment]

DEFAULT_LOOP_SLEEP_US = 10.0


def listen_current_game_tid_ok(tid_hex: str | None) -> bool:
    """Optional per-cabinet filter. Empty env = last mapped gameplay on this tap."""
    want = os.environ.get("ETHERGOLD_CURRENT_GAME_TID", "").strip().upper()
    if not want:
        return True
    return (tid_hex or "").strip().upper() == want


def emit_current_game_result(event_log: EventLog, result) -> None:
    if result.action in ("off", "skip_unchanged"):
        return
    if result.action == "skip_unmapped":
        event_log.emit(
            "current_game_unmapped",
            game=result.game_id_hex,
            tid=result.tid_hex,
            mode=result.mode,
        )
        return
    if result.action == "would_publish":
        event_log.emit(
            "current_game_would_publish",
            game_index=result.game_index,
            game=result.game_id_hex,
            tid=result.tid_hex,
            mode=result.mode,
        )
        return
    if result.action == "published":
        event_log.emit(
            "current_game_published",
            game_index=result.game_index,
            game=result.game_id_hex,
            tid=result.tid_hex,
            mode=result.mode,
            force=bool(result.force),
            detail=result.detail,
        )
        return
    if result.action == "error":
        event_log.emit(
            "current_game_publish_error",
            game_index=result.game_index,
            game=result.game_id_hex,
            tid=result.tid_hex,
            detail=result.detail,
            force=bool(result.force),
        )


def observe_listen_current_game(publisher: CurrentGamePublisher, event_log: EventLog, meta):
    """Publish DyzineFrame game index from UART gameplay (listen has no local TID)."""
    if meta is None:
        return None
    if not (
        is_gameplay_request(meta.msg_type) or is_gameplay_response(meta.msg_type)
    ):
        return None
    if not getattr(meta, "game_id_hex", None):
        return None
    if not listen_current_game_tid_ok(getattr(meta, "tid_hex", None)):
        return None
    mapped = map_wire_game_to_index(getattr(meta, "game_id_hex", None))
    # x0 wager: if HDMI is not this game (logo, Promote restart, missing sidecar),
    # rewrite current-game.json even when the file already says SuperBall.
    if is_gameplay_request(meta.msg_type) and mapped is not None:
        maybe_arm_listen_df_mismatch(publisher, getattr(meta, "game_id_hex", None))
    result = publisher.observe(
        game_id_hex=meta.game_id_hex,
        tid_hex=getattr(meta, "tid_hex", None),
        source="listen_uart_gameplay",
    )
    emit_current_game_result(event_log, result)
    return result


def attach_current_game_listen_hook(event_log: EventLog) -> CurrentGamePublisher:
    publisher = CurrentGamePublisher()
    # Cold boot: DF may have started before listen. First mapped UART gameplay
    # must rewrite current-game.json even if the index matches the stale file.
    publisher.arm_force_next("listen_boot")

    def _on_packet_complete_current_game(
        meta,
        pkt: bytes,
        packets_in_window: int,
        *,
        uart_first_byte: int | None = None,
        window_us: float | None = None,
        byte_span_us: float | None = None,
    ) -> None:
        del pkt, packets_in_window, uart_first_byte, window_us, byte_span_us
        observe_listen_current_game(publisher, event_log, meta)

    event_log.add_packet_complete_hook(_on_packet_complete_current_game)
    event_log.emit(
        "current_game_ipc_enabled",
        mode=current_game_ipc_mode(),
        boot_arm_gap_fill=True,
        listen=True,
        path=resolve_current_game_path(),
        tid_filter=os.environ.get("ETHERGOLD_CURRENT_GAME_TID", "").strip() or None,
    )
    print(
        f"Current-game IPC: mode={current_game_ipc_mode()} "
        f"path={resolve_current_game_path()}",
        flush=True,
    )
    return publisher


def listen_loop_uartd(
    capture: UartCaptureHandler,
    event_log: EventLog,
    *,
    stats_interval_s: float,
    loop_sleep_us: float,
) -> None:
    from pog422_uart_client import UartdIpcDesyncError

    state = capture.state
    last_stats_at = time.time()
    sleep_s = max(loop_sleep_us, 1.0) / 1_000_000.0
    while True:
        try:
            capture.drain()

            if stats_interval_s > 0 and time.time() - last_stats_at >= stats_interval_s:
                event_log.write_stats(
                    {
                        "mode": "listen",
                        "capture_engine": "uartd",
                        "handshake_active": state.handshake_active,
                        "bytes_collected": state.bytes_collected,
                        "bytes_discarded_inactive": state.bytes_discarded_inactive,
                    }
                )
                last_stats_at = time.time()

        except KeyboardInterrupt:
            if state.buffer and state.handshake_active:
                event_log.finalize_handshake_window_for_channel(
                    capture.channel,
                    state.buffer,
                    window_start_perf_ns=state.window_start_perf_ns,
                    window_start_mono_ns=state.window_start_mono_ns,
                    first_byte_perf_ns=state.first_byte_perf_ns,
                    last_byte_perf_ns=state.last_byte_perf_ns,
                    hs_inactive_perf_ns=capture.last_hs_inactive_perf_ns,
                )
            event_log.write_stats(
                {
                    "mode": "listen",
                    "capture_engine": "uartd",
                    "stopped": True,
                    "bytes_collected": state.bytes_collected,
                    "bytes_discarded_inactive": state.bytes_discarded_inactive,
                }
            )
            raise
        except UartdIpcDesyncError as exc:
            # Framing lost — reconnect client socket; keep events.jsonl open.
            event_log.emit(
                "uartd_ipc_desync",
                error=type(exc).__name__,
                detail=str(exc)[:160],
            )
            try:
                capture.uartd.reconnect(timeout_s=5.0)
                event_log.emit("uartd_ipc_reconnect", ok=True)
            except Exception as reconnect_exc:
                event_log.emit(
                    "uartd_ipc_reconnect",
                    ok=False,
                    error=type(reconnect_exc).__name__,
                    detail=str(reconnect_exc)[:160],
                )
                time.sleep(max(sleep_s, 0.25))
            continue
        except (BlockingIOError, TimeoutError, ConnectionError, OSError) as exc:
            # Transient IPC / socket stalls must not kill overnight witnesses.
            event_log.emit(
                "uartd_ipc_retry",
                error=type(exc).__name__,
                detail=str(exc)[:160],
            )
            time.sleep(max(sleep_s, 0.05))
            continue

        time.sleep(sleep_s)


def listen_loop(
    ser: serial.Serial,
    port: str,
    channel,
    handshake,
    event_log: EventLog,
    state: HandshakeCaptureState,
    *,
    stats_interval_s: float,
    active_low: bool,
    loop_sleep_us: float,
) -> None:
    last_stats_at = time.time()
    sleep_s = max(loop_sleep_us, 1.0) / 1_000_000.0
    while True:
        try:
            handshake_capture_tick(ser, channel, handshake, event_log, state)

            if stats_interval_s > 0 and time.time() - last_stats_at >= stats_interval_s:
                event_log.write_stats(
                    {
                        "mode": "listen",
                        "capture_engine": "pyserial",
                        "channel": channel.key,
                        "serial_port": port,
                        "handshake_active_low": active_low,
                        "handshake_active": state.handshake_active,
                        "bytes_collected": state.bytes_collected,
                        "bytes_discarded_inactive": state.bytes_discarded_inactive,
                    }
                )
                last_stats_at = time.time()

        except KeyboardInterrupt:
            if state.buffer and state.handshake_active:
                event_log.finalize_handshake_window_for_channel(
                    channel,
                    state.buffer,
                    window_start_perf_ns=state.window_start_perf_ns,
                    window_start_mono_ns=state.window_start_mono_ns,
                    first_byte_perf_ns=state.first_byte_perf_ns,
                    last_byte_perf_ns=state.last_byte_perf_ns,
                    uart_first_byte=state.uart_first_byte,
                )
            event_log.write_stats(
                {
                    "mode": "listen",
                    "capture_engine": "pyserial",
                    "stopped": True,
                    "bytes_collected": state.bytes_collected,
                    "bytes_discarded_inactive": state.bytes_discarded_inactive,
                }
            )
            raise

        time.sleep(sleep_s)


def ignore_hs_loop(
    ser: serial.Serial,
    port: str,
    channel,
    event_log: EventLog,
    state: RawCaptureState,
    *,
    stats_interval_s: float,
    loop_sleep_us: float,
) -> None:
    """Continuous UART capture — no GPIOB1 / SLV-IN window gating (pyserial only)."""
    last_stats_at = time.time()
    sleep_s = max(loop_sleep_us, 1.0) / 1_000_000.0
    while True:
        try:
            raw_capture_tick(ser, event_log, state)

            if stats_interval_s > 0 and time.time() - last_stats_at >= stats_interval_s:
                event_log.write_stats(
                    {
                        "mode": "listen",
                        "capture": "ignore_hs",
                        "capture_engine": "pyserial",
                        "channel": channel.key,
                        "serial_port": port,
                        "bytes_collected": state.bytes_collected,
                        "bytes_skipped_preamble": state.bytes_skipped_preamble,
                        "stream_buf_bytes": len(state.stream),
                        "packets_emitted": state.packets_emitted,
                    }
                )
                last_stats_at = time.time()

        except KeyboardInterrupt:
            event_log.write_stats(
                {
                    "mode": "listen",
                    "capture": "ignore_hs",
                    "capture_engine": "pyserial",
                    "stopped": True,
                    "bytes_collected": state.bytes_collected,
                    "bytes_skipped_preamble": state.bytes_skipped_preamble,
                    "stream_buf_bytes": len(state.stream),
                    "packets_emitted": state.packets_emitted,
                }
            )
            raise

        time.sleep(sleep_s)


def listen_loop_uartd_dual(
    captures: list[UartCaptureHandler],
    event_log: EventLog,
    *,
    stats_interval_s: float,
    loop_sleep_us: float,
) -> None:
    """Poll multiple uartd sockets (CHANNEL=both machine + OIS)."""
    from pog422_uart_client import UartdIpcDesyncError

    last_stats_at = time.time()
    sleep_s = max(loop_sleep_us, 1.0) / 1_000_000.0
    while True:
        try:
            for capture in captures:
                capture.drain()
            if stats_interval_s > 0 and time.time() - last_stats_at >= stats_interval_s:
                for capture in captures:
                    state = capture.state
                    event_log.write_stats(
                        {
                            "mode": "listen",
                            "capture_engine": "uartd",
                            "channel": capture.channel.key,
                            "handshake_active": state.handshake_active,
                            "bytes_collected": state.bytes_collected,
                        }
                    )
                last_stats_at = time.time()
        except UartdIpcDesyncError as exc:
            event_log.emit("uartd_ipc_desync", detail=str(exc)[:160])
            for capture in captures:
                try:
                    capture.uartd.reconnect(timeout_s=5.0)
                except Exception:
                    pass
            time.sleep(max(sleep_s, 0.25))
            continue
        except (BlockingIOError, TimeoutError, ConnectionError, OSError) as exc:
            event_log.emit("uartd_ipc_retry", detail=str(exc)[:160])
            time.sleep(max(sleep_s, 0.05))
            continue
        time.sleep(sleep_s)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="POG422 listen-only — progressive CSV + UARTD/pyserial capture."
    )
    add_common_cli(parser)
    parser.add_argument(
        "--ignore-hs",
        action="store_true",
        help="Ignore SLV-IN handshake; continuous UART (forces pyserial)",
    )
    parser.add_argument(
        "--no-progressive-csv",
        action="store_true",
        help="Disable Master 0x0A → progressives.csv writer",
    )
    parser.add_argument(
        "--progressives-csv",
        default=None,
        help="progressives.csv path (default: $ETHERGOLD_PROGRESSIVES_CSV or $ETHERGOLD_ROOT/progressives.csv)",
    )
    parser.add_argument(
        "--loop-sleep-us",
        type=float,
        default=DEFAULT_LOOP_SLEEP_US,
        help="Main capture loop sleep (default 10 µs)",
    )
    uartd = parser.add_mutually_exclusive_group()
    uartd.add_argument(
        "--uartd",
        action="store_true",
        help="Use pog422_uartd (default; POG422_UARTD=1 or unset)",
    )
    uartd.add_argument(
        "--no-uartd",
        action="store_true",
        help="Legacy pyserial + register-poll capture",
    )
    return parser


def main() -> int:
    try:
        from cabinet_identity import apply_listen_tid_filter

        tid = apply_listen_tid_filter()
        if tid:
            print(f"Listen identity TID filter={tid} (GAMESTATS TERM_NUMBER)", flush=True)
    except Exception as exc:
        print(f"WARN: listen identity filter not applied: {exc}", flush=True)
    args = build_parser().parse_args()
    # Field default CHANNEL=both — dual uartd listen (machine + OIS status hooks).
    channel_key = (args.channel or "machine").strip().lower()
    if channel_key == "both":
        channels = resolve_bridge_channels("both")
    else:
        channels = [resolve_channel(channel_key)]
    output_dir = resolve_output_dir(args.output_dir)
    logs_dir = Path(args.logs_dir)
    active_low = not args.active_high
    prod_mode = bool(args.prod or os.environ.get("FIXTURE_PROD") == "1")
    use_uartd = resolve_uartd_enabled(args, default=True)
    hs_path = resolve_hs_path()

    if hs_path == "pi" and use_uartd and not args.ignore_hs:
        print(
            "NOTE: POG422_HS_PATH=pi — GPIO HS (BCM8/9) + XR UART RX; "
            "disabling UARTD (XR GPIO1 is disconnected on J3/J4 PI)",
            flush=True,
        )
        use_uartd = False

    if args.ignore_hs and use_uartd:
        print(
            "WARN: --ignore-hs requires pyserial; disabling UARTD for this run",
            flush=True,
        )
        use_uartd = False

    if len(channels) > 1 and not use_uartd:
        print(
            "FATAL: CHANNEL=both listen requires UARTD (two sockets)",
            file=sys.stderr,
            flush=True,
        )
        return 1

    if not args.append_capture:
        cleared = clear_capture_files(output_dir, logs_dir=logs_dir)
        if cleared:
            print(f"Cleared capture: {', '.join(cleared)}", flush=True)

    identity = FixtureIdentity(
        fixture_pos=args.fixture_pos,
        pi_label=args.pi_label or f"pos{args.fixture_pos}",
        role="listen",
    )
    event_log = EventLog(
        output_dir,
        identity,
        echo=args.echo_events,
        prod_mode=prod_mode,
        hs_witness=not args.no_hs_witness,
    )

    cleared = clear_preboot_progressive_runtime()
    if cleared:
        print(
            f"Cleared pre-boot progressive runtime: {', '.join(cleared)}",
            flush=True,
        )

    progressive_csv = None
    if (
        not args.no_progressive_csv
        and channel_key != "ois"
        and not args.ignore_hs
    ):
        progressive_csv = ProgressiveCsvWriter(
            args.progressives_csv,
            event_log=event_log,
        )

        def _on_packet_complete_progressive_csv(
            meta,
            pkt: bytes,
            packets_in_window: int,
            *,
            uart_first_byte: int | None = None,
            window_us: float | None = None,
            byte_span_us: float | None = None,
        ) -> None:
            del packets_in_window, uart_first_byte, window_us, byte_span_us
            if not is_progressive_broadcast(meta.msg_type):
                return
            progressive_csv.maybe_update(pkt, source="uart")

        event_log.add_packet_complete_hook(_on_packet_complete_progressive_csv)
        print(f"Progressive CSV: {progressive_csv.csv_path}", flush=True)

    if channel_key != "ois" and not args.ignore_hs:
        attach_current_game_listen_hook(event_log)
        ew = attach_early_win_listen_hook(event_log)
        print(
            "Early win listen: "
            f"phase1={int(ew['phase1'])} keno={int(ew['keno'])} "
            f"slots={int(ew['slots'])} local_tid={ew['local_tid'] or '-'}",
            flush=True,
        )

    uartd_clients: list = []
    captures: list[UartCaptureHandler] = []
    ser: serial.Serial | None = None
    handshake = None
    hs_state = HandshakeCaptureState()
    channel = channels[0]
    port = resolve_serial_port(channel, args.port)

    if use_uartd:
        if UartDaemonClient is None:
            print(
                "FATAL: POG422_UARTD=1 but pog422_uart_client.py is missing",
                file=sys.stderr,
                flush=True,
            )
            return 1
        for ch in channels:
            sock_path = default_socket_path(
                args.fixture_pos,
                ch.key if len(channels) > 1 else None,
            )
            client = UartDaemonClient(sock_path)
            try:
                client.connect()
            except ConnectionError as exc:
                print(f"FATAL: {exc}", file=sys.stderr, flush=True)
                print(
                    "Start uartd via hat_fixture_listen.sh / run_listen_jun11_good.sh "
                    "(CHANNEL=both starts machine + ois sockets).",
                    file=sys.stderr,
                    flush=True,
                )
                return 1
            uartd_clients.append(client)
            captures.append(UartCaptureHandler(client, ch, event_log, hs_state))
    else:
        try:
            ser, handshake = open_serial(
                port, channel, active_low=active_low, hs_source=hs_path
            )
        except serial.SerialException as exc:
            print(f"FATAL: cannot open {port}: {exc}", file=sys.stderr, flush=True)
            return 1
        except OSError as exc:
            print(f"FATAL: hat ioctl on {port}: {exc}", file=sys.stderr, flush=True)
            return 1

    if args.ignore_hs:
        capture_mode = "ignore_hs"
    elif any(channel_uses_raw_hs_window(ch) for ch in channels):
        capture_mode = "handshake_raw+machine" if len(channels) > 1 else "handshake_raw"
    else:
        capture_mode = "handshake"
    capture_engine = "uartd" if use_uartd else "pyserial"
    ch_label = "+".join(ch.key for ch in channels)

    hs_bcm = None
    if hs_path == "pi" and not args.ignore_hs:
        try:
            hs_bcm = hs_in_bcm(channel.key)
        except ValueError:
            hs_bcm = None
    print(
        f"Fixture listen pos={args.fixture_pos} channel={ch_label} "
        f"out={output_dir} capture={capture_mode} engine={capture_engine} "
        f"hs_path={hs_path}"
        + (f" hs_in=BCM{hs_bcm}" if hs_bcm is not None else "")
        + f" prod={prod_mode}",
        flush=True,
    )
    if use_uartd:
        for ch, client in zip(channels, uartd_clients):
            print(
                f"UART: pog422_uartd {POG422_UARTD_VERSION} "
                f"channel={ch.key} socket={default_socket_path(args.fixture_pos, ch.key if len(channels) > 1 else None)}",
                flush=True,
            )

    uartd_client = uartd_clients[0] if uartd_clients else None
    hat_probe = probe_hat_session(
        port,
        ser=ser,
        uartd=uartd_client,
        use_uartd=use_uartd,
        use_uartd_inject=False,
    )
    # hat_probe already carries uartd_version / inject_version / hw_debug / driver_*.
    # Do not pass those again — TypeError duplicate kwargs aborts before any capture.
    event_log.emit(
        "session_start",
        **hat_probe,
        mode="listen",
        capture=capture_mode,
        capture_engine=capture_engine,
        prod=prod_mode,
        parser_build=FIXTURE_PARSER_BUILD,
        port=port,
        channel=ch_label,
        gpio=",".join(ch.gpio_name for ch in channels),
        progressive_csv=progressive_csv.csv_path if progressive_csv else None,
        uartd=use_uartd,
        uartd_socket=(
            default_socket_path(args.fixture_pos, channels[0].key if len(channels) > 1 else None)
            if use_uartd
            else None
        ),
        uartd_inject=False,
        dual_channel=len(channels) > 1,
        hs_path=hs_path,
        hs_in_bcm=hs_bcm,
    )

    try:
        if captures:
            if len(captures) > 1:
                listen_loop_uartd_dual(
                    captures,
                    event_log,
                    stats_interval_s=args.stats_interval,
                    loop_sleep_us=args.loop_sleep_us,
                )
            else:
                listen_loop_uartd(
                    captures[0],
                    event_log,
                    stats_interval_s=args.stats_interval,
                    loop_sleep_us=args.loop_sleep_us,
                )
        elif args.ignore_hs:
            assert ser is not None
            raw_state = RawCaptureState()
            ignore_hs_loop(
                ser,
                port,
                channel,
                event_log,
                raw_state,
                stats_interval_s=args.stats_interval,
                loop_sleep_us=args.loop_sleep_us,
            )
        else:
            assert ser is not None and handshake is not None
            listen_loop(
                ser,
                port,
                channel,
                handshake,
                event_log,
                hs_state,
                stats_interval_s=args.stats_interval,
                active_low=active_low,
                loop_sleep_us=args.loop_sleep_us,
            )
    except KeyboardInterrupt:
        print("Listen stopped.", flush=True)
    finally:
        for client in uartd_clients:
            try:
                client.close()
            except Exception:
                pass
        if ser is not None:
            try:
                ser.close()
            except Exception:
                pass
        closer = getattr(handshake, "close", None)
        if callable(closer):
            try:
                closer()
            except Exception:
                pass
        event_log.close()
    return 0


if __name__ == "__main__":
    sys.exit(main())