feat(scroll): upstream the scroll orchestration layer; release 3.2.0

Phases B3 and B4.

B3 -- src/common/sports_scroll.py is deliberately NOT a superset of the
ten plugin 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, and the module is drawn along it:

  promoted   orchestration -- get_all_vegas_content_items is identical
             in all eight; clear_all, get_scroll_info,
             get_dynamic_duration, is_complete and display_frame are
             96-100% similar
  promoted   settings -- one algorithm; the copies differ only in which
             league keys they walk, so the ladder is data
             (SCROLL_LEAGUE_KEYS) rather than a body per sport
  NOT        content -- prepare_scroll_content has 8 distinct bodies
             across 8 plugins (145 lines, 53% similar at worst) and
             _load_separator_icons 7 (6% at worst)

Same name, different job: prepare_scroll_content draws *this sport's*
game card. Merging those eight bodies would be exactly the mistake the
promotion rule exists to prevent, so the base raises NotImplementedError
rather than rendering something plausible -- a base that rendered
something would let a plugin ship a silently blank scroll.

The one behavior added over the plugin copies is native
global_config['target_fps'] support. The bundled copies hardcode ~100
FPS via scroll_delay and never consult the global target; Part A
threaded it through each copy by hand, and this makes that threading
legacy compatibility rather than the mechanism.

66 tests, including three against the real ScrollHelper rather than a
double -- a suite built entirely on MagicMock would sail straight past a
rename in the helper.

B4 -- bump src/__init__.py to 3.2.0 and close the CHANGELOG's Unreleased
section against it. This is the number the sunset rule keys on: the
first core release shipping the unified sports library, and therefore
the floor a plugin sets ledmatrix_min_version to before deleting its
bundled copies. The version bump and the changelog release heading move
together on purpose -- separating them would leave a commit whose
changelog announces 3.2.0 while the code still reports 3.1.0.

Nothing here changes what an existing plugin loads; adoption is B5.

Gates: 714 core unit, 66 plugin safety.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FgbA8SMutQQpXkMG8LMmC4
This commit is contained in:
Claude
2026-08-01 17:02:20 +00:00
parent 969863eebd
commit 2d6ff2bb88
6 changed files with 1040 additions and 5 deletions
+2 -1
View File
@@ -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
View File
@@ -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
+29 -1
View File
@@ -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
@@ -171,6 +175,30 @@ 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 96100% 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,7 +206,7 @@ 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 |
+1 -1
View File
@@ -4,5 +4,5 @@ LEDMatrix Display System
Core source package for the LED Matrix Display project.
"""
__version__ = "3.1.0"
__version__ = "3.2.0"
+449
View File
@@ -0,0 +1,449 @@
"""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 _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 = settings.get("scroll_speed", 50.0)
scroll_delay = settings.get("scroll_delay", 0.01)
dynamic_duration = 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
self.scroll_helper.update_scroll_position()
visible = self.scroll_helper.get_visible_portion()
if not visible:
return False
try:
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] = {}
self._current_game_type: Optional[str] = None
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)
success = scroll_display.prepare_scroll_content(
games, game_type, leagues, rankings_cache
)
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 = None
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
+534
View File
@@ -0,0 +1,534 @@
"""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"] == 50.0
# ---------------------------------------------------------------------------
# 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)
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
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 manager._current_game_type is None
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 manager._current_game_type is None
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)
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