Compare commits

..
5 Commits
Author SHA1 Message Date
ChuckBuildsandClaude Sonnet 5 2eea7a7f58 fix(baseball): don't drop favourite MiLB games on the diagnostic path
The competition-level status fallback fixed the inning lookup, but the
favourite-team debug block a few lines above still read the event top-level
game_event["status"]. MiLB events (synthesized from the MLB Stats API into an
ESPN-like shape) populate only the competition-level status, so the identical
event that extracted fine for a non-favourite raised KeyError and returned
None once the team was a favourite.

Worst possible shape for the bug: it only hit the games the user cared most
about, and only on the path meant to help diagnose them. The existing
regression test missed it because it never passes favourites, so
is_favorite_game was False and the block never ran.

Uses the validated competition-level `status`, which
_extract_game_details_common guarantees is present by that point. Adds a
favourites-passing companion test; confirmed it reproduces the KeyError
without the fix.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KEZK1P1Q1fu5pcuVrkrCFZ
2026-08-01 15:19:45 -04:00
Claude 274ea65b39 fix(sports): harden the B2/B3 capabilities against bad config and subclasses
Review pass on the phase B2-B4 changes. Every fix here is the same shape
as the crashes this PR already fixed in the hockey and baseball
extractors: a config or feed value that is present-but-wrong reaching
arithmetic or a comparison on a path with no guard.

celebrations:
- celebration_duration is coerced and floored at init. It is compared
  numerically in display() *outside* any try block, so a string from a
  hand-edited config propagated a TypeError straight out; zero or
  negative armed a celebration that could never render.
- A render failure now disarms instead of staying armed. It previously
  retried the same broken render on every frame for the rest of the
  window -- a traceback per frame, and no scorebug either.
- prune_score_baselines() for the live set. Only _check_for_win removed
  entries, so a game that left the live list any other way leaked its
  baseline and the dict grew all season.
- display() reuses has_active_celebration() rather than repeating its
  window comparison, and log lines carry a [Celebrations] prefix.

rotation:
- MAX_WEIGHT ceiling. A cycle is sum(weights) long and each step scans
  every game, so an unbounded weight from a misread config spins the
  display thread -- on a Pi that stalls rendering outright.
- register_rotation_strategy rejects a non-subclass factory at
  registration instead of failing frames later inside schedule().
- schedule() previews through type(self), so a subclass overriding
  next_game is previewed with its own ordering -- which is what the
  method promises.

sports_scroll:
- scroll_speed / scroll_delay coerced. dict.get(key, default) only helps
  when the key is absent; present-but-null reached the multiplication
  inside __init__ and the display failed to construct at all.
- update_scroll_position and get_visible_portion moved inside the try.
  They ran outside it, so a raise there reached the plugin's frame loop
  despite the comment promising none can.
- prepare_and_display guards the subclass call, so one sport's bad
  payload cannot take down the shared orchestration for the others.
- _current_game_type spells "nothing active" as "" in both classes; the
  manager said None while the display said "".

Not taken: the report that baseball's favorite-team debug path still
reads event-level status. Verified against current code -- there are no
remaining game_event["status"] reads in that file; it was fixed in
2486bdb and the finding is stale.

Gates: 747 core unit, 66 plugin safety.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FgbA8SMutQQpXkMG8LMmC4
2026-08-01 17:46:06 +00:00
Claude 6522c6d525 docs: record the validated hockey scroll-display pilot for B5
B5 cannot ship until this PR merges and 3.2.0 exists -- 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 pilot has been validated ahead of that gate: hockey's
scroll_display.py adopted against a core carrying 3.2.0 goes from 691 to
289 lines with all 16 harness renders byte-for-byte identical.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FgbA8SMutQQpXkMG8LMmC4
2026-08-01 17:40:02 +00:00
Claude aeb99d7f09 fix(sports): per-type warning cooldowns and font-load logging
Follow-ups from the second review pass, both on already-fixed findings:

_should_log accepted a warning_type and ignored it, sharing one
timestamp across every kind of warning -- so an API-error warning
silenced an unrelated cache warning for the next minute, and whichever
fired first won. Cooldowns are now keyed by type. Nothing in core calls
this method, so no behavior regressed; _last_warning_time is kept in
step for subclasses that read it directly.

