mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-20 05:29:13 +00:00
Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2eea7a7f58 | ||
|
|
274ea65b39 | ||
|
|
6522c6d525 | ||
|
|
aeb99d7f09 | ||
|
|
2d6ff2bb88 |
@@ -75,4 +75,5 @@ jobs:
|
||||
test/test_element_style.py \
|
||||
test/test_sports_core_promotions.py \
|
||||
test/test_sports_modes_promotions.py \
|
||||
test/test_sports_capabilities.py
|
||||
test/test_sports_capabilities.py \
|
||||
test/test_sports_scroll.py
|
||||
|
||||
+25
-2
@@ -13,7 +13,21 @@ here, against a version number. When you add a module plugins will import via
|
||||
`src.*`, note it in the Unreleased section and bump `src/__init__.py` in the
|
||||
release that ships it.
|
||||
|
||||
## Unreleased
|
||||
**Use `ledmatrix_min_version` in manifests, not `ledmatrix_min`.** The loader
|
||||
accepts both, but the store flags the old spelling as deprecated
|
||||
(`store_manager.py`) and only the new one is in `schema/manifest_schema.json`.
|
||||
|
||||
## 3.2.0
|
||||
|
||||
**The first release shipping the unified sports library.** This is the version
|
||||
a sports plugin floors `ledmatrix_min_version` at before deleting its bundled
|
||||
copy of `sports.py`, `scroll_display.py`, `data_sources.py` or
|
||||
`base_odds_manager.py` — the sunset rule in
|
||||
`docs/plugin-development/08-shared-sports-code.md` keys on exactly this number.
|
||||
|
||||
Adoption is deliberately staged: the modules below ship here, plugins adopt them
|
||||
behind guarded imports, and only then do the bundled copies go away. Nothing in
|
||||
this release changes what an existing plugin loads.
|
||||
|
||||
### Added
|
||||
- `src/element_style.py` — per-element style resolver backing the
|
||||
@@ -24,7 +38,6 @@ release that ships it.
|
||||
system, data sources, API extractors, scroll helper, adaptive layout, loader
|
||||
compatibility warning) plus new characterization tests for
|
||||
`src/base_classes/sports.py` ahead of the shared sports-code unification.
|
||||
|
||||
- `src/base_classes/sports/` — `sports.py` is now a package (`core.py` +
|
||||
`modes.py`). The import path is unchanged: `from src.base_classes.sports
|
||||
import SportsCore` still works.
|
||||
@@ -49,7 +62,17 @@ release that ships it.
|
||||
built-in is verified against a verbatim transcription of the plugin
|
||||
implementation it replaces. An unknown name degrades to `simple`.
|
||||
|
||||
- `src/common/sports_scroll.py` — `SportsScrollDisplay` and
|
||||
`SportsScrollDisplayManager`, the shared scroll **orchestration** layer for
|
||||
the sports scoreboards, plus native support for
|
||||
`global_config['target_fps']` (the bundled plugin copies hardcode ~100 FPS
|
||||
via `scroll_delay` and never consult the global target). Content building
|
||||
(`prepare_scroll_content`, `_load_separator_icons`) is per-sport and stays an
|
||||
override point — see `docs/SPORTS_UNIFICATION.md` for where the line falls
|
||||
and why.
|
||||
|
||||
### Changed
|
||||
- `src/__init__.py` bumped to **3.2.0** — the number the sunset rule keys on.
|
||||
- **Live games are no longer dropped when the feed omits a game clock.**
|
||||
`SportsLive._is_game_really_over` previously (in the baseball and UFC
|
||||
plugin lineages) coerced a missing or non-string clock to the literal
|
||||
|
||||
@@ -76,6 +76,10 @@ src/base_classes/sports/
|
||||
capabilities/
|
||||
celebrations.py CelebrationMixin (opt-in: 4 of 9 plugins)
|
||||
rotation.py RotationStrategy + registry
|
||||
|
||||
src/common/
|
||||
sports_scroll.py SportsScrollDisplay / …Manager — scroll orchestration
|
||||
(content building stays in the plugins)
|
||||
```
|
||||
|
||||
`from src.base_classes.sports import SportsCore` keeps working — the package
|
||||
@@ -162,15 +166,39 @@ self.rotation = get_rotation_strategy("swrr", weight_for=self._live_weight)
|
||||
`weight_for` is supplied by the host, so the *favorites* policy stays with the
|
||||
plugin and `rotation.py` never learns what a favorite is. An unknown strategy
|
||||
name degrades to `simple` rather than raising: the name comes from user config,
|
||||
and a typo should cost the boost, not the scoreboard. A plugin needing an
|
||||
ordering core does not ship calls `register_rotation_strategy` instead of core
|
||||
growing a branch.
|
||||
and a typo should cost the boost, not the scoreboard. When a plugin needs an
|
||||
ordering that core does not ship, it calls `register_rotation_strategy` to add
|
||||
its own — rather than core growing a branch for it.
|
||||
|
||||
`test_sports_capabilities.py` checks each strategy against a **verbatim
|
||||
transcription** of the plugin code it replaces, over every live-game shape up to
|
||||
four games. That differential is what B5 deletes the bundled copies on the
|
||||
strength of.
|
||||
|
||||
## Scroll display — where the promotion line falls
|
||||
|
||||
`src/common/sports_scroll.py` is deliberately *not* a superset of the ten
|
||||
`scroll_display.py` copies. A method-level comparison of the eight that share a
|
||||
shape (f1 and ufc are genuine forks) found a sharp split:
|
||||
|
||||
| Layer | Evidence | Outcome |
|
||||
|---|---|---|
|
||||
| Orchestration — `get_all_vegas_content_items`, `clear_all`, `get_scroll_info`, `get_dynamic_duration`, `is_complete`, `display_frame` | identical to 96–100% similar across all eight | **promoted** |
|
||||
| Settings — `_get_scroll_settings` | one algorithm; the copies differ *only* in which league keys they walk | **promoted**, with the ladder as data (`SCROLL_LEAGUE_KEYS`) |
|
||||
| Content — `prepare_scroll_content`, `_load_separator_icons` | 8 distinct bodies across 8 plugins (145 lines, 53% similar at worst); icons 6% | **override point, permanently** |
|
||||
|
||||
Same name, different job: `prepare_scroll_content` draws *this sport's* game
|
||||
card. Merging the eight bodies would be the exact mistake the promotion rule
|
||||
exists to prevent, so the base class raises `NotImplementedError` rather than
|
||||
rendering something plausible — a base that rendered *something* would let a
|
||||
plugin ship a silently blank scroll.
|
||||
|
||||
The one behavior the upstreamed version adds is native
|
||||
`global_config['target_fps']` support. The bundled copies hardcode ~100 FPS via
|
||||
`scroll_delay = 0.01` and never consult the global smooth-scrolling target;
|
||||
Part A threaded it through each copy by hand, and this makes that threading
|
||||
legacy compatibility rather than the mechanism.
|
||||
|
||||
## Phases
|
||||
|
||||
| Phase | Scope | Risk control |
|
||||
@@ -178,9 +206,20 @@ strength of.
|
||||
| **B0** ✅ | Characterization tests, CI unit job, `element_style`, font cwd fix, CHANGELOG discipline | — |
|
||||
| **B1** ✅ | Promote the nine universal methods; convert `sports.py` → package | Characterization suite must stay green; no behavior change intended |
|
||||
| **B2** ✅ | `CelebrationMixin` + rotation strategies as opt-in capabilities | Plugins that don't opt in have zero new code in their MRO; strategies checked against verbatim plugin transcriptions |
|
||||
| **B3** | Upstream `ScrollDisplay` as `src/common/sports_scroll.py`, reading `global_config['target_fps']` natively | Plugin copies remain until sunset |
|
||||
| **B3** ✅ | Upstream the scroll **orchestration** layer as `src/common/sports_scroll.py`, reading `global_config['target_fps']` natively | Plugin copies remain until sunset; content building stays per-sport |
|
||||
| **B4** | Bump to 3.2.0, record modules in CHANGELOG, migrate `ledmatrix_min` → `ledmatrix_min_version` | Gives plugins a version to floor on |
|
||||
| **B5** | Pilot one plugin per lineage (hockey, soccer, football) on guarded core imports; then the remaining six; then delete bundled copies | Pilot soaks before rollout; harness + golden suites gate each |
|
||||
| **B5** ⏳ | Pilot one plugin per lineage (hockey, soccer, football) on core imports; then the remaining six; then delete bundled copies | Pilot soaks before rollout; harness + golden suites gate each |
|
||||
|
||||
**B5 is blocked on this PR merging and 3.2.0 shipping** — a plugin cannot floor
|
||||
`ledmatrix_min_version` at a release that does not exist, and an unguarded
|
||||
`src.common.sports_scroll` import would break every user on 3.1.0.
|
||||
|
||||
The hockey scroll-display pilot has been **validated ahead of that gate**:
|
||||
adopted against a core carrying 3.2.0, `scroll_display.py` went from 691 to 289
|
||||
lines and all 16 harness renders (8 sizes × 2 screens) came out byte-for-byte
|
||||
identical to the pre-adoption run. The adoption recipe and the two gotchas it
|
||||
surfaced are written up in the plugins repo's
|
||||
`docs/plugin-development/08-shared-sports-code.md`.
|
||||
|
||||
## Rules for contributors
|
||||
|
||||
|
||||
+1
-1
@@ -4,5 +4,5 @@ LEDMatrix Display System
|
||||
Core source package for the LED Matrix Display project.
|
||||
"""
|
||||
|
||||
__version__ = "3.1.0"
|
||||
__version__ = "3.2.0"
|
||||
|
||||
|
||||
@@ -151,7 +151,12 @@ class Baseball(SportsCore):
|
||||
|
||||
# Only log detailed information for favorite teams
|
||||
if is_favorite_game:
|
||||
self.logger.debug(f"Full status data: {game_event['status']}")
|
||||
# Use the validated competition-level `status` here too. MiLB
|
||||
# events carry no event-level one, so this debug line raised a
|
||||
# KeyError and dropped the very games it was meant to help
|
||||
# diagnose -- and only for favourites, which is the worst way
|
||||
# for it to fail.
|
||||
self.logger.debug(f"Full status data: {status}")
|
||||
self.logger.debug(f"Status type: {game_status}, State: {status_state}")
|
||||
self.logger.debug(f"Status detail: {status['type'].get('detail', '')}")
|
||||
self.logger.debug(
|
||||
|
||||
@@ -51,7 +51,20 @@ class CelebrationMixin:
|
||||
super().__init__(*args, **kwargs)
|
||||
mode_config = getattr(self, "mode_config", {}) or {}
|
||||
self.celebration_enabled = mode_config.get("celebration_enabled", True)
|
||||
self.celebration_duration = mode_config.get("celebration_duration", 8)
|
||||
# Coerced and floored at init: this value is compared numerically on the
|
||||
# display path, where a string from a hand-edited config would raise
|
||||
# TypeError outside any try block, and a zero or negative value would
|
||||
# arm a celebration that can never render.
|
||||
raw_duration = mode_config.get("celebration_duration", 8)
|
||||
try:
|
||||
self.celebration_duration = max(1.0, float(raw_duration))
|
||||
except (TypeError, ValueError):
|
||||
self.logger.warning(
|
||||
"[Celebrations] Unusable celebration_duration %r; using 8s. "
|
||||
"Set a positive number of seconds.",
|
||||
raw_duration,
|
||||
)
|
||||
self.celebration_duration = 8.0
|
||||
# Both spellings: the soccer lineage ships `celebrate_opponent_goals`,
|
||||
# football ships `celebrate_opponent_scores`. Whichever the plugin's
|
||||
# schema declares is the one its users have set.
|
||||
@@ -122,6 +135,25 @@ class CelebrationMixin:
|
||||
# Favorites exist but this team isn't one -> it's the opponent.
|
||||
return self.celebrate_opponent_scores
|
||||
|
||||
def prune_score_baselines(self, live_games: List[Dict]) -> None:
|
||||
"""Drop baselines for games no longer live.
|
||||
|
||||
Only :meth:`_check_for_win` removes entries, and it only fires for games
|
||||
seen to go final. A game that vanishes from the live list any other way
|
||||
— postponed, dropped by the feed, or simply still live when the board
|
||||
restarts — leaves its baseline behind forever, so on a board that runs
|
||||
all season the dict grows without bound.
|
||||
|
||||
Call this from ``update()`` with the current live set, alongside the
|
||||
equivalent pruning in :meth:`SmoothWeightedRotation.next_game`.
|
||||
"""
|
||||
live_ids = {g.get("id") for g in live_games}
|
||||
self._score_baselines = {
|
||||
gid: baseline
|
||||
for gid, baseline in self._score_baselines.items()
|
||||
if gid in live_ids
|
||||
}
|
||||
|
||||
def has_active_celebration(self) -> bool:
|
||||
"""True while a celebration is within its display window."""
|
||||
celebration = self.active_celebration
|
||||
@@ -255,7 +287,7 @@ class CelebrationMixin:
|
||||
# resumes on it.
|
||||
self.current_game = dict(game)
|
||||
self.logger.info(
|
||||
f"Celebration ({kind}) armed: {phrase} "
|
||||
f"[Celebrations] {kind} armed: {phrase} "
|
||||
f"[{game.get('away_abbr')} {away_score}-{home_score} {game.get('home_abbr')}]"
|
||||
)
|
||||
|
||||
@@ -323,7 +355,7 @@ class CelebrationMixin:
|
||||
away_logo, (-2, center_y - away_logo.height // 2), away_logo
|
||||
)
|
||||
except Exception as e:
|
||||
self.logger.debug(f"Celebration logo load failed: {e}")
|
||||
self.logger.debug(f"[Celebrations] Logo load failed: {e}")
|
||||
|
||||
# Phrase across the top, shrunk to fit the panel width.
|
||||
phrase = celebration["phrase"]
|
||||
@@ -365,12 +397,19 @@ class CelebrationMixin:
|
||||
return False
|
||||
celebration = self.active_celebration
|
||||
if celebration:
|
||||
if time.time() - celebration["started_at"] < self.celebration_duration:
|
||||
if self.has_active_celebration():
|
||||
try:
|
||||
self._draw_celebration_layout(celebration, force_clear)
|
||||
return True
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error drawing celebration: {e}", exc_info=True)
|
||||
self.logger.error(
|
||||
f"[Celebrations] Error drawing celebration: {e}", exc_info=True
|
||||
)
|
||||
# Disarm rather than retry: the same render would fail on
|
||||
# every frame for the rest of the window, logging a
|
||||
# traceback each time and leaving the scorebug off screen.
|
||||
self.active_celebration = None
|
||||
self.last_game_switch = time.time()
|
||||
else:
|
||||
self.active_celebration = None
|
||||
# Reset the dwell so the scorebug resumes on the scoring/winning
|
||||
|
||||
@@ -50,6 +50,12 @@ class RotationStrategy:
|
||||
#: Name this strategy is registered under. Set by :func:`register_rotation_strategy`.
|
||||
name: str = ""
|
||||
|
||||
#: Ceiling on a per-game weight. A cycle is ``sum(weights)`` long and each
|
||||
#: step scans every game, so an unbounded weight — a misread config field,
|
||||
#: say — would spin the display thread for an unbounded time. On a Pi that
|
||||
#: stalls rendering outright, so the bound is clamped like the floor is.
|
||||
MAX_WEIGHT = 16
|
||||
|
||||
def __init__(self, weight_for: Optional[Callable[[Dict], int]] = None):
|
||||
self._weight_for = weight_for or (lambda game: 1)
|
||||
|
||||
@@ -58,7 +64,8 @@ class RotationStrategy:
|
||||
|
||||
A weight below 1 is clamped up: a zero or negative weight would starve
|
||||
a game out of the rotation entirely, which no caller means to express
|
||||
and which would make ``total_weight`` collapse.
|
||||
and which would make ``total_weight`` collapse. It is clamped down at
|
||||
:attr:`MAX_WEIGHT` for the reason documented there.
|
||||
"""
|
||||
weights: Dict[str, int] = {}
|
||||
for game in games:
|
||||
@@ -69,7 +76,7 @@ class RotationStrategy:
|
||||
weight = int(self._weight_for(game))
|
||||
except (TypeError, ValueError):
|
||||
weight = 1
|
||||
weights[gid] = max(1, weight)
|
||||
weights[gid] = min(self.MAX_WEIGHT, max(1, weight))
|
||||
return weights
|
||||
|
||||
def schedule(self, games: List[Dict]) -> List[str]:
|
||||
@@ -183,7 +190,11 @@ class SmoothWeightedRotation(RotationStrategy):
|
||||
weights = self.weights(games)
|
||||
if not weights:
|
||||
return []
|
||||
preview = SmoothWeightedRotation(self._weight_for)
|
||||
# type(self), not this class: a subclass that overrides next_game must
|
||||
# be previewed through its own ordering, or the returned order is not
|
||||
# the one repeated next_game calls would produce — which is exactly
|
||||
# what this method promises.
|
||||
preview = type(self)(self._weight_for)
|
||||
preview._current = dict(self._current)
|
||||
order: List[str] = []
|
||||
for _ in range(sum(weights.values())):
|
||||
@@ -200,12 +211,19 @@ _REGISTRY: Dict[str, Type[RotationStrategy]] = {}
|
||||
def register_rotation_strategy(name: str, factory: Type[RotationStrategy]) -> None:
|
||||
"""Register a rotation strategy under ``name``.
|
||||
|
||||
A plugin needing an ordering core does not ship registers it here rather
|
||||
than core growing a sport-specific branch. Re-registering a name replaces
|
||||
it, so a plugin may also override a built-in for itself.
|
||||
When a plugin needs an ordering that core does not ship, it registers its
|
||||
own here instead of core growing a sport-specific branch. Re-registering a
|
||||
name replaces it, so a plugin may also override a built-in for itself.
|
||||
"""
|
||||
if not name:
|
||||
raise ValueError("rotation strategy name must be a non-empty string")
|
||||
# Fail at registration, not at the first schedule() call several frames
|
||||
# later, where the cause is no longer on the stack.
|
||||
if not (isinstance(factory, type) and issubclass(factory, RotationStrategy)):
|
||||
raise TypeError(
|
||||
f"rotation strategy {name!r} must be a RotationStrategy subclass, "
|
||||
f"got {factory!r}"
|
||||
)
|
||||
factory.name = name
|
||||
_REGISTRY[name] = factory
|
||||
|
||||
|
||||
@@ -185,9 +185,12 @@ class SportsCore(ABC):
|
||||
}
|
||||
self.last_update = 0
|
||||
self.current_game = None
|
||||
# Cooldown clock for _should_log(). Initialized here rather than lazily:
|
||||
# _should_log() reads it unguarded, so whichever warning fires first
|
||||
# would otherwise raise AttributeError instead of logging.
|
||||
# Cooldown clocks for _should_log(), one per warning type. Initialized
|
||||
# here rather than lazily: _should_log() reads them unguarded, so
|
||||
# whichever warning fires first would otherwise raise AttributeError
|
||||
# instead of logging. `_last_warning_time` is the single-clock field
|
||||
# the plugin copies expose; kept for subclasses that read it.
|
||||
self._warning_cooldowns: Dict[str, float] = {}
|
||||
self._last_warning_time = 0
|
||||
self.fonts = self._load_fonts()
|
||||
|
||||
@@ -428,7 +431,7 @@ class SportsCore(ABC):
|
||||
return False
|
||||
|
||||
|
||||
def _load_fonts(self):
|
||||
def _load_fonts(self) -> Dict[str, Any]:
|
||||
"""Load fonts used by the scoreboard.
|
||||
|
||||
Paths go through :meth:`_resolve_font_path` so the bundled fonts are
|
||||
@@ -437,7 +440,7 @@ class SportsCore(ABC):
|
||||
default font whenever the process started elsewhere (the plugin safety
|
||||
harness on CI being the case that surfaced it).
|
||||
"""
|
||||
fonts = {}
|
||||
fonts: Dict[str, Any] = {}
|
||||
press_start = self._resolve_font_path("PressStart2P-Regular.ttf")
|
||||
four_by_six = self._resolve_font_path("4x6-font.ttf")
|
||||
try:
|
||||
@@ -447,9 +450,15 @@ class SportsCore(ABC):
|
||||
fonts['status'] = ImageFont.truetype(four_by_six, 6) # Using 4x6 for status
|
||||
fonts['detail'] = ImageFont.truetype(four_by_six, 6) # Added detail font
|
||||
fonts['rank'] = ImageFont.truetype(press_start, 10)
|
||||
logging.info("Successfully loaded fonts") # Changed log prefix
|
||||
self.logger.info("Successfully loaded fonts")
|
||||
except OSError:
|
||||
logging.warning("Fonts not found, using default PIL font.") # Changed log prefix
|
||||
# Name the directory we searched: the usual cause is an install
|
||||
# whose assets/fonts is missing, and the bare message sent people
|
||||
# hunting for a font-format problem instead.
|
||||
self.logger.warning(
|
||||
"Fonts not found under %s, using default PIL font.",
|
||||
self._font_root(),
|
||||
)
|
||||
fonts['score'] = ImageFont.load_default()
|
||||
fonts['time'] = ImageFont.load_default()
|
||||
fonts['team'] = ImageFont.load_default()
|
||||
@@ -634,9 +643,19 @@ class SportsCore(ABC):
|
||||
return pytz.utc
|
||||
|
||||
def _should_log(self, warning_type: str, cooldown: int = 60) -> bool:
|
||||
"""Check if we should log a warning based on cooldown period."""
|
||||
"""Whether a warning of this kind is outside its cooldown window.
|
||||
|
||||
Cooldowns are tracked **per ``warning_type``**. They previously shared
|
||||
one timestamp, so the parameter was accepted and ignored: an API-error
|
||||
warning would silence an unrelated cache warning for the next minute,
|
||||
and whichever fired first won. Nothing in core called this, so no
|
||||
behavior regressed with the fix — but every caller has always been
|
||||
entitled to assume its own warning type has its own clock.
|
||||
"""
|
||||
current_time = time.time()
|
||||
if current_time - self._last_warning_time > cooldown:
|
||||
if current_time - self._warning_cooldowns.get(warning_type, 0) > cooldown:
|
||||
self._warning_cooldowns[warning_type] = current_time
|
||||
# Kept in step for subclasses that read it directly.
|
||||
self._last_warning_time = current_time
|
||||
return True
|
||||
return False
|
||||
|
||||
@@ -0,0 +1,485 @@
|
||||
"""Shared scroll-display scaffolding for the sports scoreboards.
|
||||
|
||||
Ten plugins ship a `scroll_display.py`. A method-level comparison of the eight
|
||||
that share a shape (f1 and ufc are genuine forks) found a sharp split, and this
|
||||
module is drawn along it rather than around all of it:
|
||||
|
||||
* The **orchestration layer is converged** — ``get_all_vegas_content_items`` is
|
||||
byte-identical in all eight, and ``clear_all``, ``get_scroll_info``,
|
||||
``get_dynamic_duration``, ``is_complete`` and ``display_frame`` are 96-100%
|
||||
similar. That is what lives here.
|
||||
* The **content layer has genuinely diverged** — ``prepare_scroll_content`` has
|
||||
eight distinct bodies across eight plugins (145 lines, 53% similarity at
|
||||
worst) and ``_load_separator_icons`` seven (6% at worst). Those build each
|
||||
sport's game cards and icon strip; they are *not* drift to be merged but
|
||||
per-sport rendering. They stay override points here, permanently.
|
||||
|
||||
Promoting the content layer would be exactly the mistake
|
||||
``docs/SPORTS_UNIFICATION.md`` warns against — merging on the intuition that
|
||||
same-named methods are the same method. Same name, different job.
|
||||
|
||||
The one behavior this module adds over the plugin copies is native support for
|
||||
``global_config['target_fps']``: the bundled copies hardcode ~100 FPS via
|
||||
``scroll_delay=0.01`` and never consult the global smooth-scrolling target. A
|
||||
plugin inheriting from here gets it for free.
|
||||
|
||||
Usage::
|
||||
|
||||
class HockeyScrollDisplay(SportsScrollDisplay):
|
||||
SCROLL_LEAGUE_KEYS = ("nhl", "ncaa_mens", "ncaam_hockey")
|
||||
|
||||
def prepare_scroll_content(self, games, game_type, leagues, rankings=None):
|
||||
... # build this sport's cards
|
||||
|
||||
class HockeyScrollDisplayManager(SportsScrollDisplayManager):
|
||||
display_class = HockeyScrollDisplay
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import time
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from PIL import Image
|
||||
|
||||
from src.common.scroll_helper import ScrollHelper
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
#: Defaults every copy agreed on. A subclass overrides
|
||||
#: :meth:`SportsScrollDisplay.scroll_settings_defaults` to change them —
|
||||
#: the soccer lineage uses a 24px gap and min/max duration keys instead.
|
||||
DEFAULT_SCROLL_SETTINGS: Dict[str, Any] = {
|
||||
"scroll_speed": 50.0,
|
||||
"scroll_delay": 0.01,
|
||||
"gap_between_games": 48,
|
||||
"show_league_separators": True,
|
||||
"dynamic_duration": True,
|
||||
}
|
||||
|
||||
#: Bounds on the px/second -> px/frame conversion, applied before the helper
|
||||
#: sees the value. FPS is *not* clamped here — ScrollHelper.set_target_fps
|
||||
#: already does that, and a second copy of the range would drift from it.
|
||||
MIN_PIXELS_PER_FRAME = 0.1
|
||||
MAX_PIXELS_PER_FRAME = 5.0
|
||||
|
||||
#: Pacing to assume when scroll_delay is 0, i.e. the plugin has not set one.
|
||||
ASSUMED_FPS_WHEN_UNPACED = 100.0
|
||||
|
||||
|
||||
class SportsScrollDisplay:
|
||||
"""One scrolling strip of game cards.
|
||||
|
||||
Subclasses supply the content (:meth:`prepare_scroll_content`) and,
|
||||
optionally, the per-sport league ladder and separator icons. Everything
|
||||
else — helper configuration, frame pumping, completion, state — is here.
|
||||
"""
|
||||
|
||||
#: Config keys to walk when looking for per-league ``scroll_settings``,
|
||||
#: most-preferred first. A sport's own league names, which is the *only*
|
||||
#: reason the eight copies of ``_get_scroll_settings`` differ. Empty means
|
||||
#: the plugin has no per-league scroll settings.
|
||||
SCROLL_LEAGUE_KEYS: tuple = ()
|
||||
|
||||
#: Config block holding scroll settings when the plugin keeps them in one
|
||||
#: place rather than per league (the afl/nrl/soccer shape).
|
||||
SCROLL_CONFIG_KEY: Optional[str] = None
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
display_manager,
|
||||
config: Dict[str, Any],
|
||||
custom_logger: Optional[logging.Logger] = None,
|
||||
global_config: Optional[Dict[str, Any]] = None,
|
||||
):
|
||||
"""
|
||||
:param display_manager: the core display manager
|
||||
:param config: the plugin's configuration
|
||||
:param custom_logger: the plugin's logger, so scroll lines are attributed
|
||||
:param global_config: the LEDMatrix global config — the source of
|
||||
``target_fps``. Optional so an older caller that does not pass it
|
||||
keeps working at the config-derived pacing.
|
||||
"""
|
||||
self.display_manager = display_manager
|
||||
self.config = config
|
||||
self.logger = custom_logger or logger
|
||||
self.global_config = global_config or {}
|
||||
|
||||
if getattr(display_manager, "matrix", None) is not None:
|
||||
self.display_width = display_manager.matrix.width
|
||||
self.display_height = display_manager.matrix.height
|
||||
else:
|
||||
self.display_width = getattr(display_manager, "width", 128)
|
||||
self.display_height = getattr(display_manager, "height", 32)
|
||||
|
||||
self.scroll_helper = ScrollHelper(
|
||||
self.display_width, self.display_height, self.logger
|
||||
)
|
||||
self._configure_scroll_helper()
|
||||
|
||||
self._logo_cache: Dict[str, Image.Image] = {}
|
||||
self._separator_icons: Dict[str, Image.Image] = {}
|
||||
self._load_separator_icons()
|
||||
|
||||
self._current_games: List[Dict] = []
|
||||
self._current_game_type: str = ""
|
||||
self._current_leagues: List[str] = []
|
||||
self._vegas_content_items: List[Image.Image] = []
|
||||
self._is_scrolling = False
|
||||
self._scroll_start_time: Optional[float] = None
|
||||
self._last_log_time: float = 0
|
||||
self._log_interval: float = 5.0
|
||||
self._frame_count: int = 0
|
||||
self._fps_sample_start: float = time.time()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Override points
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def prepare_scroll_content(
|
||||
self,
|
||||
games: List[Dict],
|
||||
game_type: str,
|
||||
leagues: List[str],
|
||||
rankings_cache: Optional[Dict[str, int]] = None,
|
||||
) -> bool:
|
||||
"""Render ``games`` into one wide image and hand it to the scroll helper.
|
||||
|
||||
**Per-sport by nature, not by drift** — the eight plugin copies have
|
||||
eight different bodies because each draws its own card. Implementations
|
||||
build the strip, hand it over with
|
||||
``self.scroll_helper.set_scrolling_image(...)`` (or
|
||||
``create_scrolling_image(...)`` from a list of cards), and record
|
||||
``self._current_games`` / ``_current_game_type`` / ``_current_leagues``.
|
||||
|
||||
:returns: True when there is content to scroll.
|
||||
"""
|
||||
raise NotImplementedError(
|
||||
f"{type(self).__name__} must implement prepare_scroll_content(); "
|
||||
"it builds this sport's game cards and is not shared code."
|
||||
)
|
||||
|
||||
def _load_separator_icons(self) -> None:
|
||||
"""Populate ``self._separator_icons``. Per-sport; no-op by default."""
|
||||
|
||||
def scroll_settings_defaults(self) -> Dict[str, Any]:
|
||||
"""The baseline scroll settings before any config is applied."""
|
||||
defaults = dict(DEFAULT_SCROLL_SETTINGS)
|
||||
defaults["game_card_width"] = self.display_width
|
||||
return defaults
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Settings
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _get_scroll_settings(self, league: Optional[str] = None) -> Dict[str, Any]:
|
||||
"""Resolve scroll settings: defaults, then the most specific override.
|
||||
|
||||
Precedence: the named ``league``, then each entry of
|
||||
:attr:`SCROLL_LEAGUE_KEYS` in order, then :attr:`SCROLL_CONFIG_KEY`.
|
||||
The eight plugin copies implement exactly this and differ only in which
|
||||
league names they walk — which is why the ladder is data here rather
|
||||
than a body per sport.
|
||||
"""
|
||||
settings = self.scroll_settings_defaults()
|
||||
|
||||
candidates: List[str] = []
|
||||
if league:
|
||||
candidates.append(league)
|
||||
candidates.extend(self.SCROLL_LEAGUE_KEYS)
|
||||
for key in candidates:
|
||||
override = (self.config.get(key) or {}).get("scroll_settings")
|
||||
if override:
|
||||
return {**settings, **override}
|
||||
|
||||
if self.SCROLL_CONFIG_KEY:
|
||||
override = self.config.get(self.SCROLL_CONFIG_KEY) or {}
|
||||
if override:
|
||||
return {**settings, **override}
|
||||
return settings
|
||||
|
||||
def _resolve_target_fps(self) -> Optional[float]:
|
||||
"""The global smooth-scrolling FPS target, or None to keep config pacing.
|
||||
|
||||
Coerced before use: a malformed value in the global config must degrade
|
||||
to the existing ``scroll_delay`` pacing, never raise on a display path.
|
||||
"""
|
||||
raw = self.global_config.get("target_fps") or self.global_config.get(
|
||||
"scroll_target_fps"
|
||||
)
|
||||
try:
|
||||
return float(raw) if raw is not None else None
|
||||
except (TypeError, ValueError):
|
||||
self.logger.debug("Ignoring unusable target_fps: %r", raw)
|
||||
return None
|
||||
|
||||
def _coerce_float(self, value: Any, default: float) -> float:
|
||||
"""A usable float from config, or ``default``.
|
||||
|
||||
``dict.get(key, default)`` only helps when the key is *absent*; a key
|
||||
present with ``null`` or a string returns that value verbatim and blows
|
||||
up in the arithmetic below — inside ``__init__``, so the whole display
|
||||
fails to construct.
|
||||
"""
|
||||
if value is None:
|
||||
return default
|
||||
try:
|
||||
return float(value)
|
||||
except (TypeError, ValueError):
|
||||
self.logger.warning(
|
||||
"Ignoring unusable scroll setting %r; using %s", value, default
|
||||
)
|
||||
return default
|
||||
|
||||
def _configure_scroll_helper(self) -> None:
|
||||
"""Apply config to the scroll helper. Safe to call again after a change."""
|
||||
settings = self._get_scroll_settings()
|
||||
|
||||
scroll_speed = self._coerce_float(settings.get("scroll_speed"), 50.0)
|
||||
scroll_delay = self._coerce_float(settings.get("scroll_delay"), 0.01)
|
||||
dynamic_duration = bool(settings.get("dynamic_duration", True))
|
||||
|
||||
self.scroll_helper.set_scroll_delay(scroll_delay)
|
||||
self.scroll_helper.set_dynamic_duration_settings(
|
||||
enabled=dynamic_duration,
|
||||
min_duration=settings.get("min_duration", 30),
|
||||
max_duration=settings.get("max_duration", 600),
|
||||
buffer=0.2, # ensure the strip clears the panel completely
|
||||
)
|
||||
# Frame-based scrolling: motion advances per rendered frame rather than
|
||||
# per wall-clock second, which is what makes the pacing stable.
|
||||
self.scroll_helper.set_frame_based_scrolling(True)
|
||||
|
||||
# Config states speed in px/second; frame-based mode wants px/frame.
|
||||
if scroll_delay > 0:
|
||||
pixels_per_frame = scroll_speed * scroll_delay
|
||||
else:
|
||||
pixels_per_frame = scroll_speed / ASSUMED_FPS_WHEN_UNPACED
|
||||
pixels_per_frame = max(
|
||||
MIN_PIXELS_PER_FRAME, min(MAX_PIXELS_PER_FRAME, pixels_per_frame)
|
||||
)
|
||||
self.scroll_helper.set_scroll_speed(pixels_per_frame)
|
||||
|
||||
effective_pps = (
|
||||
pixels_per_frame / scroll_delay
|
||||
if scroll_delay > 0
|
||||
else pixels_per_frame * ASSUMED_FPS_WHEN_UNPACED
|
||||
)
|
||||
self.logger.info(
|
||||
f"ScrollHelper configured: {pixels_per_frame:.2f} px/frame, "
|
||||
f"delay={scroll_delay}s (effective {effective_pps:.1f} px/s from "
|
||||
f"{scroll_speed} px/s config), dynamic_duration={dynamic_duration}"
|
||||
)
|
||||
|
||||
# The reason this module exists upstream: the bundled copies hardcode
|
||||
# ~100 FPS via scroll_delay and never consult the global target.
|
||||
# No hasattr guard here, unlike the plugin copies: they probe because
|
||||
# they may run against an older core, whereas this module ships in the
|
||||
# same release as the ScrollHelper it calls. The helper clamps.
|
||||
target_fps = self._resolve_target_fps()
|
||||
if target_fps:
|
||||
self.scroll_helper.set_target_fps(target_fps)
|
||||
self.logger.info(f"Target FPS set to {target_fps}")
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Frame pumping
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def display_scroll_frame(self) -> bool:
|
||||
"""Advance and render one frame.
|
||||
|
||||
:returns: True if a frame was drawn; False when there is no content or
|
||||
the frame could not be rendered.
|
||||
"""
|
||||
if not self.scroll_helper.cached_image:
|
||||
return False
|
||||
|
||||
try:
|
||||
# Inside the try, not before it: advancing the position and cropping
|
||||
# the visible slice are as capable of raising as the display push,
|
||||
# and the promise below is that no frame failure reaches the
|
||||
# plugin's loop.
|
||||
self.scroll_helper.update_scroll_position()
|
||||
visible = self.scroll_helper.get_visible_portion()
|
||||
if not visible:
|
||||
return False
|
||||
|
||||
self.display_manager.image = visible
|
||||
self.display_manager.update_display()
|
||||
self._frame_count += 1
|
||||
self.scroll_helper.log_frame_rate()
|
||||
self._log_scroll_progress()
|
||||
except Exception:
|
||||
# A display failure must not propagate into the plugin's loop.
|
||||
self.logger.exception("Error displaying scroll frame")
|
||||
return False
|
||||
return True
|
||||
|
||||
def _log_scroll_progress(self) -> None:
|
||||
"""Emit a throttled progress line."""
|
||||
now = time.time()
|
||||
if now - self._last_log_time < self._log_interval:
|
||||
return
|
||||
self._last_log_time = now
|
||||
elapsed = now - self._fps_sample_start
|
||||
fps = self._frame_count / elapsed if elapsed > 0 else 0.0
|
||||
self.logger.debug(
|
||||
f"Scrolling {len(self._current_games)} {self._current_game_type} "
|
||||
f"game(s) at {fps:.1f} FPS"
|
||||
)
|
||||
|
||||
def is_scroll_complete(self) -> bool:
|
||||
"""True when the strip has scrolled fully past the panel."""
|
||||
return self.scroll_helper.is_scroll_complete()
|
||||
|
||||
def reset_scroll(self) -> None:
|
||||
"""Return the strip to its starting position, keeping the content."""
|
||||
self.scroll_helper.reset_scroll()
|
||||
self._frame_count = 0
|
||||
self._fps_sample_start = time.time()
|
||||
self.logger.debug("Scroll position reset")
|
||||
|
||||
def clear(self) -> None:
|
||||
"""Drop cached content and reset tracking state."""
|
||||
self.scroll_helper.clear_cache()
|
||||
self._current_games = []
|
||||
self._current_game_type = ""
|
||||
self._current_leagues = []
|
||||
self._vegas_content_items = []
|
||||
self._is_scrolling = False
|
||||
self._scroll_start_time = None
|
||||
self.logger.debug("Scroll display cleared")
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Introspection
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def get_dynamic_duration(self) -> int:
|
||||
"""How long this content needs to scroll fully, in seconds."""
|
||||
return self.scroll_helper.get_dynamic_duration()
|
||||
|
||||
def has_cached_content(self) -> bool:
|
||||
"""Whether content is prepared and ready to scroll."""
|
||||
return bool(self.scroll_helper.cached_image)
|
||||
|
||||
def get_current_game_count(self) -> int:
|
||||
return len(self._current_games)
|
||||
|
||||
def get_current_leagues(self) -> List[str]:
|
||||
return list(self._current_leagues)
|
||||
|
||||
def get_scroll_info(self) -> Dict[str, Any]:
|
||||
"""Helper state plus this display's tracking state, for logging/debug."""
|
||||
info = self.scroll_helper.get_scroll_info()
|
||||
info.update(
|
||||
{
|
||||
"game_count": len(self._current_games),
|
||||
"game_type": self._current_game_type,
|
||||
"leagues": self._current_leagues,
|
||||
"is_scrolling": self._is_scrolling,
|
||||
}
|
||||
)
|
||||
return info
|
||||
|
||||
|
||||
class SportsScrollDisplayManager:
|
||||
"""One :class:`SportsScrollDisplay` per game type ('live'/'recent'/'upcoming').
|
||||
|
||||
Subclasses set :attr:`display_class`; everything else was near-identical
|
||||
across the eight plugin copies.
|
||||
"""
|
||||
|
||||
#: The SportsScrollDisplay subclass to instantiate per game type.
|
||||
display_class = SportsScrollDisplay
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
display_manager,
|
||||
config: Dict[str, Any],
|
||||
custom_logger: Optional[logging.Logger] = None,
|
||||
global_config: Optional[Dict[str, Any]] = None,
|
||||
):
|
||||
self.display_manager = display_manager
|
||||
self.config = config
|
||||
self.logger = custom_logger or logger
|
||||
self.global_config = global_config or {}
|
||||
self._scroll_displays: Dict[str, SportsScrollDisplay] = {}
|
||||
# "" rather than None, matching SportsScrollDisplay's own empty value —
|
||||
# both are falsy, so `game_type or self._current_game_type` behaved
|
||||
# either way, but two spellings of "nothing active" across two classes
|
||||
# is a trap for anyone comparing state between them.
|
||||
self._current_game_type: str = ""
|
||||
|
||||
def get_scroll_display(self, game_type: str) -> SportsScrollDisplay:
|
||||
"""The display for ``game_type``, created on first use."""
|
||||
if game_type not in self._scroll_displays:
|
||||
self._scroll_displays[game_type] = self.display_class(
|
||||
self.display_manager,
|
||||
self.config,
|
||||
self.logger,
|
||||
global_config=self.global_config,
|
||||
)
|
||||
return self._scroll_displays[game_type]
|
||||
|
||||
def prepare_and_display(
|
||||
self,
|
||||
games: List[Dict],
|
||||
game_type: str,
|
||||
leagues: List[str],
|
||||
rankings_cache: Optional[Dict[str, int]] = None,
|
||||
) -> bool:
|
||||
"""Build content for ``game_type`` and make it the active strip."""
|
||||
scroll_display = self.get_scroll_display(game_type)
|
||||
try:
|
||||
success = scroll_display.prepare_scroll_content(
|
||||
games, game_type, leagues, rankings_cache
|
||||
)
|
||||
except Exception:
|
||||
# prepare_scroll_content is subclass-implemented and builds cards
|
||||
# straight from feed data, which is exactly where this PR's other
|
||||
# crashes came from. One sport's bad payload must not take down the
|
||||
# shared orchestration for the others.
|
||||
self.logger.exception(
|
||||
"Error preparing scroll content for game_type=%s", game_type
|
||||
)
|
||||
return False
|
||||
if success:
|
||||
self._current_game_type = game_type
|
||||
return success
|
||||
|
||||
def display_frame(self, game_type: Optional[str] = None) -> bool:
|
||||
"""Advance the active strip (or a named one) by one frame."""
|
||||
game_type = game_type or self._current_game_type
|
||||
if not game_type:
|
||||
return False
|
||||
scroll_display = self._scroll_displays.get(game_type)
|
||||
if scroll_display is None:
|
||||
return False
|
||||
return scroll_display.display_scroll_frame()
|
||||
|
||||
def is_complete(self, game_type: Optional[str] = None) -> bool:
|
||||
"""True when the strip has finished — including when there isn't one,
|
||||
so a caller waiting on completion is never wedged."""
|
||||
game_type = game_type or self._current_game_type
|
||||
if not game_type:
|
||||
return True
|
||||
scroll_display = self._scroll_displays.get(game_type)
|
||||
if scroll_display is None:
|
||||
return True
|
||||
return scroll_display.is_scroll_complete()
|
||||
|
||||
def clear_all(self) -> None:
|
||||
"""Clear every display and forget which one was active."""
|
||||
for scroll_display in self._scroll_displays.values():
|
||||
scroll_display.clear()
|
||||
self._current_game_type = ""
|
||||
|
||||
def get_all_vegas_content_items(self) -> List[Image.Image]:
|
||||
"""Every display's Vegas items, for splicing into the marquee."""
|
||||
items: List[Image.Image] = []
|
||||
for scroll_display in self._scroll_displays.values():
|
||||
vegas_items = getattr(scroll_display, "_vegas_content_items", None)
|
||||
if vegas_items:
|
||||
items.extend(vegas_items)
|
||||
return items
|
||||
@@ -363,6 +363,20 @@ class TestExtractGameDetailsContract:
|
||||
assert details is not None
|
||||
assert details["inning"] == 7
|
||||
|
||||
def test_baseball_live_without_top_level_status_extracts_for_favorites(self):
|
||||
# The favourite-team branch logs the status payload for diagnostics and
|
||||
# read the same event top-level key the test above proves can be
|
||||
# absent. So the identical MiLB event that extracts fine for a
|
||||
# non-favourite raised KeyError and was dropped once the team WAS a
|
||||
# favourite -- the worst shape for the bug, since it only hit the games
|
||||
# the user cared most about, and only on the diagnostic path that was
|
||||
# supposed to help debug them.
|
||||
event = make_event("413", "in", "2026-07-16T23:05:00Z", period=7)
|
||||
del event["status"]
|
||||
details = extract(Baseball, event, favorites=["TB"])
|
||||
assert details is not None
|
||||
assert details["inning"] == 7
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 2. update() flow on concrete subclasses (offline, cache-fed)
|
||||
|
||||
@@ -86,6 +86,17 @@ class TestRegistry:
|
||||
with pytest.raises(ValueError):
|
||||
register_rotation_strategy("", SimpleRotation)
|
||||
|
||||
@pytest.mark.parametrize("bad", [
|
||||
SimpleRotation(), # an instance, not the class
|
||||
str, # unrelated class
|
||||
lambda **kw: None, # a factory function
|
||||
])
|
||||
def test_a_non_strategy_factory_is_rejected(self, bad):
|
||||
"""Fail at registration, not several frames later inside schedule(),
|
||||
where the cause is no longer on the stack."""
|
||||
with pytest.raises(TypeError):
|
||||
register_rotation_strategy("bad-factory", bad)
|
||||
|
||||
def test_weight_for_is_optional(self):
|
||||
"""Default weights are equal, so every strategy degenerates to a plain
|
||||
round robin — the pre-boost behavior."""
|
||||
@@ -111,6 +122,17 @@ class TestWeights:
|
||||
strategy = get_rotation_strategy("weighted", weight_for=lambda g: bad)
|
||||
assert strategy.weights([game("a")]) == {"a": 1}
|
||||
|
||||
def test_huge_weights_are_clamped(self):
|
||||
"""A cycle is sum(weights) long and each step scans every game, so an
|
||||
unbounded weight from a misread config spins the display thread."""
|
||||
strategy = get_rotation_strategy("weighted", weight_for=lambda g: 10_000)
|
||||
assert strategy.weights([game("a")]) == {"a": RotationStrategy.MAX_WEIGHT}
|
||||
|
||||
def test_a_clamped_cycle_stays_bounded(self):
|
||||
strategy = get_rotation_strategy("weighted", weight_for=lambda g: 10_000)
|
||||
order = strategy.schedule([game("a"), game("b")])
|
||||
assert len(order) == 2 * RotationStrategy.MAX_WEIGHT
|
||||
|
||||
|
||||
class TestSimpleRotation:
|
||||
def test_one_pass_in_feed_order(self):
|
||||
@@ -218,6 +240,20 @@ class TestSmoothWeightedRotation:
|
||||
actual = [strategy.next_game(games)["id"] for _ in range(len(preview))]
|
||||
assert preview == actual
|
||||
|
||||
def test_preview_uses_the_subclass_ordering(self):
|
||||
"""schedule() promises the order repeated next_game calls produce. Built
|
||||
from the base class, a subclass that overrides next_game gets a preview
|
||||
of the wrong algorithm."""
|
||||
class Reversed(SmoothWeightedRotation):
|
||||
def next_game(self, games):
|
||||
return super().next_game(list(reversed(games)))
|
||||
|
||||
games = [game("a"), game("b"), game("c")]
|
||||
strategy = Reversed()
|
||||
preview = strategy.schedule(games)
|
||||
actual = [strategy.next_game(games)["id"] for _ in range(len(preview))]
|
||||
assert preview == actual
|
||||
|
||||
def test_empty(self):
|
||||
assert SmoothWeightedRotation().next_game([]) is None
|
||||
assert SmoothWeightedRotation().schedule([]) == []
|
||||
@@ -462,6 +498,24 @@ class TestCelebrationConfig:
|
||||
"celebrate_opponent_goals": True})
|
||||
assert manager.celebrate_opponent_scores is False
|
||||
|
||||
@pytest.mark.parametrize("bad", ["eight", None, {}, []])
|
||||
def test_unusable_duration_falls_back(self, celebrating, bad):
|
||||
"""The duration is compared numerically on the display path, outside
|
||||
any try block — a string from a hand-edited config would propagate a
|
||||
TypeError straight out of display()."""
|
||||
manager = celebrating(mode_config={"celebration_duration": bad})
|
||||
assert manager.celebration_duration == 8.0
|
||||
|
||||
@pytest.mark.parametrize("bad", [0, -5])
|
||||
def test_non_positive_duration_is_floored(self, celebrating, bad):
|
||||
"""Zero or negative would arm a celebration that can never render."""
|
||||
manager = celebrating(mode_config={"celebration_duration": bad})
|
||||
assert manager.celebration_duration == 1.0
|
||||
|
||||
def test_numeric_string_duration_is_accepted(self, celebrating):
|
||||
assert celebrating(
|
||||
mode_config={"celebration_duration": "12"}).celebration_duration == 12.0
|
||||
|
||||
|
||||
class TestScoreDetection:
|
||||
def test_first_sighting_never_celebrates(self, celebrating):
|
||||
@@ -725,6 +779,52 @@ class TestDisplayTakeover:
|
||||
assert manager.display() is True
|
||||
assert manager.display_calls == [False]
|
||||
|
||||
def test_a_render_failure_disarms_rather_than_retrying(self, celebrating):
|
||||
"""Left armed, the same render fails on every frame for the rest of the
|
||||
window — a traceback per frame, and no scorebug."""
|
||||
manager = celebrating()
|
||||
manager._check_for_score(game("g1", home_score=0))
|
||||
manager._check_for_score(game("g1", home_score=1))
|
||||
manager._draw_celebration_layout = MagicMock(side_effect=RuntimeError("boom"))
|
||||
manager.display()
|
||||
assert manager.active_celebration is None
|
||||
manager.display()
|
||||
assert manager._draw_celebration_layout.call_count == 1
|
||||
|
||||
|
||||
class TestBaselinePruning:
|
||||
"""`_score_baselines` gains an entry per game and only _check_for_win ever
|
||||
removed one, so a board running all season grows the dict without bound."""
|
||||
|
||||
def test_prunes_games_no_longer_live(self, celebrating):
|
||||
manager = celebrating()
|
||||
for gid in ("g1", "g2", "g3"):
|
||||
manager._check_for_score(game(gid, home_score=1))
|
||||
manager.prune_score_baselines([game("g2")])
|
||||
assert set(manager._score_baselines) == {"g2"}
|
||||
|
||||
def test_keeps_every_still_live_game(self, celebrating):
|
||||
manager = celebrating()
|
||||
for gid in ("g1", "g2"):
|
||||
manager._check_for_score(game(gid, home_score=1))
|
||||
manager.prune_score_baselines([game("g1"), game("g2")])
|
||||
assert set(manager._score_baselines) == {"g1", "g2"}
|
||||
|
||||
def test_empty_live_set_clears_everything(self, celebrating):
|
||||
manager = celebrating()
|
||||
manager._check_for_score(game("g1", home_score=1))
|
||||
manager.prune_score_baselines([])
|
||||
assert manager._score_baselines == {}
|
||||
|
||||
def test_pruning_does_not_disturb_a_surviving_baseline(self, celebrating):
|
||||
manager = celebrating()
|
||||
manager._check_for_score(game("g1", home_score=2))
|
||||
manager.prune_score_baselines([game("g1")])
|
||||
manager._check_for_score(game("g1", home_score=3))
|
||||
assert manager.active_celebration is not None, (
|
||||
"pruning must not drop a live game's baseline and re-trigger the "
|
||||
"first-sighting suppression")
|
||||
|
||||
def test_disabled_manager_renders_nothing(self, celebrating):
|
||||
manager = celebrating()
|
||||
manager.is_enabled = False
|
||||
|
||||
@@ -494,9 +494,32 @@ class TestShouldLogCooldown:
|
||||
def test_cooldown_expires(self, build):
|
||||
manager = build()
|
||||
assert manager._should_log("api", cooldown=60) is True
|
||||
manager._last_warning_time -= 61
|
||||
manager._warning_cooldowns["api"] -= 61
|
||||
assert manager._should_log("api", cooldown=60) is True
|
||||
|
||||
def test_cooldowns_are_tracked_per_warning_type(self, build):
|
||||
"""The parameter was accepted and ignored: one shared timestamp meant an
|
||||
API warning silenced an unrelated cache warning for the next minute."""
|
||||
manager = build()
|
||||
assert manager._should_log("api", cooldown=60) is True
|
||||
assert manager._should_log("cache", cooldown=60) is True
|
||||
assert manager._should_log("api", cooldown=60) is False
|
||||
assert manager._should_log("cache", cooldown=60) is False
|
||||
|
||||
def test_one_type_expiring_does_not_free_another(self, build):
|
||||
manager = build()
|
||||
manager._should_log("api")
|
||||
manager._should_log("cache")
|
||||
manager._warning_cooldowns["api"] -= 61
|
||||
assert manager._should_log("api") is True
|
||||
assert manager._should_log("cache") is False
|
||||
|
||||
def test_legacy_single_clock_field_is_kept_in_step(self, build):
|
||||
"""Subclasses in the plugin copies read _last_warning_time directly."""
|
||||
manager = build()
|
||||
manager._should_log("api")
|
||||
assert manager._last_warning_time == manager._warning_cooldowns["api"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 4. Seam guard rails
|
||||
|
||||
@@ -0,0 +1,585 @@
|
||||
"""Tests for src/common/sports_scroll.py (phase B3).
|
||||
|
||||
The module is drawn along the split the survey found: the orchestration layer
|
||||
is shared, the content layer is not. Two things are worth asserting beyond
|
||||
"it works":
|
||||
|
||||
* ``prepare_scroll_content`` must stay an override point — a base class that
|
||||
quietly rendered *something* would let a plugin ship a blank scroll.
|
||||
* ``target_fps`` must actually reach the helper. That is the whole reason this
|
||||
module exists upstream; the bundled plugin copies hardcode ~100 FPS.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import sys
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
from PIL import Image
|
||||
|
||||
sys.modules.setdefault("rgbmatrix", MagicMock())
|
||||
|
||||
from src.common.sports_scroll import ( # noqa: E402
|
||||
DEFAULT_SCROLL_SETTINGS,
|
||||
MAX_PIXELS_PER_FRAME,
|
||||
MIN_PIXELS_PER_FRAME,
|
||||
SportsScrollDisplay,
|
||||
SportsScrollDisplayManager,
|
||||
)
|
||||
|
||||
LOGGER = logging.getLogger("test.sports_scroll")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def display_manager():
|
||||
manager = MagicMock()
|
||||
manager.matrix.width = 128
|
||||
manager.matrix.height = 32
|
||||
return manager
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def fake_scroll_helper(monkeypatch):
|
||||
"""Replace ScrollHelper with a recording double.
|
||||
|
||||
The real helper is covered by test_scroll_helper.py; here what matters is
|
||||
*which* calls this module makes and with what values.
|
||||
"""
|
||||
created = []
|
||||
|
||||
def _factory(width, height, logger):
|
||||
helper = MagicMock()
|
||||
helper.width, helper.height = width, height
|
||||
helper.cached_image = None
|
||||
helper.is_scroll_complete.return_value = False
|
||||
helper.get_dynamic_duration.return_value = 42
|
||||
helper.get_scroll_info.return_value = {"position": 0}
|
||||
created.append(helper)
|
||||
return helper
|
||||
|
||||
monkeypatch.setattr("src.common.sports_scroll.ScrollHelper", _factory)
|
||||
return created
|
||||
|
||||
|
||||
class _Display(SportsScrollDisplay):
|
||||
"""A minimal concrete subclass, as a plugin would write it."""
|
||||
|
||||
SCROLL_LEAGUE_KEYS = ("nhl", "ncaa_mens")
|
||||
|
||||
def prepare_scroll_content(self, games, game_type, leagues, rankings_cache=None):
|
||||
self._current_games = list(games)
|
||||
self._current_game_type = game_type
|
||||
self._current_leagues = list(leagues)
|
||||
self.scroll_helper.cached_image = Image.new("RGB", (400, 32))
|
||||
return bool(games)
|
||||
|
||||
|
||||
class _Manager(SportsScrollDisplayManager):
|
||||
display_class = _Display
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def build(display_manager):
|
||||
def _build(config=None, global_config=None, cls=_Display):
|
||||
return cls(display_manager, config or {}, LOGGER, global_config=global_config)
|
||||
return _build
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Construction
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestConstruction:
|
||||
def test_dimensions_come_from_the_matrix(self, build):
|
||||
display = build()
|
||||
assert (display.display_width, display.display_height) == (128, 32)
|
||||
|
||||
def test_dimensions_fall_back_when_there_is_no_matrix(self, display_manager):
|
||||
display_manager.matrix = None
|
||||
display_manager.width, display_manager.height = 256, 64
|
||||
display = _Display(display_manager, {}, LOGGER)
|
||||
assert (display.display_width, display.display_height) == (256, 64)
|
||||
|
||||
def test_final_fallback_dimensions(self):
|
||||
"""A display manager exposing neither must not crash construction."""
|
||||
bare = MagicMock(spec=[])
|
||||
display = _Display(bare, {}, LOGGER)
|
||||
assert (display.display_width, display.display_height) == (128, 32)
|
||||
|
||||
def test_global_config_is_optional(self, build):
|
||||
"""An older caller that doesn't pass it keeps working."""
|
||||
assert build().global_config == {}
|
||||
|
||||
def test_state_starts_empty(self, build):
|
||||
display = build()
|
||||
assert display._current_games == []
|
||||
assert display._vegas_content_items == []
|
||||
assert display.get_current_game_count() == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Settings resolution — the league ladder
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestScrollSettings:
|
||||
def test_defaults_when_nothing_is_configured(self, build):
|
||||
settings = build()._get_scroll_settings()
|
||||
for key, value in DEFAULT_SCROLL_SETTINGS.items():
|
||||
assert settings[key] == value
|
||||
|
||||
def test_card_width_defaults_to_the_panel_width(self, build):
|
||||
assert build()._get_scroll_settings()["game_card_width"] == 128
|
||||
|
||||
def test_named_league_wins(self, build):
|
||||
display = build({"nhl": {"scroll_settings": {"scroll_speed": 10}},
|
||||
"ncaa_mens": {"scroll_settings": {"scroll_speed": 20}}})
|
||||
assert display._get_scroll_settings("ncaa_mens")["scroll_speed"] == 20
|
||||
|
||||
def test_ladder_is_walked_in_order(self, build):
|
||||
"""The only reason the eight plugin copies differed: which league keys
|
||||
to try, and in what order."""
|
||||
display = build({"ncaa_mens": {"scroll_settings": {"scroll_speed": 20}},
|
||||
"nhl": {"scroll_settings": {"scroll_speed": 10}}})
|
||||
assert display._get_scroll_settings()["scroll_speed"] == 10
|
||||
|
||||
def test_ladder_falls_through_to_the_next_key(self, build):
|
||||
display = build({"ncaa_mens": {"scroll_settings": {"scroll_speed": 20}}})
|
||||
assert display._get_scroll_settings()["scroll_speed"] == 20
|
||||
|
||||
def test_overrides_merge_onto_defaults(self, build):
|
||||
display = build({"nhl": {"scroll_settings": {"scroll_speed": 10}}})
|
||||
settings = display._get_scroll_settings()
|
||||
assert settings["scroll_speed"] == 10
|
||||
assert settings["gap_between_games"] == DEFAULT_SCROLL_SETTINGS[
|
||||
"gap_between_games"]
|
||||
|
||||
def test_empty_override_does_not_shadow_the_next_candidate(self, build):
|
||||
display = build({"nhl": {"scroll_settings": {}},
|
||||
"ncaa_mens": {"scroll_settings": {"scroll_speed": 20}}})
|
||||
assert display._get_scroll_settings()["scroll_speed"] == 20
|
||||
|
||||
def test_single_block_config_shape(self, build):
|
||||
"""The afl/nrl/soccer shape: one scroll_mode block, no league concept."""
|
||||
class _Single(_Display):
|
||||
SCROLL_LEAGUE_KEYS = ()
|
||||
SCROLL_CONFIG_KEY = "scroll_mode"
|
||||
|
||||
display = build({"scroll_mode": {"scroll_speed": 33}}, cls=_Single)
|
||||
assert display._get_scroll_settings()["scroll_speed"] == 33
|
||||
|
||||
def test_defaults_are_overridable_by_a_subclass(self, build):
|
||||
class _Wide(_Display):
|
||||
def scroll_settings_defaults(self):
|
||||
return {**super().scroll_settings_defaults(),
|
||||
"gap_between_games": 24}
|
||||
|
||||
assert build(cls=_Wide)._get_scroll_settings()["gap_between_games"] == 24
|
||||
|
||||
def test_a_null_league_block_is_tolerated(self, build):
|
||||
"""`config['nhl'] = None` appears in hand-edited configs."""
|
||||
display = build({"nhl": None})
|
||||
assert (display._get_scroll_settings()["scroll_speed"]
|
||||
== DEFAULT_SCROLL_SETTINGS["scroll_speed"])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helper configuration — including the reason this module exists upstream
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestConfigureScrollHelper:
|
||||
def test_speed_is_converted_to_pixels_per_frame(self, build):
|
||||
display = build({"nhl": {"scroll_settings": {
|
||||
"scroll_speed": 100.0, "scroll_delay": 0.02}}})
|
||||
# 100 px/s * 0.02 s/frame = 2 px/frame
|
||||
display.scroll_helper.set_scroll_speed.assert_called_with(2.0)
|
||||
|
||||
def test_conversion_is_clamped_low(self, build):
|
||||
display = build({"nhl": {"scroll_settings": {
|
||||
"scroll_speed": 0.001, "scroll_delay": 0.001}}})
|
||||
display.scroll_helper.set_scroll_speed.assert_called_with(MIN_PIXELS_PER_FRAME)
|
||||
|
||||
def test_conversion_is_clamped_high(self, build):
|
||||
display = build({"nhl": {"scroll_settings": {
|
||||
"scroll_speed": 5000.0, "scroll_delay": 0.5}}})
|
||||
display.scroll_helper.set_scroll_speed.assert_called_with(MAX_PIXELS_PER_FRAME)
|
||||
|
||||
def test_zero_delay_assumes_a_pacing_instead_of_dividing_by_zero(self, build):
|
||||
display = build({"nhl": {"scroll_settings": {
|
||||
"scroll_speed": 100.0, "scroll_delay": 0}}})
|
||||
display.scroll_helper.set_scroll_speed.assert_called_with(1.0)
|
||||
|
||||
def test_frame_based_scrolling_is_enabled(self, build):
|
||||
build().scroll_helper.set_frame_based_scrolling.assert_called_once_with(True)
|
||||
|
||||
def test_dynamic_duration_settings_are_applied(self, build):
|
||||
display = build({"nhl": {"scroll_settings": {
|
||||
"dynamic_duration": False, "min_duration": 5, "max_duration": 50}}})
|
||||
_, kwargs = display.scroll_helper.set_dynamic_duration_settings.call_args
|
||||
assert kwargs["enabled"] is False
|
||||
assert kwargs["min_duration"] == 5
|
||||
assert kwargs["max_duration"] == 50
|
||||
|
||||
def test_target_fps_reaches_the_helper(self, build):
|
||||
"""The whole point of upstreaming: the bundled copies hardcode ~100 FPS
|
||||
via scroll_delay and never consult the global target."""
|
||||
display = build(global_config={"target_fps": 120})
|
||||
display.scroll_helper.set_target_fps.assert_called_once_with(120.0)
|
||||
|
||||
def test_legacy_key_is_honored(self, build):
|
||||
display = build(global_config={"scroll_target_fps": 90})
|
||||
display.scroll_helper.set_target_fps.assert_called_once_with(90.0)
|
||||
|
||||
def test_modern_key_wins_over_legacy(self, build):
|
||||
display = build(global_config={"target_fps": 120, "scroll_target_fps": 90})
|
||||
display.scroll_helper.set_target_fps.assert_called_once_with(120.0)
|
||||
|
||||
def test_absent_target_fps_leaves_config_pacing_alone(self, build):
|
||||
build().scroll_helper.set_target_fps.assert_not_called()
|
||||
|
||||
@pytest.mark.parametrize("bad", ["fast", None, {}, [], "", 0])
|
||||
def test_unusable_target_fps_degrades_instead_of_raising(self, build, bad):
|
||||
"""A malformed global config must cost the FPS target, not the display."""
|
||||
display = build(global_config={"target_fps": bad})
|
||||
display.scroll_helper.set_target_fps.assert_not_called()
|
||||
|
||||
def test_string_digits_are_accepted(self, build):
|
||||
display = build(global_config={"target_fps": "120"})
|
||||
display.scroll_helper.set_target_fps.assert_called_once_with(120.0)
|
||||
|
||||
@pytest.mark.parametrize("bad", [None, "fast", {}, []])
|
||||
def test_unusable_scroll_speed_degrades_instead_of_crashing(self, build, bad):
|
||||
"""`.get(key, default)` only helps when the key is *absent*. A key
|
||||
present with null reaches the arithmetic and raises inside __init__,
|
||||
taking the whole display down before it renders anything."""
|
||||
display = build({"nhl": {"scroll_settings": {"scroll_speed": bad}}})
|
||||
# 50.0 px/s * 0.01 s/frame == 0.5 px/frame, i.e. the default speed.
|
||||
display.scroll_helper.set_scroll_speed.assert_called_with(0.5)
|
||||
|
||||
@pytest.mark.parametrize("bad", [None, "slow", {}])
|
||||
def test_unusable_scroll_delay_degrades_instead_of_crashing(self, build, bad):
|
||||
display = build({"nhl": {"scroll_settings": {"scroll_delay": bad}}})
|
||||
display.scroll_helper.set_scroll_delay.assert_called_with(0.01)
|
||||
|
||||
def test_numeric_strings_are_accepted(self, build):
|
||||
display = build({"nhl": {"scroll_settings": {
|
||||
"scroll_speed": "100", "scroll_delay": "0.02"}}})
|
||||
display.scroll_helper.set_scroll_speed.assert_called_with(2.0)
|
||||
|
||||
def test_fps_clamping_is_left_to_the_helper(self, build):
|
||||
"""Deliberately not clamped here — a second copy of the range would
|
||||
drift from ScrollHelper.set_target_fps."""
|
||||
display = build(global_config={"target_fps": 5000})
|
||||
display.scroll_helper.set_target_fps.assert_called_once_with(5000.0)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# The content seam
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestContentIsAnOverridePoint:
|
||||
def test_base_refuses_to_render(self, display_manager):
|
||||
"""Eight plugins have eight different bodies for this; a base class that
|
||||
rendered *something* would let a plugin ship a silently blank scroll."""
|
||||
display = SportsScrollDisplay(display_manager, {}, LOGGER)
|
||||
with pytest.raises(NotImplementedError, match="prepare_scroll_content"):
|
||||
display.prepare_scroll_content([], "live", [])
|
||||
|
||||
def test_the_error_names_the_offending_class(self, display_manager):
|
||||
class Incomplete(SportsScrollDisplay):
|
||||
pass
|
||||
|
||||
with pytest.raises(NotImplementedError, match="Incomplete"):
|
||||
Incomplete(display_manager, {}, LOGGER).prepare_scroll_content(
|
||||
[], "live", [])
|
||||
|
||||
def test_separator_icons_default_to_a_no_op(self, build):
|
||||
assert build()._separator_icons == {}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Frame pumping
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestFramePumping:
|
||||
def test_no_content_means_no_frame(self, build):
|
||||
assert build().display_scroll_frame() is False
|
||||
|
||||
def test_a_frame_is_drawn_and_pushed(self, build):
|
||||
display = build()
|
||||
display.prepare_scroll_content([{"id": "g1"}], "live", ["nhl"])
|
||||
display.scroll_helper.get_visible_portion.return_value = Image.new(
|
||||
"RGB", (128, 32))
|
||||
assert display.display_scroll_frame() is True
|
||||
display.scroll_helper.update_scroll_position.assert_called_once()
|
||||
display.display_manager.update_display.assert_called_once()
|
||||
|
||||
def test_no_visible_portion_means_no_frame(self, build):
|
||||
display = build()
|
||||
display.prepare_scroll_content([{"id": "g1"}], "live", ["nhl"])
|
||||
display.scroll_helper.get_visible_portion.return_value = None
|
||||
assert display.display_scroll_frame() is False
|
||||
|
||||
def test_a_display_failure_is_contained(self, build):
|
||||
"""A display error must not propagate into the plugin's render loop."""
|
||||
display = build()
|
||||
display.prepare_scroll_content([{"id": "g1"}], "live", ["nhl"])
|
||||
display.scroll_helper.get_visible_portion.return_value = Image.new(
|
||||
"RGB", (128, 32))
|
||||
display.display_manager.update_display.side_effect = RuntimeError("boom")
|
||||
assert display.display_scroll_frame() is False
|
||||
|
||||
@pytest.mark.parametrize("failing", ["update_scroll_position",
|
||||
"get_visible_portion"])
|
||||
def test_a_scroll_helper_failure_is_contained_too(self, build, failing):
|
||||
"""These ran outside the try, so a raise there reached the caller's
|
||||
frame loop despite the stated promise that none can."""
|
||||
display = build()
|
||||
display.prepare_scroll_content([{"id": "g1"}], "live", ["nhl"])
|
||||
getattr(display.scroll_helper, failing).side_effect = RuntimeError("boom")
|
||||
assert display.display_scroll_frame() is False
|
||||
|
||||
def test_frames_are_counted(self, build):
|
||||
display = build()
|
||||
display.prepare_scroll_content([{"id": "g1"}], "live", ["nhl"])
|
||||
display.scroll_helper.get_visible_portion.return_value = Image.new(
|
||||
"RGB", (128, 32))
|
||||
for _ in range(3):
|
||||
display.display_scroll_frame()
|
||||
assert display._frame_count == 3
|
||||
|
||||
def test_progress_logging_is_throttled(self, build):
|
||||
display = build()
|
||||
display._log_interval = 10_000
|
||||
display._last_log_time = 0
|
||||
display._log_scroll_progress()
|
||||
first = display._last_log_time
|
||||
display._log_scroll_progress()
|
||||
assert display._last_log_time == first
|
||||
|
||||
|
||||
class TestLifecycle:
|
||||
def test_reset_keeps_content_but_rewinds(self, build):
|
||||
display = build()
|
||||
display.prepare_scroll_content([{"id": "g1"}], "live", ["nhl"])
|
||||
display._frame_count = 9
|
||||
display.reset_scroll()
|
||||
display.scroll_helper.reset_scroll.assert_called_once()
|
||||
assert display._frame_count == 0
|
||||
assert display._current_games, "reset must not drop content"
|
||||
|
||||
def test_clear_drops_everything(self, build):
|
||||
display = build()
|
||||
display.prepare_scroll_content([{"id": "g1"}], "live", ["nhl"])
|
||||
display._vegas_content_items = [Image.new("RGB", (8, 8))]
|
||||
display.clear()
|
||||
display.scroll_helper.clear_cache.assert_called_once()
|
||||
assert display._current_games == []
|
||||
assert display._current_game_type == ""
|
||||
assert display._vegas_content_items == []
|
||||
|
||||
def test_completion_delegates_to_the_helper(self, build):
|
||||
display = build()
|
||||
display.scroll_helper.is_scroll_complete.return_value = True
|
||||
assert display.is_scroll_complete() is True
|
||||
|
||||
def test_dynamic_duration_delegates(self, build):
|
||||
assert build().get_dynamic_duration() == 42
|
||||
|
||||
def test_has_cached_content(self, build):
|
||||
display = build()
|
||||
assert display.has_cached_content() is False
|
||||
display.prepare_scroll_content([{"id": "g1"}], "live", ["nhl"])
|
||||
assert display.has_cached_content() is True
|
||||
|
||||
def test_scroll_info_merges_helper_and_local_state(self, build):
|
||||
display = build()
|
||||
display.prepare_scroll_content([{"id": "g1"}], "live", ["nhl"])
|
||||
info = display.get_scroll_info()
|
||||
assert info["position"] == 0 # from the helper
|
||||
assert info["game_count"] == 1 # from this display
|
||||
assert info["leagues"] == ["nhl"]
|
||||
|
||||
def test_leagues_are_returned_as_a_copy(self, build):
|
||||
display = build()
|
||||
display.prepare_scroll_content([{"id": "g1"}], "live", ["nhl"])
|
||||
display.get_current_leagues().append("mutated")
|
||||
assert display.get_current_leagues() == ["nhl"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Manager
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestManager:
|
||||
@pytest.fixture
|
||||
def manager(self, display_manager):
|
||||
return _Manager(display_manager, {}, LOGGER, global_config={"target_fps": 120})
|
||||
|
||||
def test_displays_are_created_lazily_and_reused(self, manager):
|
||||
first = manager.get_scroll_display("live")
|
||||
assert manager.get_scroll_display("live") is first
|
||||
assert isinstance(first, _Display)
|
||||
|
||||
def test_each_game_type_gets_its_own(self, manager):
|
||||
assert manager.get_scroll_display("live") is not manager.get_scroll_display(
|
||||
"recent")
|
||||
|
||||
def test_global_config_is_threaded_to_children(self, manager):
|
||||
"""A missed hand-off here is exactly how the plugin copies ended up
|
||||
never honoring target_fps."""
|
||||
child = manager.get_scroll_display("live")
|
||||
assert child.global_config == {"target_fps": 120}
|
||||
child.scroll_helper.set_target_fps.assert_called_once_with(120.0)
|
||||
|
||||
def test_prepare_sets_the_active_type(self, manager):
|
||||
assert manager.prepare_and_display([{"id": "g1"}], "live", ["nhl"]) is True
|
||||
assert manager._current_game_type == "live"
|
||||
|
||||
def test_failed_prepare_does_not_become_active(self, manager):
|
||||
assert manager.prepare_and_display([], "live", ["nhl"]) is False
|
||||
assert not manager._current_game_type
|
||||
|
||||
def test_a_raising_subclass_does_not_escape_the_orchestration(self, manager):
|
||||
"""prepare_scroll_content is subclass code building cards from feed
|
||||
data. One sport's bad payload must not take down the others."""
|
||||
display = manager.get_scroll_display("live")
|
||||
display.prepare_scroll_content = MagicMock(side_effect=KeyError("status"))
|
||||
assert manager.prepare_and_display([{"id": "g1"}], "live", ["nhl"]) is False
|
||||
assert not manager._current_game_type
|
||||
|
||||
def test_empty_game_type_sentinel_matches_the_display(self, manager):
|
||||
"""Both classes must spell 'nothing active' the same way; two spellings
|
||||
across two classes is a trap for anyone comparing their state."""
|
||||
manager.prepare_and_display([{"id": "g1"}], "live", ["nhl"])
|
||||
manager.clear_all()
|
||||
assert (manager._current_game_type
|
||||
== manager.get_scroll_display("live")._current_game_type == "")
|
||||
|
||||
def test_display_frame_uses_the_active_type(self, manager):
|
||||
manager.prepare_and_display([{"id": "g1"}], "live", ["nhl"])
|
||||
display = manager.get_scroll_display("live")
|
||||
display.scroll_helper.get_visible_portion.return_value = Image.new(
|
||||
"RGB", (128, 32))
|
||||
assert manager.display_frame() is True
|
||||
|
||||
def test_display_frame_with_no_active_type(self, manager):
|
||||
assert manager.display_frame() is False
|
||||
|
||||
def test_display_frame_for_an_unknown_type(self, manager):
|
||||
assert manager.display_frame("never-prepared") is False
|
||||
|
||||
def test_completion_is_true_when_there_is_nothing_to_scroll(self, manager):
|
||||
"""A caller waiting on completion must never be wedged by absence."""
|
||||
assert manager.is_complete() is True
|
||||
assert manager.is_complete("never-prepared") is True
|
||||
|
||||
def test_completion_delegates_to_the_active_display(self, manager):
|
||||
manager.prepare_and_display([{"id": "g1"}], "live", ["nhl"])
|
||||
manager.get_scroll_display("live").scroll_helper \
|
||||
.is_scroll_complete.return_value = True
|
||||
assert manager.is_complete() is True
|
||||
|
||||
def test_clear_all_clears_every_display(self, manager):
|
||||
manager.prepare_and_display([{"id": "g1"}], "live", ["nhl"])
|
||||
manager.prepare_and_display([{"id": "g2"}], "recent", ["nhl"])
|
||||
manager.clear_all()
|
||||
assert not manager._current_game_type
|
||||
for game_type in ("live", "recent"):
|
||||
assert manager.get_scroll_display(game_type)._current_games == []
|
||||
|
||||
def test_vegas_items_are_collected_across_displays(self, manager):
|
||||
for game_type in ("live", "recent"):
|
||||
manager.get_scroll_display(game_type)._vegas_content_items = [
|
||||
Image.new("RGB", (8, 8))]
|
||||
assert len(manager.get_all_vegas_content_items()) == 2
|
||||
|
||||
def test_vegas_collection_tolerates_empty_displays(self, manager):
|
||||
manager.get_scroll_display("live")
|
||||
assert manager.get_all_vegas_content_items() == []
|
||||
|
||||
def test_display_class_is_the_subclass_seam(self, display_manager):
|
||||
class Other(_Display):
|
||||
pass
|
||||
|
||||
class OtherManager(SportsScrollDisplayManager):
|
||||
display_class = Other
|
||||
|
||||
manager = OtherManager(display_manager, {}, LOGGER)
|
||||
assert isinstance(manager.get_scroll_display("live"), Other)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Integration — against the real ScrollHelper
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class _RealDisplay(SportsScrollDisplay):
|
||||
"""Builds a strip the way a plugin would, via the real helper API."""
|
||||
|
||||
SCROLL_LEAGUE_KEYS = ("nhl",)
|
||||
|
||||
def prepare_scroll_content(self, games, game_type, leagues, rankings_cache=None):
|
||||
settings = self._get_scroll_settings()
|
||||
self.scroll_helper.create_scrolling_image(
|
||||
[Image.new("RGB", (100, 32), (20, 20, 20)) for _ in games],
|
||||
item_gap=settings["gap_between_games"],
|
||||
)
|
||||
self._current_games = list(games)
|
||||
self._current_game_type = game_type
|
||||
self._current_leagues = list(leagues)
|
||||
return bool(games)
|
||||
|
||||
|
||||
class _RealManager(SportsScrollDisplayManager):
|
||||
display_class = _RealDisplay
|
||||
|
||||
|
||||
class TestAgainstTheRealScrollHelper:
|
||||
"""The mocked tests above pin *which* calls this module makes; these pin
|
||||
that those calls exist and mean what we think. Without this, a rename in
|
||||
ScrollHelper would sail past a suite built entirely on MagicMock."""
|
||||
|
||||
@pytest.fixture
|
||||
def real(self, display_manager, monkeypatch):
|
||||
from src.common.scroll_helper import ScrollHelper
|
||||
monkeypatch.setattr("src.common.sports_scroll.ScrollHelper", ScrollHelper)
|
||||
return _RealManager(
|
||||
display_manager,
|
||||
{"nhl": {"scroll_settings": {"scroll_speed": 500.0,
|
||||
"scroll_delay": 0.001}}},
|
||||
LOGGER,
|
||||
global_config={"target_fps": 120},
|
||||
)
|
||||
|
||||
def test_configuration_lands_on_the_real_helper(self, real):
|
||||
helper = real.get_scroll_display("live").scroll_helper
|
||||
assert helper.target_fps == 120.0
|
||||
assert helper.frame_based_scrolling is True
|
||||
assert helper.scroll_speed == pytest.approx(0.5) # 500 px/s * 0.001 s
|
||||
|
||||
def test_a_strip_is_built_and_scrolls_to_completion(self, real):
|
||||
import time
|
||||
|
||||
assert real.prepare_and_display(
|
||||
[{"id": "a"}, {"id": "b"}, {"id": "c"}], "live", ["nhl"]) is True
|
||||
display = real.get_scroll_display("live")
|
||||
# 3 cards of 100px + gaps, so the strip is wider than the 128px panel.
|
||||
assert display.scroll_helper.cached_image.width > 128
|
||||
|
||||
# Frame-based scrolling is wall-clock gated on scroll_delay, so the
|
||||
# loop has to actually pass time rather than spin.
|
||||
deadline = time.time() + 20
|
||||
while not real.is_complete() and time.time() < deadline:
|
||||
real.display_frame()
|
||||
time.sleep(0.0012)
|
||||
|
||||
# Asserted separately so a host too slow to sustain the frame rate
|
||||
# reports a timeout rather than looking like a scrolling defect.
|
||||
assert time.time() < deadline, (
|
||||
"scroll did not finish within 20s — the host may be too slow to "
|
||||
"sustain the configured frame rate")
|
||||
assert real.is_complete() is True
|
||||
assert display.scroll_helper.scroll_position > 128
|
||||
|
||||
def test_dynamic_duration_is_a_real_number(self, real):
|
||||
real.prepare_and_display([{"id": "a"}], "live", ["nhl"])
|
||||
assert real.get_scroll_display("live").get_dynamic_duration() > 0
|
||||
Reference in New Issue
Block a user