#!/usr/bin/env python3 """POG422 Win Detector — scan fixture bridge captures for progressive jackpot hits. Captures come from the EtherGold bridge loop (Slave + Master + bridge inject path), not passive listen on Master copper alone — the bridge is the only intercept point. Joins events.jsonl packet_complete timestamps with packets.raw hex lines (line order must match — standard bridge --append-capture layout). Usage: python3 analyze_game_results.py data/ python3 analyze_game_results.py --packets-only data/packets.raw python3 analyze_game_results.py pos1/data pos2/data --json """ from __future__ import annotations import argparse import json import sys from dataclasses import asdict from pathlib import Path from typing import Any, Iterator, Optional from decode_x9_result import ( GameResultDecode, SKILL_JPOT_CLAIM_OPCODES, decode_frame, parse_hex_line, ) def resolve_data_paths(data_dir: Path) -> tuple[Path, Path]: """Return (events.jsonl, packets.raw) under a capture data directory.""" events = data_dir / "events.jsonl" packets = data_dir / "packets.raw" return events, packets def iter_packet_complete_events(events_path: Path) -> Iterator[dict[str, Any]]: with events_path.open(encoding="utf-8") as handle: for line in handle: line = line.strip() if not line: continue try: row = json.loads(line) except json.JSONDecodeError: continue if row.get("ev") == "packet_complete": yield row def load_packets_raw(path: Path) -> list[bytes]: frames: list[bytes] = [] with path.open(encoding="utf-8") as handle: for line in handle: raw = parse_hex_line(line) if raw: frames.append(raw) return frames def decode_to_record( dec: GameResultDecode, *, meta: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: prog = dec.progressive rec: dict[str, Any] = { "msg": f"{dec.msg_type:02X}", "category": dec.category, "len": dec.wire_len, "bank": dec.bank_number, "tid": dec.tid_hex, "game": dec.game_id_hex, "seq": dec.play_sequence, "is_progressive_hit": dec.is_progressive_hit, "notes": dec.notes, } if prog is not None: rec["active_wins"] = prog.active_wins rec["progdsp_be"] = prog.progdsp_be rec["progdsp_le"] = prog.progdsp_le rec["progwin_be"] = prog.progwin_be rec["progwin_le"] = prog.progwin_le rec["progwin_dollars"] = prog.progwin_dollars rec["progdsp_dollars"] = prog.progdsp_dollars if meta: rec["utc"] = meta.get("utc") rec["t_perf_ns"] = meta.get("t_perf_ns") rec["fixture_pos"] = meta.get("fixture_pos") rec["pi_label"] = meta.get("pi_label") if dec.is_progressive_hit or dec.category == "skill_jpot_claim": rec["payload_hex"] = dec.payload_hex return rec def analyze_capture_dir(data_dir: Path) -> dict[str, Any]: events_path, packets_path = resolve_data_paths(data_dir) frames = load_packets_raw(packets_path) if packets_path.is_file() else [] events = list(iter_packet_complete_events(events_path)) if events_path.is_file() else [] hits: list[dict[str, Any]] = [] claims: list[dict[str, Any]] = [] all_results: list[dict[str, Any]] = [] for idx, raw in enumerate(frames): dec = decode_frame(raw) meta = events[idx] if idx < len(events) else None if dec.category in {"gameplay_x9", "skill_jpot_grant", "jpot_strip"}: all_results.append(decode_to_record(dec, meta=meta)) if dec.is_progressive_hit: hits.append(decode_to_record(dec, meta=meta)) if dec.msg_type in SKILL_JPOT_CLAIM_OPCODES: claims.append(decode_to_record(dec, meta=meta)) return { "data_dir": str(data_dir), "packets": len(frames), "packet_complete_events": len(events), "progressive_hits": hits, "skill_jpot_claims": claims, "game_results_seen": len(all_results), } def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="POG422 Win Detector — scan captures for prog wins") parser.add_argument( "paths", nargs="+", help="Capture data/ directory(ies) or a packets.raw file with --packets-only", ) parser.add_argument( "--packets-only", action="store_true", help="Treat path(s) as packets.raw files (no events.jsonl join)", ) parser.add_argument("--json", action="store_true", help="Emit JSON summary") args = parser.parse_args(argv) summaries: list[dict[str, Any]] = [] if args.packets_only: for path_str in args.paths: path = Path(path_str) frames = load_packets_raw(path) hits = [] claims = [] for raw in frames: dec = decode_frame(raw) rec = decode_to_record(dec) if dec.is_progressive_hit: hits.append(rec) if dec.msg_type in SKILL_JPOT_CLAIM_OPCODES: claims.append(rec) summaries.append( { "packets_raw": str(path), "packets": len(frames), "progressive_hits": hits, "skill_jpot_claims": claims, } ) else: for path_str in args.paths: summaries.append(analyze_capture_dir(Path(path_str))) if args.json: print(json.dumps(summaries, indent=2)) else: for summary in summaries: label = summary.get("data_dir") or summary.get("packets_raw") print(f"\n=== {label} ===") print( f" packets={summary.get('packets', 0)} " f"hits={len(summary.get('progressive_hits', []))} " f"skill_claims={len(summary.get('skill_jpot_claims', []))}" ) for hit in summary.get("progressive_hits", []): pw = hit.get("progwin_be") pd = hit.get("progwin_dollars") amt = f"${pd:.2f} ({pw} cents BE)" if pd is not None and pw is not None else pw print( f" HIT msg={hit['msg']} tid={hit.get('tid')} game={hit.get('game')} " f"progwin={amt} utc={hit.get('utc', '—')}" ) for claim in summary.get("skill_jpot_claims", []): print( f" CLAIM msg={claim['msg']} tid={claim.get('tid')} game={claim.get('game')} " f"utc={claim.get('utc', '—')}" ) total_hits = sum(len(s.get("progressive_hits", [])) for s in summaries) return 0 if summaries else 1 if __name__ == "__main__": sys.exit(main())