#!/usr/bin/env python3 """Phase 2 early win — TouchEasy (9309) + SuperBall Keno (930A) cash replay. Caches ``0x40`` picks + bet; on ``0x49`` counts hits vs draw balls and applies game-specific linear multipliers (``pay = bet_cents * mult``): - TouchEasy (9309): SGVI sheet 10 - SuperBall (930A): SGVI sheet 11 cash paytable (not BAJ). If the last (20th) draw ball is a hit on the player's picks, cash win is multiplied by 4. Policy: - **Local TID only** for cash fireworks (same as slots Phase 2). - If ``progwinval != 0``, skip — Phase 1 owns progressive celebration. Buy-A-Jackpot only affects when the *game* tags progressive; we do not evaluate BAJ tables here. """ from __future__ import annotations import os import sys import threading import time from dataclasses import dataclass from pathlib import Path from typing import Any, Callable, Optional _RUNTIME_ROOT = Path(__file__).resolve().parent if str(_RUNTIME_ROOT) not in sys.path: sys.path.insert(0, str(_RUNTIME_ROOT)) from decode_x9_result import ( KENO_PHASE2_GAME_IDS, KENO_RESULT_MSG, KENO_WAGER_MSG, SUPERBALL_KENO_GAME_ID, TOUCHEASY_KENO_GAME_ID, game_id_hex_from_frame, keno_eval_result_frame, keno_spot_picks_from_wager, keno_wager_bet_credits, tid_hex_from_frame, ) DEFAULT_THRESHOLD_USD = 4.00 DEFAULT_CREDIT_CENTS = 25 DEFAULT_DEDUPE_MS = 5000.0 DEFAULT_KENO_EVAL_GAMES = frozenset(KENO_PHASE2_GAME_IDS) def resolve_keno_threshold_cents() -> int: usd = os.environ.get("FIXTURE_FIREWORKS_WIN_THRESHOLD_USD", "").strip() if usd: try: return max(0, round(float(usd) * 100.0)) except ValueError: pass cents = os.environ.get("FIXTURE_EARLY_WIN_KENO_THRESHOLD_CENTS", "").strip() if cents: try: return max(0, int(cents)) except ValueError: pass return round(DEFAULT_THRESHOLD_USD * 100.0) def resolve_keno_credit_cents() -> int: raw = os.environ.get("ETHERGOLD_KENO_CREDIT_CENTS", "").strip() if not raw: raw = os.environ.get("ETHERGOLD_SLOT_CREDIT_CENTS", "").strip() if raw: try: return max(1, int(raw)) except ValueError: pass return DEFAULT_CREDIT_CENTS def keno_dedupe_key(raw: bytes) -> tuple[str, int, int, int] | None: if len(raw) < 8 or raw[0] != 0x05 or raw[-1] != 0x0D: return None tid = tid_hex_from_frame(raw) or "????" msg = raw[1] play_seq = raw[7] counter = raw[17] if len(raw) > 17 else -1 return (tid.upper(), msg, play_seq, counter) @dataclass class KenoHit: tid_hex: str game_id_hex: str | None picks: int hits: int win_cents: int pay_mult: int source: str dedupe_key: tuple[str, int, int, int] last_ball_hit: bool = False superball_x4: bool = False class Phase2KenoTrap: """Detect qualifying TouchEasy / SuperBall Keno cash wins; local fireworks.""" def __init__( self, *, threshold_cents: int = 400, credit_cents: int = DEFAULT_CREDIT_CENTS, dedupe_ms: float = DEFAULT_DEDUPE_MS, dry_run: bool = False, eval_games: frozenset[str] = DEFAULT_KENO_EVAL_GAMES, on_event: Callable[..., None] | None = None, ): self.threshold_cents = max(0, int(threshold_cents)) self.credit_cents = max(1, int(credit_cents)) self.dedupe_ms = dedupe_ms self.dry_run = dry_run self.eval_games = frozenset(g.upper() for g in eval_games) self._on_event = on_event self._lock = threading.Lock() self._seen: dict[tuple[str, int, int, int], float] = {} # tid -> (pick_spots, bet_credits) self._wagers: dict[str, tuple[tuple[int, ...], int]] = {} def _emit(self, ev: str, **fields: Any) -> None: if self._on_event is not None: self._on_event(ev, **fields) def _prune(self, now: float) -> None: cutoff = now - (self.dedupe_ms / 1000.0) self._seen = {k: ts for k, ts in self._seen.items() if ts >= cutoff} def inspect( self, frame: bytes, *, source: str, local_wire_tid: str | None = None, ) -> KenoHit | None: if len(frame) < 8 or frame[0] != 0x05 or frame[-1] != 0x0D: return None msg = frame[1] if msg == KENO_WAGER_MSG: self._remember_wager(frame) return None if msg != KENO_RESULT_MSG: return None return self._eval_result(frame, source=source, local_wire_tid=local_wire_tid) def _remember_wager(self, frame: bytes) -> None: tid = tid_hex_from_frame(frame) picks = keno_spot_picks_from_wager(frame) credits = keno_wager_bet_credits(frame) if tid and picks is not None and credits is not None: self._wagers[tid.upper()] = (picks, credits) self._emit( "early_win_keno_wager_store", tid=tid.upper(), picks=len(picks), credits=credits, game=game_id_hex_from_frame(frame), ) def _eval_result( self, frame: bytes, *, source: str, local_wire_tid: str | None, ) -> KenoHit | None: tid = (tid_hex_from_frame(frame) or "").upper() game = game_id_hex_from_frame(frame) if game and game.upper() not in self.eval_games: self._emit( "early_win_keno_skip", reason="game_unsupported", game=game, tid=tid, source=source, ) return None wager = self._wagers.get(tid) if wager is None: self._emit("early_win_keno_skip", reason="no_wager_cache", tid=tid, source=source) return None picks, bet_credits = wager res = keno_eval_result_frame( frame, pick_spots=picks, bet_credits=bet_credits, credit_cents=self.credit_cents, ) if res is None: self._emit("early_win_keno_skip", reason="decode_failed", tid=tid, source=source) return None if res.progwin_cents: self._emit( "early_win_keno_skip", reason="progwin_phase1", tid=tid, game=res.game_id_hex, progwin_cents=res.progwin_cents, source=source, ) return None qualifies = res.win_cents >= self.threshold_cents self._emit( "early_win_keno_eval", tid=tid, game=res.game_id_hex, picks=res.picks, hits=res.hits, pay_mult=res.pay_mult, base_pay_mult=res.base_pay_mult, win_cents=res.win_cents, base_win_cents=res.base_win_cents, win_dollars=round(res.win_cents / 100.0, 2), last_ball=res.last_ball, last_ball_hit=res.last_ball_hit, superball_x4=res.superball_x4, balls_source=res.balls_source, threshold_cents=self.threshold_cents, qualifies=qualifies, source=source, ) if not qualifies: return None local = (local_wire_tid or "").upper() if not local: self._emit("early_win_keno_skip", reason="local_tid_unknown", tid=tid, source=source) return None if tid != local: self._emit( "early_win_keno_skip", reason="not_local_terminal", tid=tid, local=local, source=source, ) return None key = keno_dedupe_key(frame) if key is None: return None now = time.monotonic() with self._lock: self._prune(now) if key in self._seen: self._emit("early_win_keno_dedupe", tid=tid, win_cents=res.win_cents, source=source) return None self._seen[key] = now hit = KenoHit( tid_hex=tid, game_id_hex=res.game_id_hex, picks=res.picks, hits=res.hits, win_cents=res.win_cents, pay_mult=res.pay_mult, source=source, dedupe_key=key, last_ball_hit=res.last_ball_hit, superball_x4=res.superball_x4, ) self._emit( "early_win_keno_hit", tid=tid, game=res.game_id_hex, picks=res.picks, hits=res.hits, win_cents=res.win_cents, win_dollars=round(res.win_cents / 100.0, 2), last_ball_hit=res.last_ball_hit, superball_x4=res.superball_x4, source=source, ) if self.dry_run: self._emit("early_win_keno_fireworks_dry_run", tid=tid, win_cents=res.win_cents) return hit dedupe_s = ":".join(str(x) for x in key) ipc_payload = self._trigger_local_fireworks(hit, dedupe_key=dedupe_s, local_wire_tid=local) if ipc_payload is not None: self._emit( "early_win_keno_fireworks_ipc", tid=tid, win_cents=res.win_cents, duration_sec=ipc_payload.get("duration_sec"), is_winner=ipc_payload.get("is_winner"), seq=ipc_payload.get("seq"), ) else: self._emit( "early_win_keno_fireworks_err", tid=tid, win_cents=res.win_cents, error="local IPC disabled", ) return hit def _trigger_local_fireworks( self, hit: KenoHit, *, dedupe_key: str, local_wire_tid: str | None, ) -> dict[str, Any] | None: from fireworks_ipc import local_fireworks_ipc_enabled, write_fireworks_request if not local_fireworks_ipc_enabled(): return None return write_fireworks_request( winner_tid=hit.tid_hex, progwin_cents=hit.win_cents, game_id_hex=hit.game_id_hex, dedupe_key=dedupe_key, local_wire_tid=local_wire_tid, is_progressive=False, ) def build_phase2_keno_trap_from_env( on_event: Callable[..., None] | None = None, *, default: str = "0", ) -> Phase2KenoTrap | None: if os.environ.get("FIXTURE_EARLY_WIN_PHASE2_KENO", default).strip().lower() not in ( "1", "true", "yes", ): return None dedupe_ms = float(os.environ.get("FIXTURE_EARLY_WIN_DEDUPE_MS", str(DEFAULT_DEDUPE_MS))) dry_run = os.environ.get("FIXTURE_EARLY_WIN_DRY_RUN", "0") in ("1", "true", "yes") return Phase2KenoTrap( threshold_cents=resolve_keno_threshold_cents(), credit_cents=resolve_keno_credit_cents(), dedupe_ms=dedupe_ms, dry_run=dry_run, on_event=on_event, ) def _selftest() -> list[str]: errors: list[str] = [] from decode_x9_result import parse_hex_line trap = Phase2KenoTrap(threshold_cents=400, credit_cents=25, dry_run=True) wager = parse_hex_line( "05,40,01,1D,AF,09,93,00,08,00,00,00,00,00,00,00,00,3D,90," "00,00,00,00,00,00,00,00,00,00,FF,03,00,00,00,00,00,00,00,00,61,40,0D" ) trap.inspect(wager, source="test", local_wire_tid="1DAF") # Build 5-hit result at native offset 21 (balls 1-5 + 20-34), progwin 0. result = bytearray(44) result[0], result[1], result[3], result[4] = 0x05, 0x49, 0x1D, 0xAF result[5], result[6], result[-1] = 0x09, 0x93, 0x0D balls = list(range(1, 6)) + list(range(20, 35)) result[21:41] = bytes(balls) hit = trap.inspect(bytes(result), source="test", local_wire_tid="1DAF") if hit is None or hit.win_cents != 1000 or hit.hits != 5: errors.append(f"expected 5-hit $10 dry hit, got {hit!r}") # SuperBall: 5 picks 76-80 @ $5, 2 hits incl last → $5 × 4 = $20. trap_sb = Phase2KenoTrap(threshold_cents=400, credit_cents=25, dry_run=True) w_sb = bytearray(42) w_sb[0], w_sb[1], w_sb[3], w_sb[4] = 0x05, 0x40, 0xD2, 0x49 w_sb[5], w_sb[6], w_sb[8], w_sb[-1] = 0x0A, 0x93, 20, 0x0D # mask last byte F8 → spots 76..80 (offset 38 is mask[9]) w_sb[38] = 0xF8 trap_sb.inspect(bytes(w_sb), source="test", local_wire_tid="D249") r_sb = bytearray(44) r_sb[0], r_sb[1], r_sb[3], r_sb[4] = 0x05, 0x49, 0xD2, 0x49 r_sb[5], r_sb[6], r_sb[-1] = 0x0A, 0x93, 0x0D r_sb[21:41] = bytes(list(range(1, 19)) + [76, 78]) hit_sb = trap_sb.inspect(bytes(r_sb), source="test", local_wire_tid="D249") if hit_sb is None or hit_sb.win_cents != 2000 or not hit_sb.superball_x4: errors.append(f"expected SuperBall $20 x4 dry hit, got {hit_sb!r}") # SuperBall progressive: cash trap must defer to phase 1. reasons: list[str] = [] def _on(ev: str, **fields: Any) -> None: if ev == "early_win_keno_skip": reasons.append(str(fields.get("reason"))) trap_pg = Phase2KenoTrap( threshold_cents=400, credit_cents=25, dry_run=True, on_event=_on ) trap_pg.inspect(bytes(w_sb), source="test", local_wire_tid="D249") r_pg = bytearray(r_sb) r_pg[13:17] = (400_060).to_bytes(4, "big") r_pg[21:41] = bytes([76, 77, 78, 79, 80] + list(range(1, 16))) if trap_pg.inspect(bytes(r_pg), source="test", local_wire_tid="D249") is not None: errors.append("superball progwin should not return cash hit") if "progwin_phase1" not in reasons: errors.append(f"expected progwin_phase1 skip, got {reasons}") _ = (TOUCHEASY_KENO_GAME_ID, SUPERBALL_KENO_GAME_ID) return errors if __name__ == "__main__": fails = _selftest() if fails: for msg in fails: print(msg, file=sys.stderr) raise SystemExit(1) print("early_win_phase2_keno selftest ok")