diff --git a/docs/SPORTS_UNIFICATION.md b/docs/SPORTS_UNIFICATION.md index 3cd6ab21..31858ff9 100644 --- a/docs/SPORTS_UNIFICATION.md +++ b/docs/SPORTS_UNIFICATION.md @@ -95,6 +95,29 @@ deprecation cycle. | `_custom_scorebug_layout(game, draw)` | Per-sport overlay on the base layout | no-op | | `render_skin_card(game, size)` | Skin-system entry point | built-in fallback | | `score_phrase(game)` | Celebration wording (`"GOAL"` vs `"TOUCHDOWN"`) | `"SCORE"` — only consulted when `CelebrationMixin` is present | +| `_favorite_key(game, side)` | Which view-model field identifies a team for favorites matching | `game["_abbr"]` | +| `_config_schema_path()` | Plugin's `config_schema.json` — returning it routes `_get_layout_offset` through the `src.element_style` resolver (and gives it the defaults to compare against) | `None`, i.e. the classic inline `customization.layout` read | +| `_font_root()` | Directory to resolve `assets/fonts` against | core install root | + +Two class attributes serve the same purpose for values that are per-sport +constants rather than behavior: + +| Attribute | Meaning | Default | +|---|---|---| +| `FINAL_PERIOD` | Period at/after which a zero clock can mean "over" | `4` (hockey overrides to `3`) | +| `CLOCK_COUNTS_DOWN` | Whether `0:00` means "expired" | `True` (soccer/afl/nrl override to `False` — their clocks count up, so `0:00` is kickoff) | + +### Why these are seams and not branches + +`_favorite_key` exists because NRL abbreviations are **not unique** — "NEW" is both +Newcastle Knights and New Zealand Warriors, "CAN" both Canberra and Canterbury — +so NRL matches favorites on team ID. Flattening every plugin to abbreviations +would silently select the wrong club for NRL users. The base declares the seam, +NRL fills it, and core never learns the string `"nrl"`. + +`CLOCK_COUNTS_DOWN` exists for the same reason in the opposite direction: a +soccer clock reading `0:00` means the match has not kicked off, so running the +clock-expiry branch there would evict live games. ## Phases diff --git a/src/base_classes/sports/core.py b/src/base_classes/sports/core.py index edb70f9c..b4fc92da 100644 --- a/src/base_classes/sports/core.py +++ b/src/base_classes/sports/core.py @@ -10,7 +10,7 @@ import time from abc import ABC, abstractmethod from datetime import datetime, timedelta from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple import pytz import requests @@ -32,6 +32,77 @@ try: except ImportError: OddsManager = None +# The core install root, resolved from this module's own location so nothing +# here depends on the process working directory. +# +# The index is the number of directories between this file and the root: +# src/base_classes/sports/core.py -> sports -> base_classes -> src -> root. +# It is defined ONCE, here, because hand-counting it at each use site is +# exactly what broke when this module moved from src/base_classes/sports.py +# into the package (the old depth of 2 silently started resolving to src/). +# If this module ever moves again, this is the single line to update. +_INSTALL_ROOT = Path(__file__).resolve().parents[3] + +# Shared element-style resolver. Core always ships src.element_style, so the +# guard is never taken here — it is kept because this module's methods are +# back-copied verbatim into the plugins' bundled `sports.py`, where older +# cores genuinely lack the module (the plugins fall back to the classic +# inline config read below). +try: + from src.element_style import ElementStyleResolver, defaults_from_schema_file + STYLE_AVAILABLE = True +except ImportError: # pragma: no cover - core always ships element_style + STYLE_AVAILABLE = False + + +# --- Font catalog delegation ------------------------------------------------- +# The plugin copies each carried their own alias table mapping font *family* +# names ("press_start") to filenames. That table is a duplicate of the core +# FontManager's `common_fonts` catalog, so resolve through the catalog instead +# and let the two never drift apart. Same lazy module-level shape as +# skin_runtime._get_font_manager / base_plugin._fallback_font_manager: a +# SportsCore host has no plugin_manager to borrow a FontManager from. + +_shared_font_catalog: Optional[Any] = None + + +def _font_catalog() -> Any: + """Shared read-only FontManager used purely as a font-name catalog.""" + global _shared_font_catalog + if _shared_font_catalog is None: + from src.font_manager import FontManager + _shared_font_catalog = FontManager({}) + return _shared_font_catalog + + +def _resolve_font_family_alias(font_name: str) -> str: + """Resolve a font family alias to its filename, leaving filenames as-is. + + A config `font` value may be either a family name from the FontManager + catalog ("press_start", "four_by_six", "five_by_seven") or a literal + filename; the former resolve here, the latter pass through unchanged. + """ + try: + catalogued = _font_catalog().common_fonts.get(font_name) + except Exception: + return font_name + return os.path.basename(catalogued) if catalogued else font_name + + +def _read_bdf_native_size(bdf_path: str) -> Optional[int]: + """A BDF file's one true pixel size, delegated to FontManager. + + Deliberately NOT reimplemented here: the plugin copies grew a variant + that scans the whole file and returns a partially collected value when + parsing raises, where FontManager's stops at the first STARTCHAR and + returns None on error. + """ + try: + from src.font_manager import FontManager + return FontManager._read_bdf_native_size(bdf_path) + except Exception: + return None + class SportsCore(ABC): # Which ScoreboardSkin render method this class's display path maps to. @@ -97,6 +168,13 @@ class SportsCore(ABC): self._logo_cache = {} + # Font caches for _load_custom_font_from_element_config: per-frame + # callers (font-ladder walks) resolve the same (name, size) over and + # over, and the BDF strike size means re-reading a file header. + # Both are released in cleanup(). + self._font_cache: Dict[Tuple[str, int], Any] = {} + self._bdf_native_size_cache: Dict[str, Optional[int]] = {} + # Set up headers self.headers = { 'User-Agent': 'LEDMatrix/1.0 (https://github.com/yourusername/LEDMatrix; contact@example.com)', @@ -172,8 +250,7 @@ class SportsCore(ABC): """Convert relative paths to absolute ones rooted at the project directory.""" if path.is_absolute(): return path - project_root = Path(__file__).resolve().parents[2] - return (project_root / path).resolve() + return (_INSTALL_ROOT / path).resolve() def _get_logo_directory_fallbacks(self, configured_dir: Path) -> List[Path]: """Return fallback directories to try when the configured directory is not writable.""" @@ -753,3 +830,261 @@ class SportsCore(ABC): def _custom_scorebug_layout(self, game: dict, draw_overlay: ImageDraw.ImageDraw): pass + + # ------------------------------------------------------------------ + # Promoted from the plugin copies (see docs/SPORTS_UNIFICATION.md). + # Everything below is present in all nine bundled `sports.py` copies; + # the canonical form of each is documented on the method. + # ------------------------------------------------------------------ + + def _favorite_key(self, game: Dict, side: str) -> Optional[str]: + """Override point: which view-model field identifies a team when + matching against ``favorite_teams``. + + ``side`` is ``"home"`` or ``"away"``. The default is the team + abbreviation — what eight of the nine scoreboards match on, and what + users type into their favorites list. + + NRL overrides this to the team **id**, because NRL abbreviations are + not unique: "NEW" is both Newcastle Knights and New Zealand Warriors, + "CAN" both Canberra Raiders and Canterbury Bulldogs. Matching those by + abbreviation selects the wrong club. It is a seam rather than a branch + precisely so core never has to learn the string "nrl":: + + def _favorite_key(self, game, side): + return str(game.get(f"{side}_id")) + + An override that stringifies should note that a missing id becomes the + literal ``"None"``, which would spuriously match a favorites list + containing that string. The default returns ``None``, which never + matches. + """ + return game.get(f"{side}_abbr") + + def _config_schema_path(self) -> Optional[str]: + """Override point: the plugin's ``config_schema.json``, used as the + reference for style-resolver defaults. + + None (the default) means no schema is available — layout offsets are + then read with the classic inline config lookup, which is exactly what + a plugin that never shipped resolver support does today. A plugin + opting into :mod:`src.element_style` returns its own schema path:: + + def _config_schema_path(self): + return os.path.join(os.path.dirname(os.path.abspath(__file__)), + 'config_schema.json') + + It must not be derived from this module's ``__file__``: after + promotion that resolves inside ``src/base_classes/sports/``, where no + ``config_schema.json`` exists. + """ + return None + + def _font_root(self) -> str: + """Override point: the directory ``assets/fonts`` resolves against. + + Defaults to the core install root, derived from this module's own + location — the same strategy as ``FontManager._resolve_asset_path``. + A plugin bundling its own fonts overrides this to return its plugin + directory. Never a cwd-relative path: fonts must load no matter where + the process was started from (e.g. the plugin safety harness on CI). + """ + return str(_INSTALL_ROOT) + + def _resolve_font_path(self, font_name: str) -> str: + """Locate a font file by filename, independently of the process cwd. + + Tries ``assets/fonts/`` relative to the cwd first (preserving + behavior for a process started from an install root), then under + :meth:`_font_root`. Returns the cwd-relative path unchanged when the + file is nowhere to be found, so callers log the familiar path. + """ + relative = os.path.join('assets', 'fonts', font_name) + if os.path.exists(relative): + return relative + candidate = os.path.join(self._font_root(), relative) + if os.path.exists(candidate): + return candidate + return relative + + def _get_layout_offset(self, element: str, axis: str, default: int = 0) -> int: + """ + Get layout offset for a specific element and axis. + + Args: + element: Element name (e.g., 'home_logo', 'score', 'status_text') + axis: 'x_offset' or 'y_offset' (or 'away_x_offset', 'home_x_offset' for records) + default: Default value if not configured (default: 0) + + Returns: + Offset value from config or default (always returns int) + """ + schema_path = self._config_schema_path() if STYLE_AVAILABLE else None + if schema_path: + # Shared resolver (rebuilt if the config dict was swapped out, + # matching the classic path's read-config-on-every-call semantics). + # Note it is stricter than the classic read below: a boolean offset + # degrades to the default instead of counting as 1/0, which is the + # more correct reading of a pixel offset. + resolver = getattr(self, '_style_resolver_cached', None) + if resolver is None or resolver._config is not self.config: + resolver = ElementStyleResolver( + self.config, defaults_from_schema_file(schema_path)) + self._style_resolver_cached = resolver + return resolver.offset_value(element, axis, default) + try: + layout_config = self.config.get('customization', {}).get('layout', {}) + element_config = layout_config.get(element, {}) + offset_value = element_config.get(axis, default) + + # Ensure we return an integer (handle float/string from config) + if isinstance(offset_value, (int, float)): + return int(offset_value) + elif isinstance(offset_value, str): + # Try to convert string to int + try: + return int(float(offset_value)) + except (ValueError, TypeError): + self.logger.warning( + f"Invalid layout offset value for {element}.{axis}: '{offset_value}', using default {default}" + ) + return default + else: + return default + except Exception as e: + # Gracefully handle any config access errors + self.logger.debug(f"Error reading layout offset for {element}.{axis}: {e}, using default {default}") + return default + + def _load_custom_font_from_element_config( + self, + element_config: Dict[str, Any], + default_size: int = 8, + default_font: Optional[str] = None, + ) -> ImageFont.FreeTypeFont: + """ + Load a custom font from an element configuration dictionary. + + Args: + element_config: Configuration dict for a single element containing 'font' and 'font_size' keys + default_size: Default font size if not specified in config + default_font: Default font filename when not specified in config (e.g. '4x6-font.ttf' for odds) + + Returns: + PIL ImageFont object + """ + base_default = default_font or "PressStart2P-Regular.ttf" + font_name = element_config.get('font', base_default) + font_size = int(element_config.get('font_size', default_size)) # Ensure integer for PIL + + # Resolve family aliases (e.g. "press_start") to real filenames, then + # locate the file against _font_root() rather than the cwd. + resolved_name = _resolve_font_family_alias(font_name) + font_path = self._resolve_font_path(resolved_name) + + # Memoized: per-frame callers (font-ladder walks) resolve the same + # (name, size) repeatedly -- return the previously loaded face. + cache_key = (resolved_name, font_size) + cached_font = self._font_cache.get(cache_key) + if cached_font is not None: + return cached_font + + # Try to load the font + try: + if os.path.exists(font_path): + # Try loading as TTF first (works for both TTF and some BDF files with PIL) + if font_path.lower().endswith('.ttf'): + font = ImageFont.truetype(font_path, font_size) + self.logger.debug(f"Loaded font: {font_name} at size {font_size}") + self._font_cache[cache_key] = font + return font + elif font_path.lower().endswith('.bdf'): + # BDF fonts are fixed-size bitmaps, not scalable outlines -- + # FreeType only accepts the exact pixel size baked into the + # file (its "strike") and raises "invalid pixel size" for + # anything else. Try the requested size first (in case it + # happens to match), then fall back to the file's real + # native size, so a BDF font can still be selected via + # font_size-driven configs without the caller needing to + # know its exact strike size. + # + # This retry is the OLDER lineage's behavior and it is the + # correct one: the newer copies call truetype() on a BDF at + # any size (which simply fails) or refuse BDF outright. + try: + font = ImageFont.truetype(font_path, font_size) + self.logger.debug(f"Loaded BDF font: {font_name} at size {font_size}") + self._font_cache[cache_key] = font + return font + except OSError: + if font_path in self._bdf_native_size_cache: + native_size = self._bdf_native_size_cache[font_path] + else: + native_size = _read_bdf_native_size(font_path) + self._bdf_native_size_cache[font_path] = native_size + if native_size and native_size != font_size: + try: + font = ImageFont.truetype(font_path, native_size) + self.logger.debug( + f"Loaded BDF font: {font_name} at its native size {native_size} " + f"(requested {font_size} isn't a valid strike for this file)" + ) + self._font_cache[cache_key] = font + return font + except Exception as retry_exc: + self.logger.debug( + f"BDF font {font_name} also failed to load at native " + f"size {native_size}: {retry_exc}" + ) + self.logger.warning(f"Could not load BDF font {font_name} with PIL, using default") + # Fall through to default + else: + self.logger.warning(f"Unknown font file type: {font_name}, using default") + else: + self.logger.warning(f"Font file not found: {font_path}, using default") + except Exception as e: + self.logger.error(f"Error loading font {font_name}: {e}, using default") + + # Fall back to default font. Cached under the requested (name, size) + # key too, so a misconfigured or missing font pays the disk cost once + # instead of on every frame of a font-ladder walk. + default_font_path = self._resolve_font_path( + _resolve_font_family_alias(base_default)) + try: + if os.path.exists(default_font_path): + font = ImageFont.truetype(default_font_path, font_size) + else: + self.logger.warning("Default font not found, using PIL default") + font = ImageFont.load_default() + except Exception as e: + self.logger.error(f"Error loading default font: {e}") + font = ImageFont.load_default() + self._font_cache[cache_key] = font + return font + + def cleanup(self): + """Clean up resources when plugin is unloaded.""" + # Close HTTP session + if hasattr(self, 'session') and self.session: + try: + self.session.close() + except Exception as e: + self.logger.warning(f"Error closing session: {e}") + + # Clear caches + if hasattr(self, '_logo_cache'): + self._logo_cache.clear() + # Font caches hold PIL faces; without this they are an unbounded + # per-instance leak across enable/disable cycles. + if hasattr(self, '_font_cache'): + self._font_cache.clear() + if hasattr(self, '_bdf_native_size_cache'): + self._bdf_native_size_cache.clear() + + # NOTE: self.background_service is deliberately NOT shut down here. + # get_background_service() returns a PROCESS-WIDE singleton shared by + # every scoreboard; shutting it down from one unloading plugin would + # stop background fetching for all the others. Whoever owns the + # process owns its lifecycle. Do not "fix" this. + + self.logger.info(f"{self.__class__.__name__} cleanup completed") diff --git a/src/base_classes/sports/modes.py b/src/base_classes/sports/modes.py index 0c025da4..0c9dd695 100644 --- a/src/base_classes/sports/modes.py +++ b/src/base_classes/sports/modes.py @@ -7,7 +7,7 @@ import logging import time from abc import abstractmethod from datetime import datetime, timedelta, timezone -from typing import Any, Dict +from typing import Any, Dict, List from PIL import Image, ImageDraw, ImageFont @@ -34,6 +34,71 @@ class SportsUpcoming(SportsCore): self.last_game_switch = 0 self.game_display_duration = 15 # Display each upcoming game for 15 seconds + def _select_games_for_display( + self, processed_games: List[Dict], favorite_teams: List[str] + ) -> List[Dict]: + """ + Single-pass game selection with proper deduplication and counting. + + When a game involves two favorite teams, it counts toward BOTH teams' limits. + This prevents unexpected game counts from the multi-pass algorithm. + + Team identity goes through the ``_favorite_key`` override point rather + than reading ``home_abbr``/``away_abbr`` directly, because abbreviations + are not unique in every league (NRL matches on team ID instead). + """ + sorted_games = sorted( + processed_games, + key=lambda g: g.get("start_time_utc") + or datetime.max.replace(tzinfo=timezone.utc), + ) + + if not favorite_teams: + return sorted_games + + selected_games = [] + selected_ids = set() + team_counts = {team: 0 for team in favorite_teams} + + for game in sorted_games: + game_id = game.get("id") + if game_id in selected_ids: + continue + + home = self._favorite_key(game, "home") + away = self._favorite_key(game, "away") + + home_fav = home in favorite_teams + away_fav = away in favorite_teams + + if not home_fav and not away_fav: + continue + + home_needs = home_fav and team_counts[home] < self.upcoming_games_to_show + away_needs = away_fav and team_counts[away] < self.upcoming_games_to_show + + if home_needs or away_needs: + selected_games.append(game) + selected_ids.add(game_id) + if home_fav: + team_counts[home] += 1 + if away_fav: + team_counts[away] += 1 + + self.logger.debug( + f"Selected game {away}@{home}: team_counts={team_counts}" + ) + + if all(c >= self.upcoming_games_to_show for c in team_counts.values()): + self.logger.debug("All favorite teams satisfied, stopping selection") + break + + self.logger.info( + f"Selected {len(selected_games)} games for {len(favorite_teams)} " + f"favorite teams: {team_counts}" + ) + return selected_games + def update(self): """Update upcoming games data.""" if not self.is_enabled: return @@ -369,6 +434,96 @@ class SportsRecent(SportsCore): self.update_interval = self.mode_config.get("recent_update_interval", 3600) # Check for recent games every hour self.last_game_switch = 0 self.game_display_duration = 15 # Display each recent game for 15 seconds + # Tracks when each game was first seen with an expired clock, keyed by + # game id. Promoted alongside the zero-clock helpers below; without it + # the first _get_zero_clock_duration() call raises AttributeError. + self._zero_clock_timestamps: Dict[str, float] = {} # Track games at 0:00 + + # -- Zero-clock tracking ------------------------------------------------ + # Byte-identical in all nine plugin copies. Note that afl/nrl/soccer define + # these but never call them — their clocks count up, so 0:00 means kickoff + # rather than expiry (see CLOCK_COUNTS_DOWN on SportsLive). That makes the + # pair a future `CountdownClockMixin` candidate so it stops appearing in the + # MRO of sports that cannot use it — B2 work, not now. + + def _get_zero_clock_duration(self, game_id: str) -> float: + """Track how long a game has been at 0:00 clock.""" + current_time = time.time() + if game_id not in self._zero_clock_timestamps: + self._zero_clock_timestamps[game_id] = current_time + return 0.0 + return current_time - self._zero_clock_timestamps[game_id] + + def _clear_zero_clock_tracking(self, game_id: str) -> None: + """Clear tracking when game clock moves away from 0:00 or game ends.""" + if game_id in self._zero_clock_timestamps: + del self._zero_clock_timestamps[game_id] + + def _select_recent_games_for_display( + self, processed_games: List[Dict], favorite_teams: List[str] + ) -> List[Dict]: + """ + Single-pass game selection for recent games with proper deduplication. + + When a game involves two favorite teams, it counts toward BOTH teams' limits. + Games are sorted by most recent first. + + Team identity goes through the ``_favorite_key`` override point rather + than reading ``home_abbr``/``away_abbr`` directly, because abbreviations + are not unique in every league (NRL matches on team ID instead). + """ + sorted_games = sorted( + processed_games, + key=lambda g: g.get("start_time_utc") + or datetime.min.replace(tzinfo=timezone.utc), + reverse=True, + ) + + if not favorite_teams: + return sorted_games + + selected_games = [] + selected_ids = set() + team_counts = {team: 0 for team in favorite_teams} + + for game in sorted_games: + game_id = game.get("id") + if game_id in selected_ids: + continue + + home = self._favorite_key(game, "home") + away = self._favorite_key(game, "away") + + home_fav = home in favorite_teams + away_fav = away in favorite_teams + + if not home_fav and not away_fav: + continue + + home_needs = home_fav and team_counts[home] < self.recent_games_to_show + away_needs = away_fav and team_counts[away] < self.recent_games_to_show + + if home_needs or away_needs: + selected_games.append(game) + selected_ids.add(game_id) + if home_fav: + team_counts[home] += 1 + if away_fav: + team_counts[away] += 1 + + self.logger.debug( + f"Selected recent game {away}@{home}: team_counts={team_counts}" + ) + + if all(c >= self.recent_games_to_show for c in team_counts.values()): + self.logger.debug("All favorite teams satisfied, stopping selection") + break + + self.logger.info( + f"Selected {len(selected_games)} recent games for {len(favorite_teams)} " + f"favorite teams: {team_counts}" + ) + return selected_games def update(self): """Update recent games data.""" @@ -659,6 +814,17 @@ class SportsRecent(SportsCore): return False class SportsLive(SportsCore): + # Per-sport constants for the "is this live game actually over?" check. + # These are values, not behavior, so they are class attributes rather than + # override points (see docs/SPORTS_UNIFICATION.md "Override points"). + # + # FINAL_PERIOD: the period at/after which an expired clock can mean "over". + # 4 for four-quarter sports; hockey overrides to 3. + # CLOCK_COUNTS_DOWN: whether "0:00" means the clock expired. False for + # sports whose clock counts up (soccer/afl/nrl), where 0:00 is kickoff — + # running the expiry branch there would evict games that just started. + FINAL_PERIOD = 4 + CLOCK_COUNTS_DOWN = True def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str): super().__init__(config, display_manager, cache_manager, logger, sport_key) @@ -676,11 +842,107 @@ class SportsLive(SportsCore): self.count_log_interval = 5 # Only log count data every 5 seconds # Initialize test_mode - defaults to False (live mode) self.test_mode = self.mode_config.get("test_mode", False) + # Freshness bookkeeping for _detect_stale_games(): + # {game_id: {"clock": ts, "score": ts, "last_seen": ts}} + self.game_update_timestamps = {} + self.stale_game_timeout = self.mode_config.get("stale_game_timeout", 300) # 5 minutes default @abstractmethod def _test_mode_update(self) -> None: return + def _is_game_really_over(self, game: Dict) -> bool: + """Check if a game appears to be over even if API says it's live. + + Two independent signals: + 1. ``period_text`` says "final" — universal across every sport. + 2. The clock has expired at/after :attr:`FINAL_PERIOD` — only meaningful + where :attr:`CLOCK_COUNTS_DOWN` is true. + + Fails *safe*: anything ambiguous returns False and the game keeps being + displayed. The only caller, :meth:`_detect_stale_games`, removes games + on a True, so a false positive silently drops a live game. + """ + game_str = f"{game.get('away_abbr')}@{game.get('home_abbr')}" + + # `period_text` may be present-but-None; `or ""` keeps that from raising + # AttributeError — the caller has no try/except around this call. + period_text = (game.get("period_text") or "").lower() + if "final" in period_text: + self.logger.debug( + f"_is_game_really_over({game_str}): " + f"returning True - 'final' in period_text='{period_text}'" + ) + return True + + if not self.CLOCK_COUNTS_DOWN: + # Count-up clock: 0:00 means the match has not started. + self.logger.debug( + f"_is_game_really_over({game_str}): returning False " + f"(count-up clock, period_text='{period_text}')" + ) + return False + + raw_clock = game.get("clock") + period = game.get("period", 0) + + # Only check clock-based finish if we have a valid clock string. A + # missing or non-string clock is NOT coerced to "0:00": sports without a + # game clock (e.g. baseball, where `period` is the inning) would + # otherwise be declared over from the FINAL_PERIOD-th period onward. + if isinstance(raw_clock, str) and raw_clock.strip() and period >= self.FINAL_PERIOD: + clock = raw_clock + clock_normalized = clock.replace(":", "").strip() + if clock_normalized in ("000", "00") or clock in ("0:00", ":00"): + self.logger.debug( + f"_is_game_really_over({game_str}): " + f"returning True - clock at 0:00 (clock='{clock}', period={period})" + ) + return True + + self.logger.debug( + f"_is_game_really_over({game_str}): returning False" + ) + return False + + def _detect_stale_games(self, games: List[Dict]) -> None: + """Remove games that appear stale or haven't updated. + + Mutates ``games`` **in place** and returns None. Removal is by value + (``list.remove`` uses ``dict.__eq__``), so two structurally-equal game + dicts in the same list would drop the first occurrence. + """ + current_time = time.time() + + for game in games[:]: # Copy list to iterate safely + game_id = game.get("id") + if not game_id: + continue + + # Check if game data is stale + timestamps = self.game_update_timestamps.get(game_id, {}) + last_seen = timestamps.get("last_seen", 0) + + if last_seen > 0 and current_time - last_seen > self.stale_game_timeout: + self.logger.warning( + f"Removing stale game {game.get('away_abbr')}@{game.get('home_abbr')} " + f"(last seen {int(current_time - last_seen)}s ago)" + ) + games.remove(game) + if game_id in self.game_update_timestamps: + del self.game_update_timestamps[game_id] + continue + + # Also check if game appears to be over + if self._is_game_really_over(game): + self.logger.debug( + f"Removing game that appears over: {game.get('away_abbr')}@{game.get('home_abbr')} " + f"(clock={game.get('clock')}, period={game.get('period')}, period_text={game.get('period_text')})" + ) + games.remove(game) + if game_id in self.game_update_timestamps: + del self.game_update_timestamps[game_id] + def update(self): """Update live game data and handle game switching.""" if not self.is_enabled: diff --git a/test/test_sports_core_promotions.py b/test/test_sports_core_promotions.py new file mode 100644 index 00000000..6826cd80 --- /dev/null +++ b/test/test_sports_core_promotions.py @@ -0,0 +1,558 @@ +"""Tests for the methods promoted onto SportsCore from the nine bundled +plugin copies of ``sports.py`` (phase B1 of docs/SPORTS_UNIFICATION.md). + +Three methods and their seams land here: + +- ``cleanup()`` — byte-identical in all nine copies. The tests pin the + ordering (session close, then caches, then the completion log) and the + deliberate *omission*: the process-wide background service must never be + shut down by one unloading plugin. +- ``_get_layout_offset()`` — football's resolver-backed variant, with the + classic inline config read as the fallback used by every plugin that + doesn't hand core a ``_config_schema_path()``. +- ``_load_custom_font_from_element_config()`` — baseball's body (the only + copy that handles BDF strikes correctly) under hockey's wider signature, + resolving font files through the ``_font_root()`` seam instead of the + process cwd. +""" + +import ast +import json +import logging +import os +import sys +from pathlib import Path +from unittest.mock import MagicMock + +import pytest +from PIL import Image, ImageFont + +# src.base_classes.sports transitively imports the hardware matrix driver; +# stub it so these tests can import the sports base classes off-device. +sys.modules.setdefault("rgbmatrix", MagicMock()) + +from src.base_classes.sports import SportsCore + +LOGGER = logging.getLogger("test_sports_core_promotions") + +CORE_ROOT = Path(__file__).resolve().parents[1] +FONTS_DIR = CORE_ROOT / "assets" / "fonts" +TTF_NAME = "PressStart2P-Regular.ttf" +BDF_NAME = "5x7.bdf" # a BDF whose only valid strike is 7px +BDF_NATIVE_SIZE = 7 + + +class _StubSports(SportsCore): + """Minimal concrete SportsCore — the abstract methods are never called + by anything under test here.""" + + def _fetch_data(self): + return None + + def _extract_game_details(self, game_event): + return None + + +@pytest.fixture +def build(monkeypatch, tmp_path): + """Factory for real SportsCore instances: logo dir redirected to tmp and + the process-wide background service replaced with a MagicMock so the + tests can assert nothing ever calls it.""" + monkeypatch.setattr( + SportsCore, "_initialize_logo_dir", lambda self, configured: tmp_path) + monkeypatch.setattr( + "src.base_classes.sports.core.get_background_service", + lambda *args, **kwargs: MagicMock()) + + def _build(config=None, cls=_StubSports): + display_manager = MagicMock() + display_manager.matrix.width = 128 + display_manager.matrix.height = 32 + display_manager.width = 128 + display_manager.height = 32 + display_manager.image = Image.new("RGB", (128, 32)) + cache_manager = MagicMock() + cache_manager.cache_dir = str(tmp_path) + return cls(config if config is not None else {"timezone": "UTC"}, + display_manager, cache_manager, LOGGER, "nhl") + + return _build + + +def probe(config=None): + """Unbound-call stand-in for hosts we don't need a full instance for + (same pattern as make_probe in test_sports_base_characterization).""" + host = MagicMock() + host.logger = LOGGER + host.config = config if config is not None else {} + host._font_cache = {} + host._bdf_native_size_cache = {} + host._config_schema_path.return_value = None + host._font_root.side_effect = lambda: SportsCore._font_root(host) + host._resolve_font_path.side_effect = ( + lambda name: SportsCore._resolve_font_path(host, name)) + return host + + +def offset(host, element, axis, default=0): + return SportsCore._get_layout_offset(host, element, axis, default) + + +def load_font(host, *args, **kwargs): + return SportsCore._load_custom_font_from_element_config(host, *args, **kwargs) + + +# --------------------------------------------------------------------------- +# 1. cleanup() +# --------------------------------------------------------------------------- + +class TestCleanup: + def test_closes_session_and_clears_all_caches(self, build): + manager = build() + session = MagicMock() + manager.session = session + manager._logo_cache["TB"] = object() + manager._font_cache[("PressStart2P-Regular.ttf", 8)] = object() + manager._bdf_native_size_cache["assets/fonts/5x7.bdf"] = 7 + + manager.cleanup() + + session.close.assert_called_once_with() + assert manager._logo_cache == {} + # Promoted alongside the font loader: these hold PIL faces and are + # an unbounded leak across enable/disable cycles if never released. + assert manager._font_cache == {} + assert manager._bdf_native_size_cache == {} + + def test_second_cleanup_is_a_noop(self, build): + manager = build() + manager.session = MagicMock() + manager._logo_cache["TB"] = object() + manager._font_cache[("x", 8)] = object() + + manager.cleanup() + manager.cleanup() # must not raise on already-released state + + assert manager._logo_cache == {} + assert manager._font_cache == {} + assert manager._bdf_native_size_cache == {} + assert manager.session.close.call_count == 2 + + def test_does_not_shut_down_the_shared_background_service(self, build): + # get_background_service() hands out a PROCESS-WIDE singleton shared + # by every scoreboard. One plugin unloading must not stop background + # fetching for the other eight — cleanup() touches it not at all. + manager = build() + service = manager.background_service + manager.session = MagicMock() + + manager.cleanup() + + assert service.shutdown.called is False + assert service.stop.called is False + assert service.method_calls == [], ( + "cleanup() called into the shared background service: " + f"{service.method_calls}") + + def test_completion_is_logged_even_when_session_close_raises(self, build, caplog): + manager = build() + manager.session = MagicMock() + manager.session.close.side_effect = RuntimeError("socket already gone") + manager._logo_cache["TB"] = object() + + with caplog.at_level(logging.DEBUG, logger=LOGGER.name): + manager.cleanup() + + messages = [r.message for r in caplog.records] + assert any("Error closing session" in m for m in messages) + # Ordering is load-bearing: the caches still get cleared and the + # completion log still fires after a failed close. + assert manager._logo_cache == {} + assert any("cleanup completed" in m for m in messages) + + def test_tolerates_missing_attributes(self): + # The hasattr guards exist so a partially constructed instance (an + # __init__ that raised) can still be cleaned up. + host = MagicMock(spec=["logger"]) + host.logger = LOGGER + SportsCore.cleanup(host) + + +# --------------------------------------------------------------------------- +# 2. _get_layout_offset() + the _config_schema_path() seam +# --------------------------------------------------------------------------- + +def layout_config(element, axis, value): + return {"customization": {"layout": {element: {axis: value}}}} + + +class TestLayoutOffsetClassicPath: + """The default path: _config_schema_path() returns None, so offsets come + from the inline customization.layout read every plugin ships today.""" + + def test_config_schema_path_defaults_to_none(self, build): + manager = build() + assert manager._config_schema_path() is None + + def test_reads_configured_int(self): + host = probe(layout_config("home_logo", "x_offset", 5)) + assert offset(host, "home_logo", "x_offset") == 5 + + def test_float_is_truncated_to_int(self): + host = probe(layout_config("score", "y_offset", 2.9)) + result = offset(host, "score", "y_offset") + assert result == 2 and isinstance(result, int) + + def test_numeric_string_is_coerced(self): + host = probe(layout_config("score", "x_offset", "-3.5")) + assert offset(host, "score", "x_offset") == -3 + + def test_unconfigured_element_and_axis_use_default(self): + host = probe(layout_config("score", "x_offset", 5)) + assert offset(host, "status_text", "x_offset", 7) == 7 + assert offset(host, "score", "y_offset", -1) == -1 + assert offset(probe(), "score", "x_offset", 4) == 4 + + def test_non_numeric_string_degrades_to_default(self): + host = probe(layout_config("score", "x_offset", "left")) + assert offset(host, "score", "x_offset", 3) == 3 + + def test_unsupported_type_degrades_to_default(self): + host = probe(layout_config("score", "x_offset", {"nested": 1})) + assert offset(host, "score", "x_offset", 2) == 2 + host = probe(layout_config("score", "x_offset", None)) + assert offset(host, "score", "x_offset", 2) == 2 + + def test_broken_config_object_degrades_to_default(self): + host = probe() + host.config = "not a dict" + assert offset(host, "score", "x_offset", 6) == 6 + + def test_boolean_counts_as_one(self): + # PINNED AS-IS: the classic read predates the resolver and treats a + # bool as its int value (True -> 1). See the resolver test below for + # the stricter, more correct handling. + host = probe(layout_config("score", "x_offset", True)) + assert offset(host, "score", "x_offset", 4) == 1 + + +class TestLayoutOffsetResolverPath: + """When a plugin supplies its config_schema.json, offsets resolve through + src.element_style instead.""" + + @pytest.fixture + def schema_path(self, tmp_path): + path = tmp_path / "config_schema.json" + path.write_text(json.dumps({ + "type": "object", + "properties": { + "customization": { + "type": "object", + "properties": { + "layout": {"type": "object", "properties": {}}, + }, + }, + }, + })) + return str(path) + + def host(self, schema_path, config): + host = probe(config) + host._config_schema_path.return_value = schema_path + del host._style_resolver_cached # MagicMock auto-attrs otherwise + host._style_resolver_cached = None + return host + + def test_reads_configured_offsets(self, schema_path): + host = self.host(schema_path, layout_config("home_logo", "x_offset", 5)) + assert offset(host, "home_logo", "x_offset") == 5 + + def test_numeric_string_is_coerced(self, schema_path): + host = self.host(schema_path, layout_config("score", "x_offset", "-3.5")) + assert offset(host, "score", "x_offset") == -3 + + def test_missing_value_uses_default(self, schema_path): + host = self.host(schema_path, layout_config("score", "x_offset", 5)) + assert offset(host, "score", "y_offset", 9) == 9 + + def test_bad_input_degrades_to_default(self, schema_path): + host = self.host(schema_path, layout_config("score", "x_offset", "left")) + assert offset(host, "score", "x_offset", 3) == 3 + host = self.host(schema_path, {"customization": {"layout": "nope"}}) + assert offset(host, "score", "x_offset", 3) == 3 + + def test_boolean_is_rejected_unlike_the_classic_path(self, schema_path): + # The intended behavior difference: a bool is not a pixel offset, so + # the resolver returns the default where the classic read returns 1. + host = self.host(schema_path, layout_config("score", "x_offset", True)) + assert offset(host, "score", "x_offset", 4) == 4 + + def test_resolver_is_cached_and_rebuilt_when_config_is_swapped(self, schema_path): + host = self.host(schema_path, layout_config("score", "x_offset", 5)) + assert offset(host, "score", "x_offset") == 5 + first = host._style_resolver_cached + assert offset(host, "score", "x_offset") == 5 + assert host._style_resolver_cached is first + + # on_config_change swaps the dict object; the resolver must follow. + host.config = layout_config("score", "x_offset", 11) + assert offset(host, "score", "x_offset") == 11 + assert host._style_resolver_cached is not first + + def test_missing_schema_file_still_resolves_offsets(self, tmp_path): + # Offsets don't depend on schema defaults, so an unreadable schema + # must not cost the plugin its layout customization. + host = self.host(str(tmp_path / "absent.json"), + layout_config("score", "x_offset", 5)) + assert offset(host, "score", "x_offset") == 5 + + +# --------------------------------------------------------------------------- +# 3. _load_custom_font_from_element_config() + the _font_root() seam +# --------------------------------------------------------------------------- + +class TestFontRootSeam: + def test_default_font_root_is_the_core_install_root(self, build): + manager = build() + assert Path(manager._font_root()) == CORE_ROOT + assert (Path(manager._font_root()) / "assets" / "fonts").is_dir() + + def test_resolve_font_path_honors_an_overridden_root(self, tmp_path): + fonts = tmp_path / "assets" / "fonts" + fonts.mkdir(parents=True) + (fonts / "Bundled.ttf").write_bytes(b"not really a font") + host = probe() + host._font_root.side_effect = lambda: str(tmp_path) + assert SportsCore._resolve_font_path(host, "Bundled.ttf") == str( + fonts / "Bundled.ttf") + + def test_unknown_font_returns_the_familiar_relative_path(self): + host = probe() + assert SportsCore._resolve_font_path(host, "Nope.ttf") == os.path.join( + "assets", "fonts", "Nope.ttf") + + +class TestFontLoaderSignature: + """Hockey's signature is the only safe superset: basketball's positional + ``default_font: str`` blows up on an explicit None.""" + + def test_two_arg_call(self): + font = load_font(probe(), {"font": TTF_NAME, "font_size": 10}) + assert isinstance(font, ImageFont.FreeTypeFont) + assert font.size == 10 + + def test_default_size_is_used_when_config_omits_it(self): + assert load_font(probe(), {}, 12).size == 12 + + def test_three_positional_args(self): + font = load_font(probe(), {}, 6, "4x6-font.ttf") + assert isinstance(font, ImageFont.FreeTypeFont) + assert font.size == 6 + assert font.path.endswith("4x6-font.ttf") + + def test_explicit_default_font_none(self): + # The regression this signature guards: os.path.join(..., None). + font = load_font(probe(), {"font_size": 9}, default_font=None) + assert isinstance(font, ImageFont.FreeTypeFont) + assert font.path.endswith(TTF_NAME) + + def test_config_font_wins_over_default_font(self): + font = load_font(probe(), {"font": TTF_NAME}, 8, "4x6-font.ttf") + assert font.path.endswith(TTF_NAME) + + def test_string_font_size_is_coerced(self): + assert load_font(probe(), {"font": TTF_NAME, "font_size": "11"}).size == 11 + + +class TestFontLoaderBehavior: + def test_family_alias_resolves_through_the_font_manager_catalog(self): + # "press_start" is a FontManager catalog family, not a filename; the + # promoted loader must not carry its own duplicate alias table. + host = probe() + font = load_font(host, {"font": "press_start", "font_size": 8}) + assert font.path.endswith(TTF_NAME) + assert ("PressStart2P-Regular.ttf", 8) in host._font_cache + + def test_memo_cache_returns_the_same_face(self): + host = probe() + first = load_font(host, {"font": TTF_NAME, "font_size": 8}) + second = load_font(host, {"font": TTF_NAME, "font_size": 8}) + assert first is second + assert len(host._font_cache) == 1 + # A different size is a different face. + assert load_font(host, {"font": TTF_NAME, "font_size": 9}) is not first + assert len(host._font_cache) == 2 + + def test_bdf_loads_at_its_native_strike_when_the_request_misses(self): + # BDF is a fixed-size bitmap format: FreeType raises "invalid pixel + # size" for anything but the file's own strike. Baseball's retry is + # the only copy that gets this right. + host = probe() + font = load_font(host, {"font": BDF_NAME, "font_size": 8}) + assert isinstance(font, ImageFont.FreeTypeFont) + assert font.size == BDF_NATIVE_SIZE + assert font.path.endswith(BDF_NAME) + assert set(host._bdf_native_size_cache.values()) == {BDF_NATIVE_SIZE} + # The retried face is memoized under the REQUESTED size. + assert host._font_cache[(BDF_NAME, 8)] is font + + def test_bdf_at_its_native_size_needs_no_retry(self): + host = probe() + font = load_font(host, {"font": BDF_NAME, "font_size": BDF_NATIVE_SIZE}) + assert font.size == BDF_NATIVE_SIZE + assert host._bdf_native_size_cache == {} + + def test_bdf_strike_lookup_is_memoized(self, monkeypatch): + calls = [] + real = SportsCore.__module__ + + def counting(path): + calls.append(path) + from src.font_manager import FontManager + return FontManager._read_bdf_native_size(path) + + monkeypatch.setattr(f"{real}._read_bdf_native_size", counting) + host = probe() + load_font(host, {"font": BDF_NAME, "font_size": 8}) + host._font_cache.clear() # force the load path again + load_font(host, {"font": BDF_NAME, "font_size": 8}) + assert len(calls) == 1 + + def test_missing_font_falls_back_and_caches_the_fallback(self, caplog): + host = probe() + with caplog.at_level(logging.WARNING, logger=LOGGER.name): + font = load_font(host, {"font": "DoesNotExist.ttf", "font_size": 8}) + assert isinstance(font, ImageFont.FreeTypeFont) + assert font.path.endswith(TTF_NAME) + assert any("Font file not found" in r.message for r in caplog.records) + # Cached under the requested name so a misconfiguration costs one + # disk probe, not one per frame. + assert host._font_cache[("DoesNotExist.ttf", 8)] is font + + def test_unknown_extension_falls_back(self): + host = probe() + font = load_font(host, {"font": "AUTHORS", "font_size": 8}) + assert font.path.endswith(TTF_NAME) + + def test_fallback_honors_the_supplied_default_font(self): + host = probe() + font = load_font(host, {"font": "DoesNotExist.ttf"}, 6, "4x6-font.ttf") + assert font.path.endswith("4x6-font.ttf") + + +class TestFontLoaderCwdIndependence: + """The bug the _font_root() seam exists to prevent: every plugin copy + joins 'assets/fonts' onto the process cwd, so a process started anywhere + else silently degrades to PIL's default bitmap face (the same defect + already fixed in FontManager — see CHANGELOG Unreleased/Fixed).""" + + @pytest.mark.parametrize("font_name,expected_size", + [(TTF_NAME, 8), (BDF_NAME, BDF_NATIVE_SIZE)]) + def test_fonts_load_from_an_unrelated_cwd(self, monkeypatch, font_name, + expected_size): + monkeypatch.chdir("/") + host = probe() + font = load_font(host, {"font": font_name, "font_size": 8}) + assert isinstance(font, ImageFont.FreeTypeFont), ( + f"{font_name} degraded to PIL's default face when the process " + "runs outside the install root") + assert font.size == expected_size + assert Path(font.path) == FONTS_DIR / font_name + + def test_fallback_font_also_survives_an_unrelated_cwd(self, monkeypatch): + monkeypatch.chdir("/") + font = load_font(probe(), {"font": "DoesNotExist.ttf", "font_size": 8}) + assert isinstance(font, ImageFont.FreeTypeFont) + assert Path(font.path) == FONTS_DIR / TTF_NAME + + +# --------------------------------------------------------------------------- +# 4. Seam guard rails +# --------------------------------------------------------------------------- + +class TestPromotedSeamsExist: + @pytest.mark.parametrize("name", [ + "cleanup", "_get_layout_offset", "_load_custom_font_from_element_config", + "_config_schema_path", "_font_root", "_resolve_font_path", + ]) + def test_method_is_callable_on_the_base_class(self, name): + assert callable(getattr(SportsCore, name, None)), ( + f"SportsCore.{name} is part of the promoted plugin-facing seam " + "(docs/SPORTS_UNIFICATION.md) — plugins probe for it with " + "hasattr before delegating.") + + def test_no_sport_names_leaked_into_core(self): + """core.py must never branch on which sport it is (prose and skin-id + examples in docstrings are fine — executable code is not).""" + tree = ast.parse((CORE_ROOT / "src" / "base_classes" / "sports" + / "core.py").read_text()) + docstrings = set() + for node in ast.walk(tree): + if isinstance(node, (ast.Module, ast.ClassDef, ast.FunctionDef, + ast.AsyncFunctionDef)): + first = node.body[0] if node.body else None + if (isinstance(first, ast.Expr) + and isinstance(first.value, ast.Constant) + and isinstance(first.value.value, str)): + docstrings.add(id(first.value)) + + tokens = [] + for node in ast.walk(tree): + if isinstance(node, ast.Constant) and isinstance(node.value, str): + if id(node) not in docstrings: + tokens.append(node.value) + elif isinstance(node, ast.Name): + tokens.append(node.id) + elif isinstance(node, ast.Attribute): + tokens.append(node.attr) + elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, + ast.ClassDef)): + tokens.append(node.name) + + haystack = " ".join(tokens).lower() + for sport in ("afl", "nrl", "hockey", "baseball", "basketball", + "football", "lacrosse", "soccer", "ufc"): + assert sport not in haystack, ( + f"core.py code mentions '{sport}' — core must never learn " + "sport names; add an override point instead.") + + +class TestInstallRootResolution: + """Guards the depth bug the sports.py -> package move introduced. + + The move was byte-identical in every class body, but `__file__` gained a + directory, so `Path(__file__).resolve().parents[2]` silently changed from + the repo root to `/src`. Textual identity is not semantic identity + when code measures its own location: these tests assert the resolved + values, not the index. + """ + + def test_install_root_is_the_repo_root(self): + from src.base_classes.sports.core import _INSTALL_ROOT + + # The repo root is the directory that actually holds src/ and assets/. + assert (_INSTALL_ROOT / "src").is_dir() + assert (_INSTALL_ROOT / "src" / "base_classes" / "sports").is_dir() + assert _INSTALL_ROOT.name != "src", ( + "_INSTALL_ROOT resolved to src/ — the parents[] depth is off by " + "one, which is exactly the regression the package move caused.") + + def test_resolve_project_path_roots_at_repo_not_src(self): + from src.base_classes.sports.core import SportsCore, _INSTALL_ROOT + + resolved = SportsCore._resolve_project_path(None, Path("assets/fonts")) + assert resolved == _INSTALL_ROOT / "assets" / "fonts" + assert "src" not in resolved.relative_to(_INSTALL_ROOT).parts + + def test_absolute_paths_pass_through_unchanged(self): + from src.base_classes.sports.core import SportsCore + + absolute = Path("/tmp/some/logo/dir") + assert SportsCore._resolve_project_path(None, absolute) == absolute + + def test_font_root_and_project_path_share_one_anchor(self): + """Both consumers must derive from the same constant, so a future + move needs exactly one line changed rather than two.""" + from src.base_classes.sports.core import SportsCore, _INSTALL_ROOT + + assert SportsCore._font_root(None) == str(_INSTALL_ROOT) diff --git a/test/test_sports_modes_promotions.py b/test/test_sports_modes_promotions.py new file mode 100644 index 00000000..18f2a8cf --- /dev/null +++ b/test/test_sports_modes_promotions.py @@ -0,0 +1,565 @@ +"""Tests for the methods promoted onto SportsUpcoming / SportsRecent / +SportsLive from the nine plugin copies (phase B1 of the sports unification; +see docs/SPORTS_UNIFICATION.md). + +Covered promotions: +- SportsRecent: `_get_zero_clock_duration` / `_clear_zero_clock_tracking` + (+ the `_zero_clock_timestamps` initializer). +- SportsLive: `_is_game_really_over` / `_detect_stale_games` + (+ `game_update_timestamps` / `stale_game_timeout`, and the + `FINAL_PERIOD` / `CLOCK_COUNTS_DOWN` class attributes). +- SportsUpcoming: `_select_games_for_display`. +- SportsRecent: `_select_recent_games_for_display`. + +The live pair is the risk centre: `_detect_stale_games` is the only caller +that *removes* games, so `_is_game_really_over` returning a false positive +silently drops a live game from the display. The canonical form deliberately +declines to treat a missing clock as 0:00 — the plugin variant that did so +dropped clockless sports (baseball) from the FINAL_PERIOD-th period onward. +That regression is pinned by +`test_missing_clock_at_late_period_is_not_over`. +""" + +import logging +import sys +import time +from datetime import datetime, timezone +from unittest.mock import MagicMock + +import pytest +import requests +from freezegun import freeze_time +from PIL import Image + +# src.base_classes.sports transitively imports the hardware matrix driver; +# stub it so these tests can import the sports base classes off-device. +sys.modules.setdefault("rgbmatrix", MagicMock()) + +from src.base_classes.hockey import Hockey, HockeyLive +from src.base_classes.sports import ( + SportsCore, + SportsLive, + SportsRecent, + SportsUpcoming, +) + + +# --------------------------------------------------------------------------- +# Harnesses +# --------------------------------------------------------------------------- + +class _UpcomingHarness(Hockey, SportsUpcoming): + """Cheapest concrete SportsUpcoming: hockey extractor + cache-fed data. + + `_favorite_key` is inherited from SportsCore — these harnesses + deliberately do NOT define it, so the selection tests exercise the real + seam rather than a local stand-in. + """ + + def _fetch_data(self): + return None + + +class _RecentHarness(Hockey, SportsRecent): + def _fetch_data(self): + return None + + +class _LiveHarness(HockeyLive): + def _fetch_data(self): + return None + + +class _ThreePeriodLiveHarness(_LiveHarness): + """Hockey-shaped: regulation ends after period 3.""" + + FINAL_PERIOD = 3 + + +class _CountUpLiveHarness(_LiveHarness): + """Soccer/AFL/NRL-shaped: the clock counts up, so 0:00 is kickoff.""" + + CLOCK_COUNTS_DOWN = False + + +class _IdFavoriteUpcomingHarness(_UpcomingHarness): + """NRL-shaped: abbreviations are ambiguous, so favorites match on team id.""" + + def _favorite_key(self, game, side): + team_id = game.get(f"{side}_id") + return str(team_id) if team_id is not None else None + + +class _IdFavoriteRecentHarness(_RecentHarness): + def _favorite_key(self, game, side): + team_id = game.get(f"{side}_id") + return str(team_id) if team_id is not None else None + + +@pytest.fixture +def build_manager(monkeypatch, tmp_path): + """Factory for concrete sports managers: mocked display/cache managers, + logo dir redirected to tmp, background service stubbed, and the requests + session rigged to prove nothing hits the network.""" + monkeypatch.setattr( + SportsCore, "_initialize_logo_dir", lambda self, configured: tmp_path) + monkeypatch.setattr( + "src.base_classes.sports.core.get_background_service", + lambda *args, **kwargs: MagicMock()) + + def build(cls, **mode_cfg): + config = { + "timezone": "UTC", + "display": {}, + "nhl_scoreboard": {"enabled": True, **mode_cfg}, + } + display_manager = MagicMock() + display_manager.matrix.width = 128 + display_manager.matrix.height = 32 + display_manager.width = 128 + display_manager.height = 32 + display_manager.image = Image.new("RGB", (128, 32)) + cache_manager = MagicMock() + cache_manager.get.return_value = None + cache_manager.cache_dir = str(tmp_path) + manager = cls(config, display_manager, cache_manager, + logging.getLogger("test_sports_modes_promotions"), + "nhl") + manager.session = MagicMock() + manager.session.get.side_effect = requests.exceptions.ConnectionError( + "promotion tests are offline") + return manager + + return build + + +def game(game_id="1", home="BOS", away="TOR", start=None, home_id=None, + away_id=None, **extra): + g = { + "id": game_id, + "home_abbr": home, + "away_abbr": away, + "home_id": home_id, + "away_id": away_id, + "start_time_utc": start, + } + g.update(extra) + return g + + +def at(day, hour=12): + return datetime(2026, 1, day, hour, tzinfo=timezone.utc) + + +def _ids(games): + return [g["id"] for g in games] + + +# --------------------------------------------------------------------------- +# Tier 1 — zero-clock tracking (SportsRecent) +# --------------------------------------------------------------------------- + +class TestZeroClockTracking: + def test_initializer_present_and_empty(self, build_manager): + manager = build_manager(_RecentHarness) + assert manager._zero_clock_timestamps == {} + + def test_first_call_returns_zero_and_starts_tracking(self, build_manager): + manager = build_manager(_RecentHarness) + assert manager._get_zero_clock_duration("g1") == 0.0 + assert "g1" in manager._zero_clock_timestamps + + def test_subsequent_call_returns_elapsed_seconds(self, build_manager): + manager = build_manager(_RecentHarness) + with freeze_time("2026-01-20 12:00:00") as frozen: + assert manager._get_zero_clock_duration("g1") == 0.0 + frozen.tick(45) + assert manager._get_zero_clock_duration("g1") == pytest.approx(45.0) + frozen.tick(15) + assert manager._get_zero_clock_duration("g1") == pytest.approx(60.0) + + def test_tracking_is_per_game(self, build_manager): + manager = build_manager(_RecentHarness) + with freeze_time("2026-01-20 12:00:00") as frozen: + manager._get_zero_clock_duration("g1") + frozen.tick(30) + assert manager._get_zero_clock_duration("g2") == 0.0 + assert manager._get_zero_clock_duration("g1") == pytest.approx(30.0) + + def test_clear_resets_tracking(self, build_manager): + manager = build_manager(_RecentHarness) + with freeze_time("2026-01-20 12:00:00") as frozen: + manager._get_zero_clock_duration("g1") + frozen.tick(30) + manager._clear_zero_clock_tracking("g1") + assert "g1" not in manager._zero_clock_timestamps + # Restarts from zero after clearing. + assert manager._get_zero_clock_duration("g1") == 0.0 + + def test_clear_unknown_game_is_a_noop(self, build_manager): + manager = build_manager(_RecentHarness) + manager._clear_zero_clock_tracking("never-seen") # must not raise + assert manager._zero_clock_timestamps == {} + + +# --------------------------------------------------------------------------- +# Tier 2a — _is_game_really_over (SportsLive) +# --------------------------------------------------------------------------- + +class TestIsGameReallyOver: + def test_missing_clock_at_late_period_is_not_over(self, build_manager): + """THE baseball regression: no `clock` key at all, period 7. + + The rejected variant coerced a missing clock to the literal "0:00" and + declared the game over — dropping every MLB game from the 5th inning + onward, since baseball has no game clock and `period` is the inning. + A missing clock must fail safe. + """ + manager = build_manager(_LiveHarness) + g = game(period=7, period_text="Top 7th") + assert "clock" not in g + assert manager._is_game_really_over(g) is False + + def test_none_clock_at_late_period_is_not_over(self, build_manager): + manager = build_manager(_LiveHarness) + assert manager._is_game_really_over( + game(clock=None, period=7, period_text="Top 7th")) is False + + def test_non_string_clock_at_late_period_is_not_over(self, build_manager): + manager = build_manager(_LiveHarness) + assert manager._is_game_really_over( + game(clock=0, period=9, period_text="Bot 9th")) is False + + def test_blank_clock_at_late_period_is_not_over(self, build_manager): + manager = build_manager(_LiveHarness) + assert manager._is_game_really_over( + game(clock=" ", period=5, period_text="5th")) is False + + @pytest.mark.parametrize("period_text", ["Final", "final", "Final/OT", + "FINAL", "Final - SO"]) + def test_final_period_text_is_over(self, build_manager, period_text): + manager = build_manager(_LiveHarness) + assert manager._is_game_really_over( + game(clock="12:00", period=2, period_text=period_text)) is True + + def test_none_period_text_does_not_raise(self, build_manager): + """All nine plugin copies called `.lower()` on `game.get("period_text", "")`, + which is None when the key is present-but-None; `_detect_stale_games` + has no try/except around the call.""" + manager = build_manager(_LiveHarness) + assert manager._is_game_really_over( + game(clock="12:00", period=2, period_text=None)) is False + + def test_missing_period_text_does_not_raise(self, build_manager): + manager = build_manager(_LiveHarness) + assert manager._is_game_really_over(game(clock="12:00", period=2)) is False + + @pytest.mark.parametrize("clock", ["0:00", ":00", "00", "000", " 0:00 "]) + def test_expired_clock_at_final_period_is_over(self, build_manager, clock): + manager = build_manager(_LiveHarness) + assert manager._is_game_really_over( + game(clock=clock, period=4, period_text="Q4")) is True + + def test_expired_clock_after_final_period_is_over(self, build_manager): + manager = build_manager(_LiveHarness) + assert manager._is_game_really_over( + game(clock="0:00", period=5, period_text="OT")) is True + + def test_expired_clock_before_final_period_is_not_over(self, build_manager): + manager = build_manager(_LiveHarness) + assert manager._is_game_really_over( + game(clock="0:00", period=3, period_text="Q3")) is False + + @pytest.mark.parametrize("clock", [":40", "0:40", "1:00"]) + def test_running_clock_is_not_over(self, build_manager, clock): + """Sub-minute clocks like ':40' are legitimate, not expired.""" + manager = build_manager(_LiveHarness) + assert manager._is_game_really_over( + game(clock=clock, period=4, period_text="Q4")) is False + + def test_defaults_are_four_period_countdown(self): + assert SportsLive.FINAL_PERIOD == 4 + assert SportsLive.CLOCK_COUNTS_DOWN is True + + def test_final_period_override_three(self, build_manager): + """Hockey-shaped subclass: regulation ends after period 3.""" + manager = build_manager(_ThreePeriodLiveHarness) + assert manager.FINAL_PERIOD == 3 + assert manager._is_game_really_over( + game(clock="0:00", period=3, period_text="P3")) is True + assert manager._is_game_really_over( + game(clock="0:00", period=2, period_text="P2")) is False + # And the unmodified default still requires period 4. + assert build_manager(_LiveHarness)._is_game_really_over( + game(clock="0:00", period=3, period_text="P3")) is False + + def test_count_up_clock_never_expires(self, build_manager): + """Soccer/AFL/NRL: 0:00 means kickoff, so the clock branch must not run.""" + manager = build_manager(_CountUpLiveHarness) + assert manager.CLOCK_COUNTS_DOWN is False + for period in (1, 2, 4, 9): + assert manager._is_game_really_over( + game(clock="0:00", period=period, period_text="1st Half")) is False + + def test_count_up_clock_still_honors_final_text(self, build_manager): + manager = build_manager(_CountUpLiveHarness) + assert manager._is_game_really_over( + game(clock="0:00", period=2, period_text="Final")) is True + + +# --------------------------------------------------------------------------- +# Tier 2b — _detect_stale_games (SportsLive) +# --------------------------------------------------------------------------- + +class TestDetectStaleGames: + def test_initializer_defaults(self, build_manager): + manager = build_manager(_LiveHarness) + assert manager.game_update_timestamps == {} + assert manager.stale_game_timeout == 300 + + def test_stale_timeout_is_configurable(self, build_manager): + manager = build_manager(_LiveHarness, stale_game_timeout=42) + assert manager.stale_game_timeout == 42 + + def test_mutates_caller_list_in_place_and_returns_none(self, build_manager): + manager = build_manager(_LiveHarness) + fresh = game("1", period_text="P2", clock="10:00", period=2) + over = game("2", period_text="Final", clock="0:00", period=3) + games = [fresh, over] + original = games + + result = manager._detect_stale_games(games) + + assert result is None + assert games is original # same object, mutated in place + assert _ids(games) == ["1"] + + def test_evicts_only_past_timeout_games(self, build_manager): + manager = build_manager(_LiveHarness) + with freeze_time("2026-01-20 12:00:00"): + now = time.time() + manager.game_update_timestamps = { + "1": {"last_seen": now - 10}, # fresh + "2": {"last_seen": now - 299}, # just inside the timeout + "3": {"last_seen": now - 301}, # past the timeout + } + games = [game("1", period_text="P1", clock="10:00", period=1), + game("2", period_text="P1", clock="10:00", period=1), + game("3", period_text="P1", clock="10:00", period=1)] + manager._detect_stale_games(games) + + assert _ids(games) == ["1", "2"] + assert "3" not in manager.game_update_timestamps + assert set(manager.game_update_timestamps) == {"1", "2"} + + def test_unknown_last_seen_is_never_stale(self, build_manager): + """last_seen == 0 (or no entry) means 'never recorded', not 'ancient'.""" + manager = build_manager(_LiveHarness) + games = [game("1", period_text="P1", clock="10:00", period=1), + game("2", period_text="P1", clock="10:00", period=1)] + manager.game_update_timestamps = {"1": {"last_seen": 0}} + manager._detect_stale_games(games) + assert _ids(games) == ["1", "2"] + + def test_removes_games_that_are_really_over(self, build_manager): + manager = build_manager(_LiveHarness) + manager.game_update_timestamps = {"2": {"last_seen": 0}} + games = [game("1", period_text="Q2", clock="5:00", period=2), + game("2", period_text="Final", clock="0:00", period=4)] + manager._detect_stale_games(games) + assert _ids(games) == ["1"] + assert "2" not in manager.game_update_timestamps + + def test_keeps_clockless_late_game(self, build_manager): + """The end-to-end form of the baseball regression: a clockless game in + the 7th must survive the removal path.""" + manager = build_manager(_LiveHarness) + games = [game("mlb-1", period=7, period_text="Top 7th")] + manager._detect_stale_games(games) + assert _ids(games) == ["mlb-1"] + + def test_games_without_id_are_skipped(self, build_manager): + manager = build_manager(_LiveHarness) + no_id = {"home_abbr": "BOS", "away_abbr": "TOR", + "period_text": "Final", "clock": "0:00", "period": 4} + games = [no_id] + manager._detect_stale_games(games) + # `continue` fires before the "really over" check, so it stays. + assert games == [no_id] + + def test_empty_list_is_tolerated(self, build_manager): + manager = build_manager(_LiveHarness) + games = [] + assert manager._detect_stale_games(games) is None + assert games == [] + + def test_removal_is_by_value_not_identity(self, build_manager): + """Sharp edge worth pinning: `list.remove` compares with `dict.__eq__`, + so two structurally-equal dicts drop the FIRST occurrence.""" + manager = build_manager(_LiveHarness) + first = game("1", period_text="Final", clock="0:00", period=4) + twin = dict(first) + games = [first, twin] + manager._detect_stale_games(games) + # Both entries are removed here (two iterations, two removals), but the + # first removal deletes `first`, not the dict being iterated. + assert games == [] + + +# --------------------------------------------------------------------------- +# Tier 3a — _select_games_for_display (SportsUpcoming) +# --------------------------------------------------------------------------- + +class TestSelectGamesForDisplay: + def test_no_favorites_returns_all_sorted_ascending(self, build_manager): + manager = build_manager(_UpcomingHarness) + games = [game("late", start=at(20)), game("early", start=at(10)), + game("mid", start=at(15))] + assert _ids(manager._select_games_for_display(games, [])) == [ + "early", "mid", "late"] + + def test_missing_start_time_sorts_last(self, build_manager): + manager = build_manager(_UpcomingHarness) + games = [game("none", start=None), game("early", start=at(10))] + assert _ids(manager._select_games_for_display(games, [])) == [ + "early", "none"] + + def test_filters_to_favorite_teams(self, build_manager): + manager = build_manager(_UpcomingHarness) + games = [game("1", home="BOS", away="TOR", start=at(10)), + game("2", home="NYR", away="PIT", start=at(11)), + game("3", home="MTL", away="BOS", start=at(12))] + assert _ids(manager._select_games_for_display(games, ["BOS"])) == ["1", "3"] + + def test_respects_upcoming_games_to_show_per_team(self, build_manager): + manager = build_manager(_UpcomingHarness, upcoming_games_to_show=2) + games = [game(str(i), home="BOS", away="TOR", start=at(10 + i)) + for i in range(5)] + assert _ids(manager._select_games_for_display(games, ["BOS"])) == ["0", "1"] + + def test_game_between_two_favorites_counts_for_both(self, build_manager): + manager = build_manager(_UpcomingHarness, upcoming_games_to_show=1) + games = [game("shared", home="BOS", away="TOR", start=at(10)), + game("bos2", home="BOS", away="NYR", start=at(11)), + game("tor2", home="TOR", away="PIT", start=at(12))] + # "shared" fills both BOS's and TOR's single slot, so nothing else fits. + assert _ids(manager._select_games_for_display( + games, ["BOS", "TOR"])) == ["shared"] + + def test_deduplicates_by_game_id(self, build_manager): + manager = build_manager(_UpcomingHarness, upcoming_games_to_show=5) + g = game("dupe", home="BOS", away="TOR", start=at(10)) + assert _ids(manager._select_games_for_display( + [g, dict(g)], ["BOS"])) == ["dupe"] + + def test_non_favorite_games_excluded(self, build_manager): + manager = build_manager(_UpcomingHarness) + games = [game("1", home="NYR", away="PIT", start=at(10))] + assert manager._select_games_for_display(games, ["BOS"]) == [] + + def test_favorite_key_seam_supports_id_matching(self, build_manager): + """The NRL case: two clubs share the abbreviation 'NEW', so favorites + must be matched on team id. Only the seam changes — the promoted + method is identical.""" + games = [ + game("knights", home="NEW", away="SYD", home_id=1, away_id=2, + start=at(10)), + game("warriors", home="NEW", away="SYD", home_id=99, away_id=2, + start=at(11)), + ] + abbr_manager = build_manager(_UpcomingHarness) + # Abbreviation matching cannot tell the two "NEW" clubs apart. + assert _ids(abbr_manager._select_games_for_display(games, ["NEW"])) == [ + "knights", "warriors"] + + id_manager = build_manager(_IdFavoriteUpcomingHarness) + assert _ids(id_manager._select_games_for_display(games, ["99"])) == [ + "warriors"] + + def test_favorite_key_none_never_matches(self, build_manager): + """`_favorite_key` returning None (missing id) must not match, even + against a favorites list holding the string 'None'.""" + manager = build_manager(_IdFavoriteUpcomingHarness) + games = [game("1", home="BOS", away="TOR", start=at(10))] # no ids + assert manager._select_games_for_display(games, ["None"]) == [] + + +# --------------------------------------------------------------------------- +# Tier 3b — _select_recent_games_for_display (SportsRecent) +# --------------------------------------------------------------------------- + +class TestSelectRecentGamesForDisplay: + def test_no_favorites_returns_all_sorted_descending(self, build_manager): + manager = build_manager(_RecentHarness) + games = [game("early", start=at(10)), game("late", start=at(20)), + game("mid", start=at(15))] + assert _ids(manager._select_recent_games_for_display(games, [])) == [ + "late", "mid", "early"] + + def test_missing_start_time_sorts_last(self, build_manager): + manager = build_manager(_RecentHarness) + games = [game("none", start=None), game("late", start=at(20))] + assert _ids(manager._select_recent_games_for_display(games, [])) == [ + "late", "none"] + + def test_respects_recent_games_to_show_per_team(self, build_manager): + manager = build_manager(_RecentHarness, recent_games_to_show=2) + games = [game(str(i), home="BOS", away="TOR", start=at(10 + i)) + for i in range(5)] + # Most recent first. + assert _ids(manager._select_recent_games_for_display( + games, ["BOS"])) == ["4", "3"] + + def test_game_between_two_favorites_counts_for_both(self, build_manager): + manager = build_manager(_RecentHarness, recent_games_to_show=1) + games = [game("shared", home="BOS", away="TOR", start=at(20)), + game("bos2", home="BOS", away="NYR", start=at(19)), + game("tor2", home="TOR", away="PIT", start=at(18))] + assert _ids(manager._select_recent_games_for_display( + games, ["BOS", "TOR"])) == ["shared"] + + def test_deduplicates_by_game_id(self, build_manager): + manager = build_manager(_RecentHarness, recent_games_to_show=5) + g = game("dupe", home="BOS", away="TOR", start=at(10)) + assert _ids(manager._select_recent_games_for_display( + [g, dict(g)], ["BOS"])) == ["dupe"] + + def test_non_favorite_games_excluded(self, build_manager): + manager = build_manager(_RecentHarness) + games = [game("1", home="NYR", away="PIT", start=at(10))] + assert manager._select_recent_games_for_display(games, ["BOS"]) == [] + + def test_favorite_key_seam_supports_id_matching(self, build_manager): + games = [ + game("knights", home="NEW", away="SYD", home_id=1, away_id=2, + start=at(10)), + game("warriors", home="NEW", away="SYD", home_id=99, away_id=2, + start=at(11)), + ] + id_manager = build_manager(_IdFavoriteRecentHarness) + assert _ids(id_manager._select_recent_games_for_display( + games, ["99"])) == ["warriors"] + + +# --------------------------------------------------------------------------- +# Seam / inertness guards +# --------------------------------------------------------------------------- + +class TestPromotionShape: + def test_promoted_methods_live_on_the_right_classes(self): + assert hasattr(SportsRecent, "_get_zero_clock_duration") + assert hasattr(SportsRecent, "_clear_zero_clock_tracking") + assert hasattr(SportsRecent, "_select_recent_games_for_display") + assert hasattr(SportsUpcoming, "_select_games_for_display") + assert hasattr(SportsLive, "_is_game_really_over") + assert hasattr(SportsLive, "_detect_stale_games") + + def test_modes_does_not_define_the_favorite_key_seam(self): + """`_favorite_key` is a SportsCore override point. modes.py must call + it, never define it — this test fails if the seam is added in the + wrong file.""" + import src.base_classes.sports.modes as modes + + for cls in (SportsUpcoming, SportsRecent, SportsLive): + assert "_favorite_key" not in vars(cls) + assert "_favorite_key" not in modes.__dict__