from __future__ import annotations """OHD progressive display schedule builders for prog_display_mode 1..4.""" import json from dataclasses import dataclass from datetime import datetime, timezone from typing import Dict, List, Optional, Sequence, Tuple from ethergold_reader import ( EthergoldReader, PROGRESSIVE_GAME_IDS, parse_display_to_pennies, ) # Mode 1: seconds per progressive game (clamped in load_config). MODE_1_MIN_SEC = 10 MODE_1_MAX_SEC = 60 MAX_FLOOR_PROGRESSIVE_GAMES = 12 MAX_FLOOR_PROGRESSIVE_GAMES_WITH_BG21 = 11 BLACK_GOLD_21_GAME_ID = 15 DEFAULT_FOCUS_FRACTION = 0.75 DEFAULT_ROTATION_BATCH = 2 DEFAULT_ACTIVITY_MIN_SAMPLES = 12 DEFAULT_ACTIVITY_WINDOW_SEC = 60.0 # Obsolete for mode-4 scheduling (kept for config compat / modes diagnosis only). # Thresholds are leader-minus-runner-up margin (pennies), written as decimal dollars (0.50 = 50¢). DEFAULT_ACTIVITY_TIERS_RAW = "0.50:100, 0.40:80, 0.30:70, 0.15:60, 0.05:50" # Mode 4: weakest active gets mode_1_display_time; others scale up by score ratio. MODE_4_DWELL_MAX_MULT = 12 PROGRESSIVE_GAME_LABELS: Dict[int, str] = { 1: "Pot-O-Gold", 2: "Quick Cash", 3: "Touch 6", 4: "Bingo", 5: "Keno", 6: "Super Pick", 7: "Spin Jack", 8: "Black Gold", 9: "TouchEasy Keno", 10: "SuperBall Keno", 11: "SuperBall Keno", 12: "Triple 7's", 13: "Triple 7's", 14: "SpinBall Bonus", 16: "Other", } Schedule = List[Tuple[int, float]] def coalesce_schedule(schedule: Schedule) -> Schedule: """Merge adjacent segments for the same game (one continuous display, no re-fade).""" if not schedule: return [] merged: Schedule = [schedule[0]] for game_id, duration in schedule[1:]: last_game, last_duration = merged[-1] if game_id == last_game: merged[-1] = (last_game, last_duration + float(duration)) else: merged.append((game_id, float(duration))) return merged @dataclass(frozen=True) class FloorActivityRank: """Ranked floor games by activity score (pennies per CSV capture).""" ranked: Tuple[Tuple[int, float], ...] metric: str leader_game: Optional[int] leader_score: float runner_up_score: float margin: float def rank_floor_activity( reader: EthergoldReader, floor_games: Sequence[int], min_samples: int, ) -> FloorActivityRank: """Rank all floor games; margin = leader score minus runner-up (legacy / diagnosis).""" diagnostics = reader.get_activity_diagnostics(floor_games) use_avg = any( int(diagnostics.get(game_id, {}).get("samples", 0)) >= min_samples for game_id in floor_games ) scores: Dict[int, float] = {} for game_id in floor_games: detail = diagnostics.get(game_id, {}) samples = int(detail.get("samples", 0)) if use_avg and samples >= min_samples: scores[game_id] = float(detail.get("avg_pennies", 0.0)) else: scores[game_id] = float(detail.get("last_delta_pennies", 0)) metric = "rolling_avg" if use_avg else "latest_delta" ranked = tuple( sorted(scores.items(), key=lambda item: (-item[1], item[0])) ) leader_game: Optional[int] = None leader_score = 0.0 runner_up_score = 0.0 margin = 0.0 if ranked and ranked[0][1] > 0: leader_game = ranked[0][0] leader_score = ranked[0][1] runner_up_score = ranked[1][1] if len(ranked) > 1 else 0.0 margin = max(0.0, leader_score - runner_up_score) return FloorActivityRank( ranked=ranked, metric=metric, leader_game=leader_game, leader_score=leader_score, runner_up_score=runner_up_score, margin=margin, ) @dataclass(frozen=True) class Mode34BuildInfo: focus_game: Optional[int] focus_fraction: float tier_percent: Optional[int] strategy: str fallback_reason: str @dataclass(frozen=True) class Mode4RatioBuildInfo: """Last mode-4 ratio dwell build (dashboard / diagnosis).""" strategy: str fallback_reason: str min_active_score: float active_count: int clamp_hits: int dwell_sec: Tuple[Tuple[int, float], ...] # (game_id, planned dwell) @dataclass(frozen=True) class OhdProgressiveSettings: floor_games: Tuple[int, ...] game_time_sec: float focus_fraction: float rotation_batch: int activity_min_samples: int activity_window_sec: float activity_tiers: Tuple[Tuple[int, int], ...] def clamp_mode_1_display_time(seconds: int) -> int: return max(MODE_1_MIN_SEC, min(MODE_1_MAX_SEC, int(seconds))) def _progressive_capable_game_ids() -> Tuple[int, ...]: return tuple(g for g in range(1, 17) if g != BLACK_GOLD_21_GAME_ID) def sanitize_progressive_floor_games( game_ids: Sequence[int], *, black_gold_21_enabled: bool, ) -> Tuple[int, ...]: """Strip Black Gold 21 and invalid IDs. Never invents a default floor list.""" max_count = ( MAX_FLOOR_PROGRESSIVE_GAMES_WITH_BG21 if black_gold_21_enabled else MAX_FLOOR_PROGRESSIVE_GAMES ) capable = set(_progressive_capable_game_ids()) parsed: List[int] = [] for game_id in game_ids: gid = int(game_id) if gid == BLACK_GOLD_21_GAME_ID: print( f"WARNING - progressive_floor_games ignores game {BLACK_GOLD_21_GAME_ID} " "(Black Gold 21 has no progressive)", flush=True, ) continue if gid not in capable: print(f"WARNING - progressive_floor_games skipping invalid game_id {gid}", flush=True) continue if gid not in parsed: parsed.append(gid) games = tuple(parsed) if len(games) > max_count: print( f"WARNING - progressive_floor_games has {len(games)} entries; " f"trimming to {max_count} (black_gold_21_enabled={black_gold_21_enabled})", flush=True, ) games = games[:max_count] return games def parse_progressive_floor_games( raw: str, *, black_gold_21_enabled: bool, ) -> Tuple[int, ...]: """Parse an explicit floor list. Empty input → empty tuple (no invented 1–12).""" if not raw.strip(): return () parsed: List[int] = [] for part in raw.split(","): text = part.strip() if not text: continue parsed.append(int(text)) return sanitize_progressive_floor_games( parsed, black_gold_21_enabled=black_gold_21_enabled, ) def parse_activity_tiers(raw: str) -> Tuple[Tuple[int, int], ...]: tiers: List[Tuple[int, int]] = [] text = raw.strip() or DEFAULT_ACTIVITY_TIERS_RAW for part in text.split(","): piece = part.strip() if not piece: continue if ":" not in piece: print( f"WARNING - invalid activity_tiers entry {piece!r}; " "expected margin_cents_as_dollars:percent (0.50:100 = 50¢ margin → 100%)", flush=True, ) continue dollars_text, percent_text = piece.split(":", 1) pennies = parse_display_to_pennies(dollars_text.strip()) percent = int(percent_text.strip()) percent = max(0, min(100, percent)) tiers.append((pennies, percent)) tiers.sort(key=lambda item: item[0], reverse=True) if not tiers: return parse_activity_tiers(DEFAULT_ACTIVITY_TIERS_RAW) return tuple(tiers) def focus_percent_for_average_delta( average_delta_pennies: float, tiers: Sequence[Tuple[int, int]], ) -> int: for threshold, percent in tiers: if average_delta_pennies >= threshold: return percent return 0 def mode_1_total_cycle_sec(settings: OhdProgressiveSettings) -> float: return len(settings.floor_games) * float(settings.game_time_sec) def build_mode_1_schedule( settings: OhdProgressiveSettings, rotation_index: int = 0, ) -> Tuple[Schedule, int]: duration = float(settings.game_time_sec) games = settings.floor_games if not games: return [], 0 schedule: Schedule = [] for offset in range(len(games)): game_id = games[(rotation_index + offset) % len(games)] schedule.append((game_id, duration)) next_index = (rotation_index + len(games)) % len(games) return schedule, next_index def mode_4_dwell_seconds( score: float, *, min_active_score: float, base_sec: float, max_mult: int = MODE_4_DWELL_MAX_MULT, ) -> Tuple[float, bool]: """Return (dwell_sec, clamped). Weakest active gets base_sec; others scale by score ratio.""" base = float(base_sec) if min_active_score <= 0 or score <= 0: return base, False raw = base * (float(score) / float(min_active_score)) lo = base hi = base * max(1, int(max_mult)) dwell = max(lo, min(hi, raw)) return dwell, abs(dwell - raw) > 0.01 def build_mode_4_ratio_schedule( rank: FloorActivityRank, settings: OhdProgressiveSettings, ) -> Tuple[Schedule, Mode4RatioBuildInfo]: """Active games only; dwell proportional to activity score vs weakest active.""" base = float(settings.game_time_sec) active = [(gid, score) for gid, score in rank.ranked if score > 0] if not active: schedule, _ = build_mode_1_schedule(settings, 0) info = Mode4RatioBuildInfo( strategy="mode1_fallback", fallback_reason="no_activity", min_active_score=0.0, active_count=0, clamp_hits=0, dwell_sec=tuple((gid, base) for gid, _ in schedule), ) return schedule, info min_score = min(score for _, score in active) schedule: Schedule = [] dwell_pairs: List[Tuple[int, float]] = [] clamp_hits = 0 for game_id, score in active: dwell, clamped = mode_4_dwell_seconds( score, min_active_score=min_score, base_sec=base ) if clamped: clamp_hits += 1 schedule.append((game_id, dwell)) dwell_pairs.append((game_id, dwell)) info = Mode4RatioBuildInfo( strategy="ratio_dwell", fallback_reason=f"rank_{rank.metric}", min_active_score=min_score, active_count=len(active), clamp_hits=clamp_hits, dwell_sec=tuple(dwell_pairs), ) return coalesce_schedule(schedule), info def build_interleaved_focus_schedule( focus_game: int, settings: OhdProgressiveSettings, focus_fraction: float, rotation_index: int, ) -> Tuple[Schedule, int]: """Normal rotation (excluding highlight) interrupted by highlight slots.""" active = settings.floor_games game_time = float(settings.game_time_sec) if focus_game not in active: return build_mode_1_schedule(settings, rotation_index) if focus_fraction >= 1.0: total = mode_1_total_cycle_sec(settings) return [(focus_game, total)], rotation_index rotation = [g for g in active if g != focus_game] if not rotation: total = mode_1_total_cycle_sec(settings) return [(focus_game, total)], rotation_index total = mode_1_total_cycle_sec(settings) focus_target = total * focus_fraction other_target = total - focus_target focus_remaining = focus_target other_remaining = other_target rot_idx = rotation_index schedule: Schedule = [] batch_size = max(1, int(settings.rotation_batch)) while focus_remaining > 0.01 or other_remaining > 0.01: batch = 0 while other_remaining > 0.01 and batch < batch_size: game_id = rotation[rot_idx % len(rotation)] rot_idx += 1 seg = min(game_time, other_remaining) if seg <= 0: break schedule.append((game_id, seg)) other_remaining -= seg batch += 1 if focus_remaining > 0.01: seg = min(game_time, focus_remaining) schedule.append((focus_game, seg)) focus_remaining -= seg elif other_remaining <= 0.01: break return schedule, rot_idx % len(rotation) class OhdProgressiveScheduler: def __init__(self, settings: OhdProgressiveSettings, reader: EthergoldReader): self.settings = settings self.reader = reader self._rotation_index = 0 self._last_build_info: Optional[Mode34BuildInfo] = None self._last_mode4_info: Optional[Mode4RatioBuildInfo] = None self._last_schedule: Schedule = [] self.apply_reader_settings() def update_settings(self, settings: OhdProgressiveSettings) -> None: self.settings = settings self.apply_reader_settings() def apply_reader_settings(self) -> None: self.reader.configure_activity_averaging( self.settings.activity_min_samples, self.settings.activity_window_sec, ) def _resolve_focus_game( self, floor_games: Sequence[int] ) -> Tuple[Optional[int], str]: rank = rank_floor_activity( self.reader, floor_games, self.settings.activity_min_samples ) if rank.leader_game is not None: return rank.leader_game, f"rank_{rank.metric}" return None, "none" def modes_status_snapshot(self, mode: int) -> Dict[str, object]: """Structured live snapshot for /ohd/api/modes-status and diagnosis dump.""" settings = self.settings floor = list(settings.floor_games) rank = rank_floor_activity(self.reader, floor, settings.activity_min_samples) diagnostics = self.reader.get_activity_diagnostics(floor) dwell_map = {gid: secs for gid, secs in (self._last_mode4_info.dwell_sec if self._last_mode4_info else ())} if not dwell_map and mode == 4 and rank.leader_score > 0: # Preview planned dwells even before first build_schedule call. _sched, preview = build_mode_4_ratio_schedule(rank, settings) dwell_map = {gid: secs for gid, secs in preview.dwell_sec} mode4_preview = preview else: mode4_preview = self._last_mode4_info games_detail: Dict[str, Dict[str, object]] = {} active_count = 0 for game_id in floor: detail = diagnostics.get(game_id, {}) samples = int(detail.get("samples", 0)) avg = float(detail.get("avg_pennies", 0.0)) delta = float(detail.get("last_delta_pennies", 0)) score = ( avg if rank.metric == "rolling_avg" and samples >= settings.activity_min_samples else delta ) active = score > 0 if active: active_count += 1 rank_pos = next( (idx + 1 for idx, (gid, _) in enumerate(rank.ranked) if gid == game_id), 0, ) planned = dwell_map.get(game_id) games_detail[str(game_id)] = { "name": PROGRESSIVE_GAME_LABELS.get(game_id, f"Game {game_id}"), "samples": samples, "avg_c": round(avg, 2), "delta_c": int(delta), "score_c": round(score, 2), "rank": rank_pos, "active": active, "skipped": not active, "planned_dwell_sec": None if planned is None else round(float(planned), 1), "ratio_vs_weakest": ( None if not active or mode4_preview is None or mode4_preview.min_active_score <= 0 else round(score / mode4_preview.min_active_score, 3) ), } schedule_rows = [ {"game": gid, "name": PROGRESSIVE_GAME_LABELS.get(gid, f"Game {gid}"), "sec": round(secs, 1)} for gid, secs in self._last_schedule ] schedule_total = round(sum(secs for _, secs in self._last_schedule), 1) last = self._last_build_info payload: Dict[str, object] = { "utc": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "mode": mode, "config": { "prog_display_mode": mode, "mode_1_display_time": float(settings.game_time_sec), "ohd_focus_fraction": float(settings.focus_fraction), "ohd_rotation_batch": int(settings.rotation_batch), "activity_avg_min_samples": int(settings.activity_min_samples), "activity_avg_window_sec": float(settings.activity_window_sec), "activity_tiers_obsolete_for_mode4": True, }, "floor_games": floor, "active_games": active_count, "idle_games": max(0, len(floor) - active_count), "rank_metric": rank.metric, "leader": rank.leader_game, "leader_score_c": round(rank.leader_score, 2), "runner_up_score_c": round(rank.runner_up_score, 2), "margin_c": round(rank.margin, 2), "ranked": [ {"game": gid, "score_c": round(score, 2)} for gid, score in rank.ranked ], "games": games_detail, "schedule": schedule_rows, "schedule_total_sec": schedule_total, "last_build_mode3": None if last is None else { "focus": last.focus_game, "focus_frac": round(last.focus_fraction, 3), "tier_pct": last.tier_percent, "strategy": last.strategy, "reason": last.fallback_reason, }, "mode4": None if mode4_preview is None else { "strategy": mode4_preview.strategy, "reason": mode4_preview.fallback_reason, "min_active_score_c": round(mode4_preview.min_active_score, 2), "active_count": mode4_preview.active_count, "clamp_hits": mode4_preview.clamp_hits, "dwell_max_mult": MODE_4_DWELL_MAX_MULT, }, # Legacy keys for /EtherGold/mode4-weights.txt consumers "min_samples": settings.activity_min_samples, "min_samples_note": "~12 master broadcasts at ~5s (~60s) before rolling avg rank", "window_sec": settings.activity_window_sec, "tiers_c": [{"margin_thresh_c": t, "pct": p} for t, p in settings.activity_tiers], "margin_tier_pct": focus_percent_for_average_delta( rank.margin, settings.activity_tiers ), "sole_active_lock": rank.leader_score > 0 and active_count == 1, "last_build": None if mode4_preview is None and last is None else ( { "focus": None, "focus_frac": 0.0, "tier_pct": None, "strategy": mode4_preview.strategy, "reason": mode4_preview.fallback_reason, } if mode4_preview is not None and mode == 4 else { "focus": last.focus_game if last else None, "focus_frac": round(last.focus_fraction, 3) if last else 0.0, "tier_pct": last.tier_percent if last else None, "strategy": last.strategy if last else "", "reason": last.fallback_reason if last else "", } ), } if self._last_schedule: counts: Dict[int, float] = {} for game_id, duration in self._last_schedule: counts[game_id] = counts.get(game_id, 0.0) + duration payload["last_schedule_sec"] = { str(game_id): round(secs, 1) for game_id, secs in counts.items() } return payload def format_mode4_diagnosis(self, mode: int) -> str: """One JSON line: live activity weights and last schedule build (on-demand dump).""" return json.dumps(self.modes_status_snapshot(mode), separators=(",", ":")) def build_schedule(self, mode: int) -> Schedule: settings = self.settings settings = OhdProgressiveSettings( floor_games=settings.floor_games, game_time_sec=float(clamp_mode_1_display_time(int(settings.game_time_sec))), focus_fraction=settings.focus_fraction, rotation_batch=settings.rotation_batch, activity_min_samples=settings.activity_min_samples, activity_window_sec=settings.activity_window_sec, activity_tiers=settings.activity_tiers, ) if mode == 1: schedule, self._rotation_index = build_mode_1_schedule( settings, self._rotation_index ) self._last_schedule = schedule self._last_mode4_info = None return schedule if mode == 2: focus_game = self.reader.get_highest_value_game(settings.floor_games) if focus_game is None: schedule, self._rotation_index = build_mode_1_schedule( settings, self._rotation_index ) self._last_schedule = schedule self._last_mode4_info = None return schedule schedule, self._rotation_index = build_interleaved_focus_schedule( focus_game, settings, settings.focus_fraction, self._rotation_index, ) self._last_build_info = Mode34BuildInfo( focus_game, settings.focus_fraction, int(round(settings.focus_fraction * 100)), "interleaved_focus", "mode2_highest_jackpot", ) self._last_schedule = schedule self._last_mode4_info = None return schedule if mode == 3: rank = rank_floor_activity( self.reader, settings.floor_games, settings.activity_min_samples ) focus_game = rank.leader_game focus_source = f"rank_{rank.metric}" if focus_game is not None else "none" if focus_game is None: self._last_build_info = Mode34BuildInfo( None, 0.0, None, "mode1_fallback", "no_activity" ) schedule, self._rotation_index = build_mode_1_schedule( settings, self._rotation_index ) self._last_schedule = schedule self._last_mode4_info = None return schedule focus_fraction = settings.focus_fraction tier_percent: Optional[int] = int(round(focus_fraction * 100)) build_reason = f"mode3_{focus_source}" schedule, self._rotation_index = build_interleaved_focus_schedule( focus_game, settings, focus_fraction, self._rotation_index, ) self._last_build_info = Mode34BuildInfo( focus_game, focus_fraction, tier_percent, "interleaved_focus", build_reason, ) self._last_schedule = schedule self._last_mode4_info = None return schedule if mode == 4: rank = rank_floor_activity( self.reader, settings.floor_games, settings.activity_min_samples ) schedule, info = build_mode_4_ratio_schedule(rank, settings) self._last_mode4_info = info self._last_build_info = Mode34BuildInfo( rank.leader_game, 0.0, None, info.strategy, info.fallback_reason, ) self._last_schedule = schedule return schedule schedule, self._rotation_index = build_mode_1_schedule( settings, self._rotation_index ) self._last_schedule = schedule self._last_mode4_info = None return schedule def summarize_schedule(mode: int, schedule: Schedule, settings: OhdProgressiveSettings) -> str: total = sum(duration for _, duration in schedule) counts: Dict[int, float] = {} for game_id, duration in schedule: counts[game_id] = counts.get(game_id, 0.0) + duration focus_bits = ", ".join( f"g{game_id}={secs:.1f}s" for game_id, secs in sorted(counts.items(), key=lambda item: -item[1])[:4] ) return ( f"OHD prog mode {mode}: {len(settings.floor_games)} floor games, " f"{len(schedule)} segments, {total:.1f}s cycle ({focus_bits}…)" ) def load_ohd_progressive_settings( config, *, floor_games_override: Optional[Sequence[int]] = None, black_gold_21_override: Optional[bool] = None, ) -> OhdProgressiveSettings: if black_gold_21_override is not None: black_gold_21_enabled = black_gold_21_override else: black_gold_21_enabled = ( config.getboolean("TOPBOX", "BLACK_GOLD_21_ENABLED") if config.has_option("TOPBOX", "BLACK_GOLD_21_ENABLED") else False ) if floor_games_override is not None: floor_games = sanitize_progressive_floor_games( floor_games_override, black_gold_21_enabled=black_gold_21_enabled, ) else: floor_raw = ( config.get("TOPBOX", "PROGRESSIVE_FLOOR_GAMES") if config.has_option("TOPBOX", "PROGRESSIVE_FLOOR_GAMES") else "" ) floor_games = parse_progressive_floor_games( floor_raw, black_gold_21_enabled=black_gold_21_enabled, ) focus_fraction = ( config.getfloat("TOPBOX", "OHD_FOCUS_FRACTION") if config.has_option("TOPBOX", "OHD_FOCUS_FRACTION") else DEFAULT_FOCUS_FRACTION ) rotation_batch = ( config.getint("TOPBOX", "OHD_ROTATION_BATCH") if config.has_option("TOPBOX", "OHD_ROTATION_BATCH") else DEFAULT_ROTATION_BATCH ) activity_min_samples = ( config.getint("TOPBOX", "ACTIVITY_AVG_MIN_SAMPLES") if config.has_option("TOPBOX", "ACTIVITY_AVG_MIN_SAMPLES") else DEFAULT_ACTIVITY_MIN_SAMPLES ) activity_window_sec = ( config.getfloat("TOPBOX", "ACTIVITY_AVG_WINDOW_SEC") if config.has_option("TOPBOX", "ACTIVITY_AVG_WINDOW_SEC") else DEFAULT_ACTIVITY_WINDOW_SEC ) activity_tiers_raw = ( config.get("TOPBOX", "ACTIVITY_TIERS") if config.has_option("TOPBOX", "ACTIVITY_TIERS") else DEFAULT_ACTIVITY_TIERS_RAW ) return OhdProgressiveSettings( floor_games=floor_games, game_time_sec=float(clamp_mode_1_display_time(config.getint("TOPBOX", "MODE_1_DISPLAY_TIME"))), focus_fraction=max(0.0, min(1.0, focus_fraction)), rotation_batch=max(1, rotation_batch), activity_min_samples=max(1, activity_min_samples), activity_window_sec=max(1.0, activity_window_sec), activity_tiers=parse_activity_tiers(activity_tiers_raw), )