#!/usr/bin/env python3 """Lab-only synthetic T340+ slave seat (gameplay state machine). Forged UART→ETH traffic for one lab TID, all catalog games: wager (x0) → wait Master outcome (x9) → forge ack (x1) → round done. No skill claim / jackpot path. 0D synth not required (inbound 0D still OK). Never enable with FIXTURE_PROD=1 unless LAB_GAMEPLAY_SYNTH_ALLOW_PROD=1. """ from __future__ import annotations import os import random import threading import time from dataclasses import dataclass, field from pathlib import Path from queue import Empty, SimpleQueue from typing import Any, Callable, Iterable from pog422_fixture_core import ( BINGO_CARD_BYTE_LEN, BINGO_WAGER_CARD_OFFSET, EOP, REQUEST_TO_RESPONSE, SOH, SUPERPICK_PLAY_MASK, SUPERPICK_WAGER_CARD_OFFSET, TOUCH6_LOT6_PLAY_MASK, TOUCH6_WAGER_CARD_OFFSET, is_gameplay_response, parse_packet_meta, pog422_frame_checksum_ok, pog422_set_frame_checksum, post_x9_local_grace_active, slv_in_active, ) # --- Lab TID registry (document in LAB_GAMEPLAY_SYNTH.md) --- # Preferred synthetic range; also allow real-style TIDs (e.g. 40CC) for Master-serve pilots. LAB_GAMEPLAY_TID_MIN = 0xE000 LAB_GAMEPLAY_TID_MAX = 0xE0FF LAB_GAMEPLAY_TID_PILOT = "E001" LAB_GAMEPLAY_TID_PILOT_ALT = "40CC" # Spot / pick mask layouts (decode_x9_result / core selftests) KENO_WAGER_SPOT_MASK_OFFSET = 29 KENO_WAGER_SPOT_MASK_LEN = 10 KENO_SPOT_COUNT = 10 _TEMPLATES_DIR = Path(__file__).resolve().parent / "lab_gameplay_templates" ForwardFn = Callable[[bytes], None] EmitFn = Callable[..., None] # Fallback if touch6_wager.hex missing (selftest / older trees). _TOUCH6_WAGER_BASE = bytes( int(x, 16) for x in ( # pos2 full-Dev autoplay — 2026-07-20 (TID 38D7, 1 card, byte8=0x04) "05,20,01,38,D7,07,91,06,04,00,00,00,00,00,00,00,00,2A,5E,04,00,00,00,00,00," "00,00,00,00,00,41,12,01,0F,3F,00,00,0C,00,46,49,0C,00,50,31,0D,FB,5B,0D" ).split(",") ) _SUPERPICK_WAGER_BASE = bytes( int(x, 16) for x in ( # Capture-style 0x60 — game id 9508 (08,95); synth rewrites TID+seq+CRC (+game if needed) "05,60,01,1D,AF,08,95,06,04,00,00,00,00,00,00,00,00,61,BC,00,00,00,00,00," "00,00,00,00,00,01,31,41,0E,C3,40,00,0D,9A,50,00,0C,13,05,08,0C,C1,6B,0D" ).split(",") ) # Bench 2026-07-04 — pos2 TID 38D7, game 9206, 4-card 80 B (capture-replicate). _BINGO_WAGER_80 = bytes( int(x, 16) for x in ( "05,30,01,38,D7,06,92,00,10,00,00,00,00,00,00,00,00,22,DC,0B,00,00,00,00,00," "00,00,00,00,00,54,0C,99,01,83,10,14,2C,AB,00,10,4D,14,2A,AA,00,82,32,44,45," "19,0C,21,23,A2,40,18,19,05,70,0E,41,D2,08,00,1D,4A,0A,2C,06,00,00,00,00,00," "00,00,68,B0,0D" ).split(",") ) # TouchEasy Keno — 10 spots 1–10, bet 8 (decode_x9_result selftest). _KENO_WAGER_BASE = bytes( int(x, 16) for x in ( "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" ).split(",") ) # Triple 7's 8-line slot (decode_x9_result wager80). _TRIPLE7S_WAGER_BASE = bytes( int(x, 16) for x in ( "05,80,01,38,D7,0C,97,05,10,00,00,00,00,00,00,00,00,7A,A6,0C," "00,00,00,00,00,00,00,00,00,69,CA,0D" ).split(",") ) # Minimal 32 B poker / spinjack scaffolds (TID + byte8 + checksum rewritten). _POKER_WAGER_BASE = bytes( int(x, 16) for x in ( "05,50,01,38,D7,00,94,01,01,00,00,00,00,00,00,00,00,00,00,00," "00,00,00,00,00,00,00,00,00,00,00,0D" ).split(",") ) _SPINJACK_WAGER_BASE = bytes( int(x, 16) for x in ( "05,70,01,38,D7,0F,96,01,02,00,00,00,00,00,00,00,00,00,00,00," "00,00,00,00,00,00,00,00,00,00,00,0D" ).split(",") ) @dataclass(frozen=True) class BetTierSpec: """One bet tier from Game_Bet_Tiers.xlsx (25¢ credits).""" credits: int n_units: int # cards / lines in play label: str @dataclass(frozen=True) class GameSpec: key: str game_id_hex: str msg_type: int per_card_min: int per_card_max: int tiers: tuple[BetTierSpec, BetTierSpec, BetTierSpec] # Source: Game_Bet_Tiers.xlsx (operator 2026-07-20); all values are 25¢ credits. GAME_SPECS: dict[str, GameSpec] = { "touch6": GameSpec( key="touch6", game_id_hex="9107", msg_type=0x20, per_card_min=4, per_card_max=4, tiers=( BetTierSpec(4, 1, "min"), BetTierSpec(8, 2, "medium"), BetTierSpec(16, 4, "high"), ), ), "superpick": GameSpec( key="superpick", game_id_hex="9508", # floor bench POG_SUPR (was 9507 legacy) msg_type=0x60, per_card_min=1, per_card_max=10, tiers=( BetTierSpec(1, 1, "min"), BetTierSpec(8, 2, "medium"), BetTierSpec(40, 4, "high"), ), ), "bingo": GameSpec( key="bingo", game_id_hex="9206", msg_type=0x30, per_card_min=4, per_card_max=4, tiers=( BetTierSpec(4, 1, "min"), BetTierSpec(8, 2, "medium"), BetTierSpec(16, 4, "high"), ), ), "poker": GameSpec( key="poker", game_id_hex="9400", msg_type=0x50, per_card_min=1, per_card_max=40, tiers=( BetTierSpec(1, 1, "min"), BetTierSpec(8, 1, "medium"), BetTierSpec(20, 1, "high"), ), ), "keno": GameSpec( key="keno", game_id_hex="9309", msg_type=0x40, per_card_min=1, per_card_max=40, tiers=( BetTierSpec(1, 1, "min"), BetTierSpec(8, 1, "medium"), BetTierSpec(20, 1, "high"), ), ), "spinjack": GameSpec( key="spinjack", game_id_hex="960F", msg_type=0x70, per_card_min=2, per_card_max=50, tiers=( BetTierSpec(2, 1, "min"), BetTierSpec(10, 1, "medium"), BetTierSpec(50, 1, "high"), ), ), "triple7s": GameSpec( key="triple7s", game_id_hex="970C", msg_type=0x80, per_card_min=1, per_card_max=2, tiers=( BetTierSpec(1, 1, "min"), BetTierSpec(8, 8, "medium"), BetTierSpec(16, 8, "high"), ), ), } DEFAULT_GAMES = ("touch6", "superpick", "bingo", "poker") def load_hex_template(path: Path) -> bytes | None: """Load first hex CSV frame line from a .hex template file.""" if not path.is_file(): return None for line in path.read_text(encoding="utf-8").splitlines(): s = line.strip() if not s or s.startswith("#"): continue parts = [p.strip() for p in s.replace(" ", "").split(",") if p.strip()] try: frame = bytes(int(p, 16) for p in parts) except ValueError: continue if len(frame) >= 4 and frame[0] == SOH and frame[-1] == 0x0D: if pog422_frame_checksum_ok(frame): return frame return None def touch6_golden_frame() -> bytes: """Copper golden Touch6 0x20 (prefer lab_gameplay_templates/touch6_wager.hex).""" loaded = load_hex_template(_TEMPLATES_DIR / "touch6_wager.hex") return loaded if loaded is not None else _TOUCH6_WAGER_BASE @dataclass class LabGameplayConfig: enabled: bool = False tid_hex: str = "" games: tuple[str, ...] = DEFAULT_GAMES session_s: float = 300.0 period_s: float = 3.0 jitter_s: float = 0.5 phase_s: float = 0.0 shuffle: bool = False bet_rotate: bool = True reroll_picks: bool = False allow_prod: bool = False # Touch6: replicate copper golden — rewrite TID + seq + CRC only (default on). touch6_replicate: bool = True fixture_pos: int | None = None @dataclass class _PickState: """Stable picks after first draw (per process / re-arm).""" touch6_cards: list[tuple[int, ...]] = field(default_factory=list) superpick_cards: list[tuple[int, ...]] = field(default_factory=list) keno_spots: tuple[int, ...] | None = None def _env_truthy(name: str, default: str = "0") -> bool: return os.environ.get(name, default).strip().lower() in ("1", "true", "yes", "on") def _env_float(name: str, default: float) -> float: raw = os.environ.get(name, "").strip() if not raw: return default return float(raw) def _env_int(name: str, default: int) -> int: raw = os.environ.get(name, "").strip() if not raw: return default return int(raw, 0) def parse_tid_hex(raw: str | None) -> str | None: if not raw: return None s = raw.strip().upper().replace("0X", "") if len(s) != 4: return None try: int(s, 16) except ValueError: return None return s def tid_in_lab_range(tid_hex: str) -> bool: v = int(tid_hex, 16) return LAB_GAMEPLAY_TID_MIN <= v <= LAB_GAMEPLAY_TID_MAX def parse_games_list(raw: str | None) -> tuple[str, ...]: if not raw or not raw.strip(): return DEFAULT_GAMES keys: list[str] = [] for part in raw.split(","): k = part.strip().lower() if not k: continue if k == "slots": k = "triple7s" if k not in GAME_SPECS: raise ValueError(f"unknown gameplay game key: {k}") keys.append(k) if not keys: raise ValueError("empty LAB_GAMEPLAY_GAMES list") return tuple(keys) def config_from_env(*, fixture_pos: int | None = None) -> LabGameplayConfig: pos = fixture_pos if pos is None: pos_raw = os.environ.get("FIXTURE_POS", "").strip() pos = int(pos_raw) if pos_raw.isdigit() else None phase = _env_float("LAB_GAMEPLAY_PHASE_S", -1.0) if phase < 0: # Stagger by hop so N seats are not lockstep on session boundaries. phase = float((hash(pos or 0) % 97) * 0.31) if pos else 0.0 return LabGameplayConfig( enabled=_env_truthy("LAB_GAMEPLAY_SYNTH", "0"), tid_hex=parse_tid_hex(os.environ.get("LAB_GAMEPLAY_TID")) or "", games=parse_games_list(os.environ.get("LAB_GAMEPLAY_GAMES")), session_s=_env_float("LAB_GAMEPLAY_SESSION_S", 300.0), period_s=_env_float("LAB_GAMEPLAY_PERIOD_S", 3.0), jitter_s=_env_float("LAB_GAMEPLAY_JITTER_S", 0.5), phase_s=phase, shuffle=_env_truthy("LAB_GAMEPLAY_SHUFFLE", "0"), bet_rotate=_env_truthy("LAB_GAMEPLAY_BET_ROTATE", "1"), reroll_picks=_env_truthy("LAB_GAMEPLAY_REROLL_PICKS", "0"), allow_prod=_env_truthy("LAB_GAMEPLAY_SYNTH_ALLOW_PROD", "0"), touch6_replicate=_env_truthy("LAB_GAMEPLAY_TOUCH6_REPLICATE", "1"), fixture_pos=pos, ) def merge_cli_config( base: LabGameplayConfig, *, enabled: bool | None = None, tid: str | None = None, games: str | None = None, session_s: float | None = None, period_s: float | None = None, jitter_s: float | None = None, phase_s: float | None = None, shuffle: bool | None = None, allow_prod: bool | None = None, ) -> LabGameplayConfig: cfg = LabGameplayConfig( enabled=base.enabled if enabled is None else bool(enabled), tid_hex=base.tid_hex, games=base.games, session_s=base.session_s if session_s is None else float(session_s), period_s=base.period_s if period_s is None else float(period_s), jitter_s=base.jitter_s if jitter_s is None else float(jitter_s), phase_s=base.phase_s if phase_s is None else float(phase_s), shuffle=base.shuffle if shuffle is None else bool(shuffle), bet_rotate=base.bet_rotate, reroll_picks=base.reroll_picks, allow_prod=base.allow_prod if allow_prod is None else bool(allow_prod), touch6_replicate=base.touch6_replicate, fixture_pos=base.fixture_pos, ) if tid is not None: parsed = parse_tid_hex(tid) cfg.tid_hex = parsed or "" if games is not None: cfg.games = parse_games_list(games) return cfg def validate_config(cfg: LabGameplayConfig, *, prod_mode: bool) -> str | None: """Return fatal error string, or None if OK to start (or disabled).""" if not cfg.enabled: return None if prod_mode and not cfg.allow_prod: return "LAB_GAMEPLAY_SYNTH cannot run with --prod (set LAB_GAMEPLAY_SYNTH_ALLOW_PROD=1 to override)" if not cfg.tid_hex: return "LAB_GAMEPLAY_SYNTH requires LAB_GAMEPLAY_TID (4 hex)" # Prefer E000–E0FF or known pilot alts (40CC); any valid 4-hex is allowed. if not parse_tid_hex(cfg.tid_hex): return f"LAB_GAMEPLAY_TID invalid: {cfg.tid_hex!r}" if not cfg.games: return "LAB_GAMEPLAY_GAMES is empty" for g in cfg.games: if g not in GAME_SPECS: return f"unknown game in LAB_GAMEPLAY_GAMES: {g}" if cfg.session_s <= 0 or cfg.period_s <= 0: return "LAB_GAMEPLAY_SESSION_S and PERIOD_S must be > 0" return None def _set_tid(frame: bytearray, tid_hex: str) -> None: tid = int(tid_hex, 16) frame[3] = (tid >> 8) & 0xFF frame[4] = tid & 0xFF def _set_game_id(frame: bytearray, game_id_hex: str) -> None: gid = int(game_id_hex, 16) frame[5] = gid & 0xFF frame[6] = (gid >> 8) & 0xFF def _set_seq_and_byte8(frame: bytearray, seq: int, byte8: int) -> None: frame[7] = seq & 0xFF frame[8] = byte8 & 0xFF def _finalize(frame: bytearray) -> bytes: pog422_set_frame_checksum(frame) out = bytes(frame) if not pog422_frame_checksum_ok(out): raise RuntimeError("lab_gameplay_synth checksum failed after rewrite") return out def rewrite_tid_seq_crc(golden: bytes, *, tid_hex: str, play_sequence: int) -> bytes: """Capture-replicate: keep payload; only TID, play_sequence, and CRC change.""" frame = bytearray(golden) _set_tid(frame, tid_hex) frame[7] = play_sequence & 0xFF return _finalize(frame) def ack_msg_for_response(response_msg: int) -> int | None: """Map Master x9 opcode → slave x1 ack (0x29→0x21, 0x39→0x31, …).""" if response_msg not in REQUEST_TO_RESPONSE.values(): return None return (int(response_msg) & 0xF0) | 0x01 def build_gameplay_ack( *, ack_msg: int, tid_hex: str, game_id_hex: str, play_sequence: int, ) -> bytes: """Build 12-byte gameplay ack (all families share this wire length).""" frame = bytearray(12) frame[0] = SOH frame[1] = ack_msg & 0xFF frame[2] = 0x01 _set_tid(frame, tid_hex) _set_game_id(frame, game_id_hex) frame[7] = play_sequence & 0xFF frame[8] = 0x00 frame[11] = EOP return _finalize(frame) def _clamp_credits(credits: int, lo: int, hi: int) -> int: return max(lo, min(hi, int(credits))) def _touch6_mask(spots: Iterable[int]) -> int: mask = TOUCH6_LOT6_PLAY_MASK for spot in spots: s = int(spot) if 1 <= s <= 25: mask |= 1 << (s - 1) return mask def _superpick_mask(digits: Iterable[int]) -> int: mask = SUPERPICK_PLAY_MASK for i, d in enumerate(digits): if i >= 6: break mask |= (int(d) & 0xF) << (i * 4) return mask def _keno_spot_mask(spots: Iterable[int]) -> bytes: buf = bytearray(KENO_WAGER_SPOT_MASK_LEN) for spot in spots: s = int(spot) if not (1 <= s <= 80): continue bi = (s - 1) // 8 bit = (s - 1) % 8 buf[bi] |= 1 << bit return bytes(buf) def _ensure_picks(state: _PickState, *, reroll: bool) -> None: if reroll or not state.touch6_cards: state.touch6_cards = [ tuple(sorted(random.sample(range(1, 26), 6))) for _ in range(4) ] if reroll or not state.superpick_cards: state.superpick_cards = [ tuple(random.randint(0, 9) for _ in range(6)) for _ in range(4) ] if reroll or state.keno_spots is None: state.keno_spots = tuple(sorted(random.sample(range(1, 81), KENO_SPOT_COUNT))) def build_wager( game_key: str, *, tid_hex: str, tier_index: int, play_sequence: int, picks: _PickState, reroll_picks: bool = False, touch6_replicate: bool = True, ) -> tuple[bytes, dict[str, Any]]: """Build one slave wager frame for *game_key* at bet tier *tier_index* (0..2).""" spec = GAME_SPECS[game_key] tier = spec.tiers[tier_index % 3] _ensure_picks(picks, reroll=reroll_picks) # Sheet totals are already session credits (not always per_card×units). if game_key in ("touch6", "bingo", "superpick", "keno", "triple7s"): credits = int(tier.credits) else: credits = _clamp_credits(tier.credits, spec.per_card_min, spec.per_card_max) meta: dict[str, Any] = { "game": game_key, "game_id": spec.game_id_hex, "wager_msg": f"{spec.msg_type:02X}", "tier": tier_index % 3 + 1, "tier_label": tier.label, "credits": credits, "n_units": tier.n_units, "play_sequence": play_sequence & 0xFF, "tid": tid_hex.upper(), } if game_key == "touch6": golden = touch6_golden_frame() if touch6_replicate: # Copper capture-replicate: TID + seq + CRC only. frame = rewrite_tid_seq_crc( golden, tid_hex=tid_hex, play_sequence=play_sequence ) meta["tier_label"] = "replicate" meta["credits"] = int(frame[8]) meta["n_units"] = max(1, int(frame[8]) // 4) if frame[8] in (4, 8, 12, 16) else 4 meta["card_source"] = "touch6_wager.hex" meta["golden_tid"] = parse_packet_meta(golden).tid_hex return frame, meta frame = bytearray(golden if len(golden) == 48 else _TOUCH6_WAGER_BASE) _set_tid(frame, tid_hex) _set_game_id(frame, spec.game_id_hex) _set_seq_and_byte8(frame, play_sequence, credits) n = tier.n_units for i in range(4): off = TOUCH6_WAGER_CARD_OFFSET + i * 4 if i < n: mask = _touch6_mask(picks.touch6_cards[i]) frame[off : off + 4] = int(mask).to_bytes(4, "little") else: frame[off : off + 4] = b"\x00\x00\x00\x00" meta["picks"] = [list(picks.touch6_cards[i]) for i in range(n)] return _finalize(frame), meta if game_key == "superpick": # Capture-replicate: keep picks/byte8; rewrite TID + game id + seq + CRC. # Random pick rewrite previously yielded Master-silent 0x60 (no 0x69). frame = bytearray(_SUPERPICK_WAGER_BASE) _set_tid(frame, tid_hex) _set_game_id(frame, spec.game_id_hex) frame[7] = play_sequence & 0xFF meta["tier_label"] = "replicate" meta["credits"] = int(frame[8]) meta["n_units"] = 1 meta["card_source"] = "superpick_wager_base" return _finalize(frame), meta if game_key == "bingo": frame = bytearray(_BINGO_WAGER_80) _set_tid(frame, tid_hex) _set_game_id(frame, spec.game_id_hex) _set_seq_and_byte8(frame, play_sequence, credits) n = tier.n_units # Capture-replicate: keep first N card slots; zero the rest (OFF). for i in range(4): off = BINGO_WAGER_CARD_OFFSET + i * BINGO_CARD_BYTE_LEN if i >= n: frame[off : off + BINGO_CARD_BYTE_LEN] = bytes(BINGO_CARD_BYTE_LEN) meta["cards"] = n meta["card_source"] = "bingo_30_80" return _finalize(frame), meta if game_key == "keno": frame = bytearray(_KENO_WAGER_BASE) _set_tid(frame, tid_hex) _set_game_id(frame, spec.game_id_hex) _set_seq_and_byte8(frame, play_sequence, credits) assert picks.keno_spots is not None mask = _keno_spot_mask(picks.keno_spots) end = KENO_WAGER_SPOT_MASK_OFFSET + KENO_WAGER_SPOT_MASK_LEN frame[KENO_WAGER_SPOT_MASK_OFFSET:end] = mask meta["spots"] = list(picks.keno_spots) return _finalize(frame), meta if game_key == "poker": frame = bytearray(_POKER_WAGER_BASE) _set_tid(frame, tid_hex) _set_game_id(frame, spec.game_id_hex) _set_seq_and_byte8(frame, play_sequence, credits) return _finalize(frame), meta if game_key == "spinjack": frame = bytearray(_SPINJACK_WAGER_BASE) _set_tid(frame, tid_hex) _set_game_id(frame, spec.game_id_hex) _set_seq_and_byte8(frame, play_sequence, credits) return _finalize(frame), meta if game_key == "triple7s": frame = bytearray(_TRIPLE7S_WAGER_BASE) _set_tid(frame, tid_hex) _set_game_id(frame, spec.game_id_hex) _set_seq_and_byte8(frame, play_sequence, credits) return _finalize(frame), meta raise ValueError(f"no builder for game {game_key}") def busy_reason(bridge: Any, tid_hex: str) -> str | None: """Skip wager tick when bus / local round is busy.""" try: if getattr(bridge, "_tx_lock", None) is not None and bridge._tx_lock.locked(): return "tx_lock" except Exception: pass try: if slv_in_active(bridge.handshake): return "hs_busy" except Exception: pass try: if callable(getattr(bridge, "_local_close_pending", None)) and bridge._local_close_pending(): return "await_local_0d" except Exception: pass try: if callable(getattr(bridge, "_trunk_round_active_for_tid", None)): if bridge._trunk_round_active_for_tid(tid_hex): return "trunk_round" except Exception: pass try: if post_x9_local_grace_active(bridge.state): return "post_x9_grace" except Exception: pass return None def arm_local_slave_tid(bridge: Any, tid_hex: str, emit: EmitFn) -> str | None: """Pre-set local seat TID. Refuse if a different real seat was already learned.""" tid = tid_hex.upper() existing = getattr(bridge, "_local_slave_tid", None) if existing is not None and str(existing).upper() != tid: return f"LAB_GAMEPLAY_TID {tid} collides with learned local_slave_tid {existing}" if existing is None: bridge._local_slave_tid = tid emit( "local_slave_tid_learned", tid=tid, source="lab_gameplay_synth", msg=None, ) return None class LabGameplaySynth: """Background seat state machine for one synthetic TID (all catalog games). States (per round): IDLE → forge wager (x0) → WAIT_X9 WAIT_X9 → on Master outcome (x9) forge ack (x1) → IDLE (round closed) No skill claim / jackpot processing. 0D synth is not required; inbound Master 0D is still accepted by the normal bridge path if present. """ def __init__( self, cfg: LabGameplayConfig, *, bridge: Any, forward: ForwardFn, emit: EmitFn, stop: threading.Event, ) -> None: self.cfg = cfg self.bridge = bridge self.forward = forward self.emit = emit self.stop = stop self.picks = _PickState() self._thread: threading.Thread | None = None self._ack_thread: threading.Thread | None = None self._seq = 1 self._tier_i = 0 self._outcome_q: SimpleQueue[Any] = SimpleQueue() self._lock = threading.Lock() self._phase = "idle" # idle | wait_x9 | wait_close def start(self) -> None: self._ack_thread = threading.Thread( target=self._ack_loop, name="lab_gameplay_ack", daemon=True, ) self._ack_thread.start() self._thread = threading.Thread( target=self._run, name="lab_gameplay_synth", daemon=True, ) self._thread.start() def join(self, timeout: float | None = None) -> None: if self._thread is not None: self._thread.join(timeout=timeout) if self._ack_thread is not None: self._ack_thread.join(timeout=0.5) def on_master_outcome(self, frame: bytes, meta: Any) -> None: """Bridge hook: Master x9 for this lab TID (queue — never block inject lock).""" if meta is None: return tid = getattr(meta, "tid_hex", None) if not tid or str(tid).upper() != self.cfg.tid_hex.upper(): return msg = getattr(meta, "msg_type", None) if msg is None or not is_gameplay_response(msg): return # Only ack while a synth round is open (or trunk still active). with self._lock: phase = self._phase if phase == "idle": try: if not ( callable(getattr(self.bridge, "_trunk_round_active_for_tid", None)) and self.bridge._trunk_round_active_for_tid(self.cfg.tid_hex) ): return except Exception: return self._outcome_q.put(meta) def _ack_loop(self) -> None: while not self.stop.is_set(): try: meta = self._outcome_q.get(timeout=0.2) except Empty: continue try: self._forge_ack_for_outcome(meta) except Exception as exc: self.emit( "lab_gameplay_synth_skip", reason="ack_failed", tid=self.cfg.tid_hex, detail=str(exc), ) def _forge_ack_for_outcome(self, meta: Any) -> None: ack_msg = ack_msg_for_response(int(meta.msg_type)) if ack_msg is None: return tid = str(meta.tid_hex).upper() game = str(meta.game_id_hex or "").upper() if not game: return seq = int(meta.play_sequence if meta.play_sequence is not None else 0) & 0xFF frame = build_gameplay_ack( ack_msg=ack_msg, tid_hex=tid, game_id_hex=game, play_sequence=seq, ) self.forward(frame) # Round complete for soak: x1 on the wire is enough (no skill/jpot, no 0D synth). # Inbound Master 0D is still accepted by the normal bridge path if it arrives. try: if callable(getattr(self.bridge, "_trunk_round_mark_x1", None)): self.bridge._trunk_round_mark_x1(tid) rk = None rk_fn = getattr(self.bridge, "_round_key", None) if callable(rk_fn): rk = rk_fn(meta) # Drop copper MRT wait armed by eth→UART x9 (no real T340 on this hop). if rk is not None: await_x1 = getattr(self.bridge, "_mrt_risk_await_x1", None) if isinstance(await_x1, dict): await_x1.pop(rk, None) await_0d = getattr(self.bridge, "_mrt_risk_await_0d", None) if isinstance(await_0d, dict): await_0d.pop(rk, None) pending = getattr(self.bridge, "_status_0d_x1_pending", None) if isinstance(pending, set): pending.discard(rk) if callable(getattr(self.bridge, "_trunk_round_close", None)): self.bridge._trunk_round_close(tid, game) except Exception as exc: self.emit( "lab_gameplay_synth_skip", reason="ack_close_failed", tid=tid, detail=str(exc), ) with self._lock: self._phase = "idle" self.emit( "lab_gameplay_synth_ack", msg=f"{ack_msg:02X}", len=len(frame), tid=tid, game=game, play_sequence=seq, x9_msg=getattr(meta, "msg_type_hex", None), close="on_x1", ) def _sleep(self, seconds: float) -> None: end = time.monotonic() + max(0.0, seconds) while not self.stop.is_set(): left = end - time.monotonic() if left <= 0: return self.stop.wait(min(left, 0.25)) def _next_period(self) -> float: j = self.cfg.jitter_s if j <= 0: return self.cfg.period_s return max(0.2, self.cfg.period_s + random.uniform(-j, j)) def _sync_phase_from_trunk(self) -> None: """Return to idle once bridge trunk round for lab TID has closed.""" try: active = ( callable(getattr(self.bridge, "_trunk_round_active_for_tid", None)) and self.bridge._trunk_round_active_for_tid(self.cfg.tid_hex) ) except Exception: active = False with self._lock: if self._phase != "idle" and not active: self._phase = "idle" elif self._phase == "idle" and active: self._phase = "wait_close" def _run(self) -> None: cfg = self.cfg err = arm_local_slave_tid(self.bridge, cfg.tid_hex, self.emit) if err: self.emit("lab_gameplay_synth_skip", reason="tid_collision", detail=err) self.emit("error", where="lab_gameplay_synth", message=err) return catalog = list(cfg.games) self._sleep(cfg.phase_s) self.emit( "lab_gameplay_synth_armed", tid=cfg.tid_hex, games=",".join(catalog), session_s=cfg.session_s, period_s=cfg.period_s, phase_s=cfg.phase_s, seat_sm=1, ) while not self.stop.is_set(): if cfg.shuffle: random.shuffle(catalog) for game_key in catalog: if self.stop.is_set(): return self._run_session(game_key) def _run_session(self, game_key: str) -> None: cfg = self.cfg if cfg.reroll_picks: _ensure_picks(self.picks, reroll=True) session_end = time.monotonic() + cfg.session_s self.emit( "lab_gameplay_session_start", game=game_key, tid=cfg.tid_hex, session_s=cfg.session_s, game_id=GAME_SPECS[game_key].game_id_hex, ) while not self.stop.is_set() and time.monotonic() < session_end: self._sync_phase_from_trunk() reason = busy_reason(self.bridge, cfg.tid_hex) if reason: self.emit( "lab_gameplay_synth_skip", reason=reason, game=game_key, tid=cfg.tid_hex, ) self._sleep(0.35) continue tier_i = self._tier_i if cfg.bet_rotate else 0 try: frame, meta = build_wager( game_key, tid_hex=cfg.tid_hex, tier_index=tier_i, play_sequence=self._seq, picks=self.picks, reroll_picks=False, touch6_replicate=cfg.touch6_replicate, ) except Exception as exc: self.emit( "lab_gameplay_synth_skip", reason="build_failed", game=game_key, detail=str(exc), ) self._sleep(self._next_period()) continue pm = parse_packet_meta(frame) try: self.forward(frame) except Exception as exc: self.emit( "lab_gameplay_synth_skip", reason="forward_failed", game=game_key, detail=str(exc), ) self._sleep(self._next_period()) continue try: # Open local trunk round so busy-skip waits for x9/x1/0D close. if callable(getattr(self.bridge, "_trunk_round_note_request", None)): self.bridge._trunk_round_note_request(pm) except Exception: pass with self._lock: self._phase = "wait_x9" self.emit( "lab_gameplay_synth_wager", msg=pm.msg_type_hex, len=len(frame), **meta, ) self._seq = (self._seq % 255) + 1 if cfg.bet_rotate: self._tier_i = (self._tier_i + 1) % 3 self._sleep(self._next_period()) self.emit( "lab_gameplay_session_end", game=game_key, tid=cfg.tid_hex, game_id=GAME_SPECS[game_key].game_id_hex, ) def try_start_lab_gameplay_synth( cfg: LabGameplayConfig, *, bridge: Any, forward: ForwardFn, emit: EmitFn, stop: threading.Event, prod_mode: bool = False, ) -> LabGameplaySynth | None: err = validate_config(cfg, prod_mode=prod_mode) if err: print(f"FATAL: {err}", flush=True) raise SystemExit(2) if not cfg.enabled: return None print( f"WARNING: LAB gameplay synth ENABLED tid={cfg.tid_hex} " f"games={','.join(cfg.games)} session_s={cfg.session_s} — " "full seat SM (x0→x9→x1 close-on-ack); NEVER use in production", flush=True, ) synth = LabGameplaySynth( cfg, bridge=bridge, forward=forward, emit=emit, stop=stop ) synth.start() try: bridge._lab_gameplay_synth = synth except Exception: pass return synth def _selftest() -> None: picks = _PickState() for key in DEFAULT_GAMES + ("keno", "spinjack", "triple7s"): for tier in range(3): frame, meta = build_wager( key, tid_hex=LAB_GAMEPLAY_TID_PILOT, tier_index=tier, play_sequence=1 + tier, picks=picks, touch6_replicate=(key == "touch6"), ) assert frame[0] == SOH and frame[-1] == 0x0D assert pog422_frame_checksum_ok(frame) pm = parse_packet_meta(frame) assert pm.tid_hex == LAB_GAMEPLAY_TID_PILOT assert pm.msg_type == GAME_SPECS[key].msg_type if key == "touch6": # Replicate keeps golden byte8 (autoplay 1-card 0x04) — only TID/seq/CRC. assert frame[8] == 0x04 assert meta["tier_label"] == "replicate" assert meta["credits"] == 4 elif key == "superpick": assert meta["tier_label"] == "replicate" assert frame[8] == 0x04 assert parse_packet_meta(frame).game_id_hex == "9508" else: assert meta["credits"] == GAME_SPECS[key].tiers[tier].credits # Ack opcode map + 12 B frame for every family. for wager, result in REQUEST_TO_RESPONSE.items(): ack = ack_msg_for_response(result) assert ack == (wager | 0x01) af = build_gameplay_ack( ack_msg=ack, tid_hex="40CC", game_id_hex="9107", play_sequence=3, ) assert len(af) == 12 and pog422_frame_checksum_ok(af) assert af[1] == ack and parse_packet_meta(af).tid_hex == "40CC" # 40CC allowed (outside E000–E0FF). f40, m40 = build_wager( "touch6", tid_hex=LAB_GAMEPLAY_TID_PILOT_ALT, tier_index=0, play_sequence=7, picks=picks, touch6_replicate=True, ) assert parse_packet_meta(f40).tid_hex == "40CC" assert f40[7] == 7 assert m40["golden_tid"] == "38D7" cfg40 = LabGameplayConfig(enabled=True, tid_hex="40CC", games=("touch6",)) assert validate_config(cfg40, prod_mode=False) is None # Keno spots stable across tiers. assert picks.keno_spots is not None and len(picks.keno_spots) == 10 f2, _ = build_wager( "keno", tid_hex=LAB_GAMEPLAY_TID_PILOT, tier_index=2, play_sequence=9, picks=picks, ) assert f2[KENO_WAGER_SPOT_MASK_OFFSET : KENO_WAGER_SPOT_MASK_OFFSET + 10] == _keno_spot_mask( picks.keno_spots ) # Bingo 1-card clears slots 2–4. b1, m1 = build_wager( "bingo", tid_hex="E001", tier_index=0, play_sequence=1, picks=picks ) assert m1["credits"] == 4 and m1["n_units"] == 1 assert b1[8] == 4 off = BINGO_WAGER_CARD_OFFSET + BINGO_CARD_BYTE_LEN assert b1[off : off + BINGO_CARD_BYTE_LEN] == bytes(BINGO_CARD_BYTE_LEN) cfg = LabGameplayConfig(enabled=True, tid_hex="E001", games=DEFAULT_GAMES) assert validate_config(cfg, prod_mode=False) is None assert validate_config(cfg, prod_mode=True) is not None print("lab_gameplay_synth selftest OK") if __name__ == "__main__": _selftest()