_load_fonts logged through the module-level logger, dropping the manager
context, and had no return type hint. It now uses self.logger (set well
before _load_fonts runs) and names the directory it searched -- the bare
"Fonts not found" sent people hunting for a font-format problem when the
actual cause is an install missing assets/fonts.

Gates: 717 core unit, 66 plugin safety.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FgbA8SMutQQpXkMG8LMmC4
2026-08-01 17:37:51 +00:00
Claude 2d6ff2bb88 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
2026-08-01 17:02:20 +00:00
13 changed files with 1382 additions and 31 deletions
+2 -1
View File
@@ -75,4 +75,5 @@ jobs:
test/test_element_style.py \ test/test_element_style.py \
test/test_sports_core_promotions.py \ test/test_sports_core_promotions.py \
test/test_sports_modes_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 `src.*`, note it in the Unreleased section and bump `src/__init__.py` in the
release that ships it. 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 ### Added
- `src/element_style.py` — per-element style resolver backing the - `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 system, data sources, API extractors, scroll helper, adaptive layout, loader
compatibility warning) plus new characterization tests for compatibility warning) plus new characterization tests for
`src/base_classes/sports.py` ahead of the shared sports-code unification. `src/base_classes/sports.py` ahead of the shared sports-code unification.
- `src/base_classes/sports/` — `sports.py` is now a package (`core.py` + - `src/base_classes/sports/` — `sports.py` is now a package (`core.py` +
`modes.py`). The import path is unchanged: `from src.base_classes.sports `modes.py`). The import path is unchanged: `from src.base_classes.sports
import SportsCore` still works. import SportsCore` still works.
@@ -49,7 +62,17 @@ release that ships it.
built-in is verified against a verbatim transcription of the plugin built-in is verified against a verbatim transcription of the plugin
implementation it replaces. An unknown name degrades to `simple`. 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 ### 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.** - **Live games are no longer dropped when the feed omits a game clock.**
`SportsLive._is_game_really_over` previously (in the baseball and UFC `SportsLive._is_game_really_over` previously (in the baseball and UFC
plugin lineages) coerced a missing or non-string clock to the literal plugin lineages) coerced a missing or non-string clock to the literal
+44 -5
View File
@@ -76,6 +76,10 @@ src/base_classes/sports/
capabilities/ capabilities/
celebrations.py CelebrationMixin (opt-in: 4 of 9 plugins) celebrations.py CelebrationMixin (opt-in: 4 of 9 plugins)
rotation.py RotationStrategy + registry 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 `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 `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 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, 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 and a typo should cost the boost, not the scoreboard. When a plugin needs an
ordering core does not ship calls `register_rotation_strategy` instead of core ordering that core does not ship, it calls `register_rotation_strategy` to add
growing a branch. its own — rather than core growing a branch for it.
`test_sports_capabilities.py` checks each strategy against a **verbatim `test_sports_capabilities.py` checks each strategy against a **verbatim
transcription** of the plugin code it replaces, over every live-game shape up to 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 four games. That differential is what B5 deletes the bundled copies on the
strength of. 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 ## Phases
| Phase | Scope | Risk control | | Phase | Scope | Risk control |
@@ -178,9 +206,20 @@ strength of.
| **B0** ✅ | Characterization tests, CI unit job, `element_style`, font cwd fix, CHANGELOG discipline | — | | **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 | | **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 | | **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 | | **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 ## Rules for contributors
+1 -1
View File
@@ -4,5 +4,5 @@ LEDMatrix Display System
Core source package for the LED Matrix Display project. Core source package for the LED Matrix Display project.
""" """
__version__ = "3.1.0" __version__ = "3.2.0"
+6 -1
View File
@@ -151,7 +151,12 @@ class Baseball(SportsCore):
# Only log detailed information for favorite teams # Only log detailed information for favorite teams
if is_favorite_game: 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 type: {game_status}, State: {status_state}")
self.logger.debug(f"Status detail: {status['type'].get('detail', '')}") self.logger.debug(f"Status detail: {status['type'].get('detail', '')}")
self.logger.debug( self.logger.debug(
@@ -51,7 +51,20 @@ class CelebrationMixin:
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)
mode_config = getattr(self, "mode_config", {}) or {} mode_config = getattr(self, "mode_config", {}) or {}
self.celebration_enabled = mode_config.get("celebration_enabled", True) 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`, # Both spellings: the soccer lineage ships `celebrate_opponent_goals`,
# football ships `celebrate_opponent_scores`. Whichever the plugin's # football ships `celebrate_opponent_scores`. Whichever the plugin's
# schema declares is the one its users have set. # 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. # Favorites exist but this team isn't one -> it's the opponent.
return self.celebrate_opponent_scores 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: def has_active_celebration(self) -> bool:
"""True while a celebration is within its display window.""" """True while a celebration is within its display window."""
celebration = self.active_celebration celebration = self.active_celebration
@@ -255,7 +287,7 @@ class CelebrationMixin:
# resumes on it. # resumes on it.
self.current_game = dict(game) self.current_game = dict(game)
self.logger.info( 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')}]" 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 away_logo, (-2, center_y - away_logo.height // 2), away_logo
) )
except Exception as e: 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 across the top, shrunk to fit the panel width.
phrase = celebration["phrase"] phrase = celebration["phrase"]
@@ -365,12 +397,19 @@ class CelebrationMixin:
return False return False
celebration = self.active_celebration celebration = self.active_celebration
if celebration: if celebration:
if time.time() - celebration["started_at"] < self.celebration_duration: if self.has_active_celebration():
try: try:
self._draw_celebration_layout(celebration, force_clear) self._draw_celebration_layout(celebration, force_clear)
return True return True
except Exception as e: 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: else:
self.active_celebration = None self.active_celebration = None
# Reset the dwell so the scorebug resumes on the scoring/winning # 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 this strategy is registered under. Set by :func:`register_rotation_strategy`.
name: str = "" 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): def __init__(self, weight_for: Optional[Callable[[Dict], int]] = None):
self._weight_for = weight_for or (lambda game: 1) 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 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 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] = {} weights: Dict[str, int] = {}
for game in games: for game in games:
@@ -69,7 +76,7 @@ class RotationStrategy:
weight = int(self._weight_for(game)) weight = int(self._weight_for(game))
except (TypeError, ValueError): except (TypeError, ValueError):
weight = 1 weight = 1
weights[gid] = max(1, weight) weights[gid] = min(self.MAX_WEIGHT, max(1, weight))
return weights return weights
def schedule(self, games: List[Dict]) -> List[str]: def schedule(self, games: List[Dict]) -> List[str]:
@@ -183,7 +190,11 @@ class SmoothWeightedRotation(RotationStrategy):
weights = self.weights(games) weights = self.weights(games)
if not weights: if not weights:
return [] 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) preview._current = dict(self._current)
order: List[str] = [] order: List[str] = []
for _ in range(sum(weights.values())): 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: def register_rotation_strategy(name: str, factory: Type[RotationStrategy]) -> None:
"""Register a rotation strategy under ``name``. """Register a rotation strategy under ``name``.
A plugin needing an ordering core does not ship registers it here rather When a plugin needs an ordering that core does not ship, it registers its
than core growing a sport-specific branch. Re-registering a name replaces own here instead of core growing a sport-specific branch. Re-registering a
it, so a plugin may also override a built-in for itself. name replaces it, so a plugin may also override a built-in for itself.
""" """
if not name: if not name:
raise ValueError("rotation strategy name must be a non-empty string") 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 factory.name = name
_REGISTRY[name] = factory _REGISTRY[name] = factory
+28 -9
View File
@@ -185,9 +185,12 @@ class SportsCore(ABC):
} }
self.last_update = 0 self.last_update = 0
self.current_game = None self.current_game = None
# Cooldown clock for _should_log(). Initialized here rather than lazily: # Cooldown clocks for _should_log(), one per warning type. Initialized
# _should_log() reads it unguarded, so whichever warning fires first # here rather than lazily: _should_log() reads them unguarded, so
# would otherwise raise AttributeError instead of logging. # 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._last_warning_time = 0
self.fonts = self._load_fonts() self.fonts = self._load_fonts()
@@ -428,7 +431,7 @@ class SportsCore(ABC):
return False return False
def _load_fonts(self): def _load_fonts(self) -> Dict[str, Any]:
"""Load fonts used by the scoreboard. """Load fonts used by the scoreboard.
Paths go through :meth:`_resolve_font_path` so the bundled fonts are 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 default font whenever the process started elsewhere (the plugin safety
harness on CI being the case that surfaced it). harness on CI being the case that surfaced it).
""" """
fonts = {} fonts: Dict[str, Any] = {}
press_start = self._resolve_font_path("PressStart2P-Regular.ttf") press_start = self._resolve_font_path("PressStart2P-Regular.ttf")
four_by_six = self._resolve_font_path("4x6-font.ttf") four_by_six = self._resolve_font_path("4x6-font.ttf")
try: try:
@@ -447,9 +450,15 @@ class SportsCore(ABC):
fonts['status'] = ImageFont.truetype(four_by_six, 6) # Using 4x6 for status fonts['status'] = ImageFont.truetype(four_by_six, 6) # Using 4x6 for status
fonts['detail'] = ImageFont.truetype(four_by_six, 6) # Added detail font fonts['detail'] = ImageFont.truetype(four_by_six, 6) # Added detail font
fonts['rank'] = ImageFont.truetype(press_start, 10) fonts['rank'] = ImageFont.truetype(press_start, 10)
logging.info("Successfully loaded fonts") # Changed log prefix self.logger.info("Successfully loaded fonts")
except OSError: 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['score'] = ImageFont.load_default()
fonts['time'] = ImageFont.load_default() fonts['time'] = ImageFont.load_default()
fonts['team'] = ImageFont.load_default() fonts['team'] = ImageFont.load_default()
@@ -634,9 +643,19 @@ class SportsCore(ABC):
return pytz.utc return pytz.utc
def _should_log(self, warning_type: str, cooldown: int = 60) -> bool: 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() 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 self._last_warning_time = current_time
return True return True
return False return False
+485
View File
@@ -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
+14
View File
@@ -363,6 +363,20 @@ class TestExtractGameDetailsContract:
assert details is not None assert details is not None
assert details["inning"] == 7 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) # 2. update() flow on concrete subclasses (offline, cache-fed)
+100
View File
@@ -86,6 +86,17 @@ class TestRegistry:
with pytest.raises(ValueError): with pytest.raises(ValueError):
register_rotation_strategy("", SimpleRotation) 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): def test_weight_for_is_optional(self):
"""Default weights are equal, so every strategy degenerates to a plain """Default weights are equal, so every strategy degenerates to a plain
round robin — the pre-boost behavior.""" round robin — the pre-boost behavior."""
@@ -111,6 +122,17 @@ class TestWeights:
strategy = get_rotation_strategy("weighted", weight_for=lambda g: bad) strategy = get_rotation_strategy("weighted", weight_for=lambda g: bad)
assert strategy.weights([game("a")]) == {"a": 1} 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: class TestSimpleRotation:
def test_one_pass_in_feed_order(self): 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))] actual = [strategy.next_game(games)["id"] for _ in range(len(preview))]
assert preview == actual 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): def test_empty(self):
assert SmoothWeightedRotation().next_game([]) is None assert SmoothWeightedRotation().next_game([]) is None
assert SmoothWeightedRotation().schedule([]) == [] assert SmoothWeightedRotation().schedule([]) == []
@@ -462,6 +498,24 @@ class TestCelebrationConfig:
"celebrate_opponent_goals": True}) "celebrate_opponent_goals": True})
assert manager.celebrate_opponent_scores is False 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: class TestScoreDetection:
def test_first_sighting_never_celebrates(self, celebrating): def test_first_sighting_never_celebrates(self, celebrating):
@@ -725,6 +779,52 @@ class TestDisplayTakeover:
assert manager.display() is True assert manager.display() is True
assert manager.display_calls == [False] 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): def test_disabled_manager_renders_nothing(self, celebrating):
manager = celebrating() manager = celebrating()
manager.is_enabled = False manager.is_enabled = False
+24 -1
View File
@@ -494,9 +494,32 @@ class TestShouldLogCooldown:
def test_cooldown_expires(self, build): def test_cooldown_expires(self, build):
manager = build() manager = build()
assert manager._should_log("api", cooldown=60) is True 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 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 # 4. Seam guard rails
+585
View File
@@ -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