#!/usr/bin/env python3 """Compare Keno/game play visibility across fixture positions 1-4. Pos 1-2 (copper listen) should match packet_complete sets. Pos 3 (bridge on trunk) should match pos 1 on UART packet_complete. Pos 4 (ETH witness only when not playing locally) should eth_rx every trunk play. Usage: python3 compare_four_positions.py "../" python3 compare_four_positions.py ../sessions/my-archive --origin-tid 38D7 """ from __future__ import annotations import argparse import json from collections import Counter from pathlib import Path from fixture_pc_paths import resolve_events_jsonl WAGER_MSGS = frozenset({"30", "40", "60"}) OUTCOME_MSGS = frozenset({"39", "49", "69", "89"}) ACK_MSGS = frozenset({"31", "41", "61"}) PLAY_MSGS = WAGER_MSGS | OUTCOME_MSGS | ACK_MSGS # pos 4: frames observed on Ethernet (trunk forwarded by pos 3) POS4_WITNESS_EVENTS = frozenset({"eth_rx", "eth_tx", "eth_to_uart_done"}) def load_events(root: Path, pos: int) -> list[dict]: path = resolve_events_jsonl(root, pos) if path is None: return [] rows = [] for line in path.read_text(encoding="utf-8").splitlines(): if line.strip(): rows.append(json.loads(line)) return rows def play_key(row: dict) -> tuple | None: msg = row.get("msg") if not msg or msg not in PLAY_MSGS: return None tid = row.get("tid") if not tid: return None # game is often absent on eth_rx/eth_tx; match on msg+tid+len return (msg, tid, row.get("len")) def extract_play_keys(rows: list[dict], *, pos: int) -> list[tuple]: keys: list[tuple] = [] if pos in (1, 2): for row in rows: if row.get("ev") != "packet_complete": continue k = play_key(row) if k: keys.append(k) elif pos == 3: witness = frozenset({"packet_complete", "eth_rx", "eth_tx", "eth_to_uart_done"}) for row in rows: if row.get("ev") not in witness: continue k = play_key(row) if k: keys.append(k) else: for row in rows: if row.get("ev") not in POS4_WITNESS_EVENTS: continue k = play_key(row) if k: keys.append(k) return keys def filter_origin(keys: list[tuple], origin_tid: str | None) -> list[tuple]: if not origin_tid: return keys tid = origin_tid.upper() return [k for k in keys if k[1] == tid] def multiset(keys: list[tuple]) -> Counter: return Counter(keys) def format_key(k: tuple) -> str: msg, tid, ln = k return f"0x{msg} tid={tid} len={ln}" def compare_sets( ref: Counter, other: Counter, ref_label: str, other_label: str, ) -> list[str]: lines: list[str] = [] missing = ref - other extra = other - ref if not missing and not extra: lines.append(f" {other_label}: OK — matches {ref_label} ({sum(ref.values())} play frames)") return lines lines.append(f" {other_label}: MISMATCH vs {ref_label}") for k, n in sorted(missing.items()): lines.append(f" MISSING x{n}: {format_key(k)}") for k, n in sorted(extra.items()): lines.append(f" EXTRA x{n}: {format_key(k)}") return lines def summarize_wagers(keys: list[tuple]) -> Counter: return Counter(k for k in keys if k[0] in WAGER_MSGS) def main() -> int: parser = argparse.ArgumentParser( description="Compare game play frames across fixture positions 1-4" ) parser.add_argument( "root", type=Path, help="Fixture root or session archive (contains folders 1-4)", ) parser.add_argument( "--origin-tid", help="Only compare frames from this slave TID (e.g. 38D7 for pos 2 machine)", ) parser.add_argument( "--reference", type=int, choices=(1, 2), default=1, help="Copper reference position (default 1 = Master listen)", ) parser.add_argument("-o", "--output", type=Path, help="Write report to file") args = parser.parse_args() root = args.root.resolve() by_pos: dict[int, list[tuple]] = {} for pos in (1, 2, 3, 4): keys = extract_play_keys(load_events(root, pos), pos=pos) by_pos[pos] = filter_origin(keys, args.origin_tid) ref_pos = args.reference ref = multiset(by_pos[ref_pos]) lines = [ "=" * 72, "FIXTURE FOUR-POSITION PLAY COMPARE", f" Root: {root}", f" Reference: pos {ref_pos} packet_complete (copper trunk)", f" Pos 4 witness: eth_rx / eth_tx / eth_to_uart_done", ] if args.origin_tid: lines.append(f" Filter: origin TID {args.origin_tid.upper()}") lines.append("=" * 72) lines.append("") lines.append("## Wager counts (0x30/0x40/0x60) per position") for pos in (1, 2, 3, 4): w = summarize_wagers(by_pos[pos]) total = sum(w.values()) games = sorted( { r.get("game") for r in load_events(root, pos) if r.get("ev") == "packet_complete" and r.get("game") } ) lines.append(f" Pos {pos}: {total} wagers games={games or '-'}") lines.append("") lines.append("## Full play-frame multiset vs reference") for pos in (2, 3, 4): if pos == ref_pos: continue label = f"Pos {pos}" if pos == 4: label += " (ETH witness)" elif pos == 3: label += " (UART packet_complete on bridge node)" lines.extend(compare_sets(ref, multiset(by_pos[pos]), f"pos {ref_pos}", label)) lines.append("") lines.append("## Interpretation") lines.append(" Pos 1 vs 2: should be identical (both on copper trunk).") lines.append(" Pos 3 UART: should match pos 1 (same trunk; bridge also logs eth_tx).") lines.append(" Pos 4 ETH: should include every pos 1 play frame (via pos 3 bridge).") lines.append(" Any MISSING at pos 4 with pos 1/2/3 OK -> Ethernet path defect.") lines.append(" Pos 3 MISSING with pos 1/2 OK -> bridge UART/parse issue (not ETH to pos 4).") report = "\n".join(lines) + "\n" print(report, end="") if args.output: args.output.write_text(report, encoding="utf-8") print(f"Wrote {args.output}") return 0 if __name__ == "__main__": raise SystemExit(main())