from __future__ import annotations """Player-side controller for the full-screen progressive slide. This module bridges the reusable odometer widget into the DyzineFrame player. It owns one cached widget per progressive game, reads live values from EthergoldReader (local progressives.csv or remote http(s) URL), and renders a complete slide with background, bezel, and odometer digits already composed. Odometer engines keep advancing via tick_game(s) even while logos/ads play so the display stays in sync with the floor LED sign when the progressive slide returns (no snap-to-current on re-entry). """ from typing import Callable, Dict, Iterable, Optional, Tuple import pygame from ethergold_reader import ( EthergoldReader, PROGRESSIVE_GAME_IDS, format_dollars, resolve_progressives_source, ) from progressive_fade import run_staged_progressive_fade_in from reusable_odometer import create_current_widget, optimize_theme_surfaces class PlayerProgressiveDisplay: def __init__( self, *, resolution: str = "1920x1080", progressive_reader: Optional[EthergoldReader] = None, progressives_source: str | None = None, remote_poll_interval_s: float | None = None, chase_rate_scale: float = 1.0, catchup_lag_threshold: int = 20, catchup_lag_bonus_per_cent: float = 0.0, snap_behind_threshold_pennies: int = 1000, snap_behind_to_lag_pennies: int = 200, ): self.resolution = resolution self._chase_rate_scale = max(0.05, float(chase_rate_scale)) self._catchup_lag_threshold = max(0, int(catchup_lag_threshold)) self._catchup_lag_bonus_per_cent = max(0.0, float(catchup_lag_bonus_per_cent)) self._snap_behind_threshold_pennies = max(0, int(snap_behind_threshold_pennies)) self._snap_behind_to_lag_pennies = max(0, int(snap_behind_to_lag_pennies)) if progressive_reader is not None: self.reader = progressive_reader else: source = progressives_source or resolve_progressives_source() self.reader = EthergoldReader( source, remote_poll_interval_s=remote_poll_interval_s, ) print( f"PlayerProgressiveDisplay: progressives source " f"({self.reader.source_kind}) -> {self.reader.source}", flush=True, ) if self.reader.source_kind == "remote": print( f"PlayerProgressiveDisplay: remote poll interval " f"{self.reader.remote_poll_interval_s:.1f}s " "(background thread, master broadcast cadence ~5s)", flush=True, ) self.reader.start_background_poll() elif self.reader.source_kind == "local": self.reader.load(force=True) self._widgets: Dict[int, object] = {} self._failed_games: set[int] = set() self._slide_surface: Optional[pygame.Surface] = None self._slide_size: Optional[Tuple[int, int]] = None @staticmethod def supports_game(game_id: int) -> bool: return game_id in PROGRESSIVE_GAME_IDS def _get_widget(self, game_id: int): if game_id in self._failed_games: return None widget = self._widgets.get(game_id) if widget is not None: return widget try: widget = create_current_widget( game_id, self.resolution, include_background=True, include_frame=True, ) optimize_theme_surfaces(widget.theme) widget.engine.config.chase_rate_scale = self._chase_rate_scale widget.engine.config.catchup_lag_threshold = self._catchup_lag_threshold widget.engine.config.catchup_lag_bonus_per_cent = self._catchup_lag_bonus_per_cent widget.engine.config.snap_behind_threshold_pennies = self._snap_behind_threshold_pennies widget.engine.config.snap_behind_to_lag_pennies = self._snap_behind_to_lag_pennies except Exception as exc: print(f"PlayerProgressiveDisplay: disabling game {game_id}: {exc}") self._failed_games.add(game_id) return None self._widgets[game_id] = widget return widget def _ensure_surface(self, size: Tuple[int, int]) -> pygame.Surface: if self._slide_surface is not None and self._slide_size == size: return self._slide_surface self._slide_surface = pygame.Surface(size, pygame.SRCALPHA) self._slide_size = size return self._slide_surface def _sync_widget_live(self, widget, game_id: int) -> None: """Pull latest CSV target; snap only on the first live value ever seen.""" self.reader.reload_if_changed() target = self.reader.peek_target(game_id) if target is None: widget.clear_live_target() return widget.set_target_pennies(target, first_value_immediate=True, mark_live=True) def tick_game(self, game_id: int, dt_seconds: float) -> None: """Advance odometer engine without rendering (logos, ads, splash, etc.).""" if not self.supports_game(game_id): return widget = self._get_widget(game_id) if widget is None: return self._sync_widget_live(widget, game_id) if dt_seconds > 0: widget.update(dt_seconds) def tick_games(self, game_ids: Iterable[int], dt_seconds: float) -> None: self.reader.reload_if_changed() for game_id in game_ids: self.tick_game(game_id, dt_seconds) def prepare_for_slide(self, game_id: int) -> bool: """Ensure widget exists and targets are current; do not snap on re-entry.""" if not self.supports_game(game_id): return False widget = self._get_widget(game_id) if widget is None: return False self._sync_widget_live(widget, game_id) return True def render_slide( self, game_id: int, dt_seconds: float, output_size: Optional[Tuple[int, int]] = None, ) -> Tuple[Optional[pygame.Surface], bool]: """Return a fully rendered progressive slide and whether it changed.""" if not self.supports_game(game_id): return None, False widget = self._get_widget(game_id) if widget is None: return None, False before_target = widget.target_pennies before_display = widget.display_pennies self._sync_widget_live(widget, game_id) if dt_seconds > 0: widget.update(dt_seconds) changed = ( widget.target_pennies != before_target or widget.display_pennies != before_display or widget.is_animating() ) native_size = widget.theme.layout.screen_size if native_size == (0, 0): native_size = output_size or (1920, 1080) rendered = self._ensure_surface(native_size) rendered.fill((0, 0, 0, 0)) widget.draw(rendered) if output_size is None or output_size == native_size: return rendered, changed scaled = pygame.transform.scale(rendered, output_size) return scaled, changed def play_staged_fade_in( self, game_id: int, win: pygame.Surface, output_size: Tuple[int, int], *, under: Optional[pygame.Surface] = None, tick_background: Optional[Callable[[float], None]] = None, should_abort: Optional[Callable[[], bool]] = None, on_keydown: Optional[Callable[[str], None]] = None, ) -> bool: """Three-beat fade: BG, then frame+blocking, then digits.""" if not self.prepare_for_slide(game_id): return False widget = self._widgets.get(game_id) if widget is None: return False return run_staged_progressive_fade_in( win, widget, output_size, under=under, tick_background=tick_background, should_abort=should_abort, on_keydown=on_keydown, ) def debug_timing(self, game_id: int) -> Optional[dict]: """Live chase stats for comparing scroll rate to the floor LED sign.""" if not self.supports_game(game_id): return None widget = self._widgets.get(game_id) if widget is None: widget = self._get_widget(game_id) if widget is None: return None engine = widget.engine lag = engine.lag_pennies() return { "display_pennies": widget.display_pennies, "target_pennies": widget.target_pennies, "lag_pennies": lag, "catchup_mode": engine.catchup_mode, "chase_cents_per_sec": engine.effective_pennies_per_second(lag), "curve_cents_per_sec": engine.pennies_per_second(lag), "contributions_active": engine._contributions_active(), "catchup_bonus_per_sec": ( max( 0.0, (lag - max(0, int(engine.config.catchup_lag_threshold))) * max(0.0, float(engine.config.catchup_lag_bonus_per_cent)), ) if engine._contributions_active() and lag > max(0, int(engine.config.catchup_lag_threshold)) else 0.0 ), } def debug_status(self, game_id: int) -> Optional[str]: if not self.supports_game(game_id): return None widget = self._get_widget(game_id) if widget is None: return None live_target = self.reader.peek_target(game_id) if live_target is None: return ( f"Progressive slide active: game={game_id} " f"reader={self.reader.source_kind} " f"source=blank (no $ until fresh progressives.csv)" ) return ( f"Progressive slide active: game={game_id} " f"reader={self.reader.source_kind} " f"source=live " f"target={format_dollars(widget.target_pennies)} " f"display={format_dollars(widget.display_pennies)}" )