mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-08-03 17:58:04 +00:00
Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e26ed29385 | ||
|
|
4f28d4eb64 | ||
|
|
21825cbfbc |
@@ -72,4 +72,8 @@ jobs:
|
|||||||
test/test_adaptive_layout.py \
|
test/test_adaptive_layout.py \
|
||||||
test/test_loader_compat_warning.py \
|
test/test_loader_compat_warning.py \
|
||||||
test/test_sports_base_characterization.py \
|
test/test_sports_base_characterization.py \
|
||||||
test/test_element_style.py
|
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_scroll.py
|
||||||
|
|||||||
+90
-1
@@ -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,11 +38,86 @@ 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` +
|
||||||
|
`modes.py`). The import path is unchanged: `from src.base_classes.sports
|
||||||
|
import SportsCore` still works.
|
||||||
|
- Nine methods promoted onto the sports base classes from the plugins'
|
||||||
|
bundled copies, plus the override points `_favorite_key`,
|
||||||
|
`_config_schema_path` and `_font_root` and the class attributes
|
||||||
|
`FINAL_PERIOD` / `CLOCK_COUNTS_DOWN`. See `docs/SPORTS_UNIFICATION.md`.
|
||||||
|
A plugin may start calling these once its manifest floors
|
||||||
|
`ledmatrix_min_version` at the release that ships them.
|
||||||
|
|
||||||
|
- `src/base_classes/sports/capabilities/` — opt-in capabilities for the sports
|
||||||
|
scoreboards, composed by inheritance rather than gated by config branches
|
||||||
|
inside the base classes:
|
||||||
|
- `CelebrationMixin` — the score/win takeover, merging the goal and score
|
||||||
|
dialects behind the `score_phrase()` / `win_phrase()` hooks, the
|
||||||
|
`COALESCE_SCORING_SEQUENCE` class attribute and the `_favorite_key` seam.
|
||||||
|
Reads both the `celebrate_opponent_goals` and `celebrate_opponent_scores`
|
||||||
|
config spellings. Sports that do not mix it in have none of this code in
|
||||||
|
their MRO.
|
||||||
|
- `RotationStrategy` + a name registry (`swrr`, `weighted`, `simple`,
|
||||||
|
plus `register_rotation_strategy` for plugin-supplied orderings). Each
|
||||||
|
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
|
||||||
|
`"0:00"` and then treated the game as finished once `period >= 4`. Baseball
|
||||||
|
has no game clock and `period` is the inning, so live MLB games disappeared
|
||||||
|
from the scoreboard from the 5th inning onward; UFC was affected the same
|
||||||
|
way. The clock check is now skipped when the clock is unusable, and the
|
||||||
|
period threshold is the per-sport `FINAL_PERIOD` (hockey ends in P3).
|
||||||
|
Sports whose clocks count up — soccer, AFL, NRL — set
|
||||||
|
`CLOCK_COUNTS_DOWN = False` and never run the check at all, since `0:00`
|
||||||
|
there means kickoff rather than expiry.
|
||||||
|
|
||||||
### Fixed
|
### Fixed
|
||||||
- `FontManager` resolves `assets/fonts` against the core install root instead
|
- `FontManager` resolves `assets/fonts` against the core install root instead
|
||||||
of the process working directory, so font loading works when the process
|
of the process working directory, so font loading works when the process
|
||||||
starts elsewhere (e.g. the plugin safety harness on CI).
|
starts elsewhere (e.g. the plugin safety harness on CI).
|
||||||
|
- Hockey events whose competitors carry no `statistics` array are no longer
|
||||||
|
discarded. The extractor read `competitor["statistics"]` unguarded, so a
|
||||||
|
`KeyError` inside the generator dropped the entire event despite valid
|
||||||
|
scores and status; shot counts now fall back to `0`.
|
||||||
|
- Live baseball events that populate status only at the competition level are
|
||||||
|
no longer discarded. The extractor read the event top-level
|
||||||
|
`game_event["status"]` for the inning; real ESPN events duplicate it, but
|
||||||
|
MiLB events synthesized from the MLB Stats API do not, so the lookup raised
|
||||||
|
a bare `KeyError`. It now reads the already-validated competition-level
|
||||||
|
status.
|
||||||
|
- `SportsLive._is_game_really_over` no longer crashes the live-update pass when
|
||||||
|
a feed sends an explicit null `period`. `None >= FINAL_PERIOD` raised
|
||||||
|
`TypeError`, and the only caller (`_detect_stale_games`) has no `try/except`
|
||||||
|
— the same failure shape as the already-fixed null `period_text`.
|
||||||
|
- An expired clock spelled `"00:00"` now ends the game. The check compared the
|
||||||
|
colon-stripped clock against a hand-listed set of literals, which `"0000"` is
|
||||||
|
not a member of, so a finished game with a two-digit-minute clock stayed on
|
||||||
|
the scoreboard indefinitely. The comparison is now numeric.
|
||||||
|
- `SportsCore._load_fonts` resolves `assets/fonts` through the `_font_root()`
|
||||||
|
seam instead of the process working directory. Started outside the install
|
||||||
|
root, every scoreboard font silently degraded to PIL's default bitmap face.
|
||||||
|
- `SportsCore._should_log` no longer raises `AttributeError` on the first
|
||||||
|
warning of a run; `_last_warning_time` is initialized in `__init__` rather
|
||||||
|
than lazily by an unrelated method.
|
||||||
|
- `SportsCore._resolve_project_path` resolved relative logo directories
|
||||||
|
against `<root>/src` instead of the repo root after `sports.py` became a
|
||||||
|
package — the class bodies moved byte-identically but `__file__` gained a
|
||||||
|
directory. Both it and `_font_root` now derive from one `_INSTALL_ROOT`
|
||||||
|
constant.
|
||||||
|
|
||||||
## 3.1.0
|
## 3.1.0
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,235 @@
|
|||||||
|
# Sports Code Unification — Architecture
|
||||||
|
|
||||||
|
How the nine sports scoreboard plugins converge onto shared core code **without**
|
||||||
|
becoming nine clients of a god class.
|
||||||
|
|
||||||
|
## The problem
|
||||||
|
|
||||||
|
Nine plugins (`afl`, `baseball`, `basketball`, `football`, `hockey`, `lacrosse`,
|
||||||
|
`nrl`, `soccer`, `ufc`) each ship a ~3,000-line `sports.py` descended from this
|
||||||
|
repo's `src/base_classes/sports.py`. They have drifted into three lineages, and
|
||||||
|
only 28 of the 66 methods appearing across them are present in all nine. One
|
||||||
|
logical fix (the UTC start-time bug) cost 75 files.
|
||||||
|
|
||||||
|
Merging everything into one base class would fix the duplication and create a
|
||||||
|
worse problem: a single 2,500-line class that all nine plugins inherit, where any
|
||||||
|
change has a nine-plugin blast radius and per-sport behavior survives only as
|
||||||
|
`if self.sport == "hockey"` branches.
|
||||||
|
|
||||||
|
## Three properties, three mechanisms
|
||||||
|
|
||||||
|
These are independent concerns. Conflating them is what produces god classes.
|
||||||
|
|
||||||
|
### Upgradability — a plugin keeps working across core versions
|
||||||
|
|
||||||
|
| Rule | Mechanism |
|
||||||
|
|---|---|
|
||||||
|
| Plugin loads on a core that predates a module | Guarded import with a bundled fallback (`try: from src.X import Y / except ModuleNotFoundError: from y import Y`) |
|
||||||
|
| Plugin loads on a core that predates a *method* | Capability probing — `hasattr(SportsCore, "_detect_stale_games")` — never a version comparison. The loader's compat check is advisory-only (it logs and continues), so probing is the real protection. |
|
||||||
|
| Core changes never break a plugin's rendering | The **view-model contract**: `_extract_game_details_common` returns a dict whose `GUARANTEED_KEYS` are frozen by `test/test_skin_system.py::TestViewModelContract`. Keys may be added, never renamed or removed. |
|
||||||
|
| A plugin can drop its bundled copy safely | The **sunset rule**: only when its manifest floors `ledmatrix_min_version` at the first core release shipping the module (recorded in `CHANGELOG.md`). |
|
||||||
|
|
||||||
|
The core API is **additive-only**. A method the plugins call is never removed or
|
||||||
|
given a new required parameter; new behavior arrives as new methods with
|
||||||
|
defaults, or as capabilities they opt into.
|
||||||
|
|
||||||
|
### Reusability — write once, nine plugins benefit
|
||||||
|
|
||||||
|
Only code that is **identical in intent across all nine** moves into the base
|
||||||
|
class. That set is small and knowable — it is exactly the methods present in every
|
||||||
|
copy today (phase B1 below). Everything else stays where it is until it earns
|
||||||
|
promotion.
|
||||||
|
|
||||||
|
### Modularity — a change to one feature cannot reach a plugin that doesn't use it
|
||||||
|
|
||||||
|
This is the property the naive merge destroys, and it is enforced structurally:
|
||||||
|
|
||||||
|
1. **Capabilities are separate modules composed by inheritance, not config
|
||||||
|
branches inside the base class.** Hockey has no celebrations, so
|
||||||
|
`HockeyLive` does not inherit `CelebrationMixin` — the celebration code is not
|
||||||
|
merely disabled for hockey, it is *not in hockey's MRO at all*. No shared
|
||||||
|
state, no dead branches, no risk. Contrast with
|
||||||
|
`if self.celebrations_enabled:` inside `SportsLive`, where a bug in
|
||||||
|
celebration code can still crash a plugin that never wanted the feature.
|
||||||
|
|
||||||
|
2. **Variant behavior is a strategy object chosen by name, not a branch.**
|
||||||
|
Live rotation exists in three dialects across the lineages; core ships all
|
||||||
|
three behind `rotation_strategy: "swrr" | "weighted" | "simple"` and a plugin
|
||||||
|
may register its own. Core never learns sport names.
|
||||||
|
|
||||||
|
3. **Sport-specific behavior is a documented override point.** The base class
|
||||||
|
declares the seam; the plugin fills it. Basketball's tournament-round parsing
|
||||||
|
and baseball's BDF sizing stay in their plugins forever — they are not
|
||||||
|
candidates for promotion, and core must never grow a branch for them.
|
||||||
|
|
||||||
|
4. **Files bound the blast radius.** Capabilities live in their own modules so a
|
||||||
|
diff shows at a glance which plugins a change can reach.
|
||||||
|
|
||||||
|
## Layering
|
||||||
|
|
||||||
|
```
|
||||||
|
src/base_classes/sports/
|
||||||
|
__init__.py re-exports the public API (import path unchanged)
|
||||||
|
core.py SportsCore — fetch, cache, config, logos, fonts, odds,
|
||||||
|
view-model extraction, the skin seam
|
||||||
|
modes.py SportsUpcoming / SportsRecent / SportsLive
|
||||||
|
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
|
||||||
|
`__init__` re-exports, so the conversion is invisible to every existing importer.
|
||||||
|
|
||||||
|
## Override points (the plugin-facing seam)
|
||||||
|
|
||||||
|
The base class calls these; plugins implement or override them. This table is the
|
||||||
|
contract — additions require a default implementation, removals require a
|
||||||
|
deprecation cycle.
|
||||||
|
|
||||||
|
| Hook | Purpose | Default |
|
||||||
|
|---|---|---|
|
||||||
|
| `_fetch_data()` | Sport's schedule source | abstract |
|
||||||
|
| `_extract_game_details(event)` | Sport-specific view-model fields on top of the common ones | delegates to `_extract_game_details_common` |
|
||||||
|
| `_draw_scorebug_layout(game, force_clear)` | Sport's card rendering | base layout |
|
||||||
|
| `_custom_scorebug_layout(game, draw)` | Per-sport overlay on the base layout | no-op |
|
||||||
|
| `render_skin_card(game, size)` | Skin-system entry point | built-in fallback |
|
||||||
|
| `score_phrase(points, team_abbr)` | Celebration wording (`"GOOOOAAALLL!"` vs `"TOUCHDOWN!"`). `points` is the score delta, which sports with variable-value scores use to name the play | `"<abbr> SCORES!"` — only consulted when `CelebrationMixin` is present |
|
||||||
|
| `win_phrase(team_abbr)` | Win-celebration wording | `"<abbr> WINS!"` — mixin only |
|
||||||
|
| `_favorite_key(game, side)` | Which view-model field identifies a team for favorites matching | `game["<side>_abbr"]` |
|
||||||
|
| `_config_schema_path()` | Plugin's `config_schema.json` — returning it routes `_get_layout_offset` through the `src.element_style` resolver (and gives it the defaults to compare against) | `None`, i.e. the classic inline `customization.layout` read |
|
||||||
|
| `_font_root()` | Directory to resolve `assets/fonts` against | core install root |
|
||||||
|
|
||||||
|
Two class attributes serve the same purpose for values that are per-sport
|
||||||
|
constants rather than behavior:
|
||||||
|
|
||||||
|
| Attribute | Meaning | Default |
|
||||||
|
|---|---|---|
|
||||||
|
| `FINAL_PERIOD` | Period at/after which a zero clock can mean "over" | `4` (hockey overrides to `3`) |
|
||||||
|
| `CLOCK_COUNTS_DOWN` | Whether `0:00` means "expired" | `True` (soccer/afl/nrl override to `False` — their clocks count up, so `0:00` is kickoff) |
|
||||||
|
| `COALESCE_SCORING_SEQUENCE` | Fold score increments arriving during an active celebration into that one celebration | `False` (football overrides to `True` — a touchdown lands as +6, then +1 for the extra point) |
|
||||||
|
|
||||||
|
### Why these are seams and not branches
|
||||||
|
|
||||||
|
`_favorite_key` exists because NRL abbreviations are **not unique** — "NEW" is both
|
||||||
|
Newcastle Knights and New Zealand Warriors, "CAN" both Canberra and Canterbury —
|
||||||
|
so NRL matches favorites on team ID. Flattening every plugin to abbreviations
|
||||||
|
would silently select the wrong club for NRL users. The base declares the seam,
|
||||||
|
NRL fills it, and core never learns the string `"nrl"`.
|
||||||
|
|
||||||
|
`CLOCK_COUNTS_DOWN` exists for the same reason in the opposite direction: a
|
||||||
|
soccer clock reading `0:00` means the match has not kicked off, so running the
|
||||||
|
clock-expiry branch there would evict live games.
|
||||||
|
|
||||||
|
`COALESCE_SCORING_SEQUENCE` is the third of the same kind. In football one
|
||||||
|
scoring play arrives as two score updates, so the follow-up must be folded into
|
||||||
|
the first celebration; in soccer two increments a few seconds apart are two real
|
||||||
|
goals, and folding them would swallow one. Neither default is "right" — which is
|
||||||
|
precisely why it is a declared per-sport constant rather than a hidden
|
||||||
|
assumption baked into the shared body.
|
||||||
|
|
||||||
|
## Capabilities
|
||||||
|
|
||||||
|
```
|
||||||
|
capabilities/
|
||||||
|
celebrations.py CelebrationMixin opt-in: afl, nrl, soccer, football
|
||||||
|
rotation.py RotationStrategy + registry
|
||||||
|
```
|
||||||
|
|
||||||
|
**`CelebrationMixin`** merges the two dialects the lineages grew
|
||||||
|
(`_check_for_goal`/`celebrate_opponent_goals` vs
|
||||||
|
`_check_for_score`/`celebrate_opponent_scores`). Their bodies were identical
|
||||||
|
apart from three things, each now a seam: wording (`score_phrase`), follow-up
|
||||||
|
suppression (`COALESCE_SCORING_SEQUENCE`), and team identity (`_favorite_key`,
|
||||||
|
so NRL matches on id). Both config spellings are read, so a plugin adopting the
|
||||||
|
mixin keeps working with the keys already in its published schema.
|
||||||
|
|
||||||
|
Mix it in **before** the mode class — `class SoccerLive(CelebrationMixin,
|
||||||
|
SportsLive)` — so the celebration `display()` runs first and falls through to
|
||||||
|
the scorebug via `super()`.
|
||||||
|
|
||||||
|
**Rotation strategies.** The three "dialects" turned out to be one algorithm
|
||||||
|
(Smooth Weighted Round-Robin) in two shapes: an incremental picker holding state
|
||||||
|
across calls (afl/nrl/soccer) and a precomputed per-cycle list
|
||||||
|
(football/baseball/basketball, and hockey with a different loop shape). They
|
||||||
|
agree within a cycle and differ only at the boundary — the incremental form has
|
||||||
|
no restart seam — so core ships both rather than declaring a winner:
|
||||||
|
|
||||||
|
```python
|
||||||
|
self.rotation = get_rotation_strategy("swrr", weight_for=self._live_weight)
|
||||||
|
```
|
||||||
|
|
||||||
|
`weight_for` is supplied by the host, so the *favorites* policy stays with the
|
||||||
|
plugin and `rotation.py` never learns what a favorite is. An unknown strategy
|
||||||
|
name degrades to `simple` rather than raising: the name comes from user config,
|
||||||
|
and a typo should cost the boost, not the scoreboard. When a plugin needs an
|
||||||
|
ordering that core does not ship, it calls `register_rotation_strategy` to add
|
||||||
|
its own — rather than core growing a branch for it.
|
||||||
|
|
||||||
|
`test_sports_capabilities.py` checks each strategy against a **verbatim
|
||||||
|
transcription** of the plugin code it replaces, over every live-game shape up to
|
||||||
|
four games. That differential is what B5 deletes the bundled copies on the
|
||||||
|
strength of.
|
||||||
|
|
||||||
|
## Scroll display — where the promotion line falls
|
||||||
|
|
||||||
|
`src/common/sports_scroll.py` is deliberately *not* a superset of the ten
|
||||||
|
`scroll_display.py` copies. A method-level comparison of the eight that share a
|
||||||
|
shape (f1 and ufc are genuine forks) found a sharp split:
|
||||||
|
|
||||||
|
| Layer | Evidence | Outcome |
|
||||||
|
|---|---|---|
|
||||||
|
| Orchestration — `get_all_vegas_content_items`, `clear_all`, `get_scroll_info`, `get_dynamic_duration`, `is_complete`, `display_frame` | identical to 96–100% similar across all eight | **promoted** |
|
||||||
|
| Settings — `_get_scroll_settings` | one algorithm; the copies differ *only* in which league keys they walk | **promoted**, with the ladder as data (`SCROLL_LEAGUE_KEYS`) |
|
||||||
|
| Content — `prepare_scroll_content`, `_load_separator_icons` | 8 distinct bodies across 8 plugins (145 lines, 53% similar at worst); icons 6% | **override point, permanently** |
|
||||||
|
|
||||||
|
Same name, different job: `prepare_scroll_content` draws *this sport's* game
|
||||||
|
card. Merging the eight bodies would be the exact mistake the promotion rule
|
||||||
|
exists to prevent, so the base class raises `NotImplementedError` rather than
|
||||||
|
rendering something plausible — a base that rendered *something* would let a
|
||||||
|
plugin ship a silently blank scroll.
|
||||||
|
|
||||||
|
The one behavior the upstreamed version adds is native
|
||||||
|
`global_config['target_fps']` support. The bundled copies hardcode ~100 FPS via
|
||||||
|
`scroll_delay = 0.01` and never consult the global smooth-scrolling target;
|
||||||
|
Part A threaded it through each copy by hand, and this makes that threading
|
||||||
|
legacy compatibility rather than the mechanism.
|
||||||
|
|
||||||
|
## Phases
|
||||||
|
|
||||||
|
| Phase | Scope | Risk control |
|
||||||
|
|---|---|---|
|
||||||
|
| **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 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 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
|
||||||
|
|
||||||
|
- **Promote on evidence, not intuition.** A method moves to core when every copy
|
||||||
|
has it and they agree on intent. Otherwise it stays in the plugins.
|
||||||
|
- **Never add a sport name to core.** If core needs to know which sport it is,
|
||||||
|
the design is wrong — add an override point instead.
|
||||||
|
- **A capability that is not opted into must not execute.** If you find yourself
|
||||||
|
writing `if self.<capability>_enabled` inside a base class, it belongs in a
|
||||||
|
mixin.
|
||||||
|
- **Touch the view-model keys only additively.** Published skins depend on them.
|
||||||
|
- **Every promotion lands with the characterization suite green**, and every
|
||||||
|
pilot adoption lands with that plugin's harness and golden suites green.
|
||||||
@@ -0,0 +1,122 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Assert that a release tag, the CHANGELOG, and `src.__version__` all agree.
|
||||||
|
|
||||||
|
Run it *before* creating a tag to check yourself:
|
||||||
|
|
||||||
|
python scripts/check_release_version.py v3.2.0
|
||||||
|
|
||||||
|
Wiring it into CI (on pushed `v*` tags and published releases) is a follow-up
|
||||||
|
PR, so for now it is a manual pre-flight: run it before creating the tag and a
|
||||||
|
mismatch shows up here rather than as a silent wrong answer on user devices.
|
||||||
|
|
||||||
|
Why this exists: `v3.1.0` was tagged 2026-05-31 while `src/__init__.py` still
|
||||||
|
said `"1.0.0"`; the bump to `"3.1.0"` did not land until 2026-07-12. Devices
|
||||||
|
installed from that release report `1.0.0`, which is below the `(2, 0, 0)` floor
|
||||||
|
in `PluginLoader._warn_if_incompatible`, so they are silently exempt from every
|
||||||
|
plugin compatibility warning. Plugin `ledmatrix_min_version` floors are only as
|
||||||
|
trustworthy as this agreement. See `docs/SPORTS_UNIFICATION.md`, phase B4.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import argparse
|
||||||
|
import re
|
||||||
|
import sys
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||||
|
sys.path.insert(0, str(REPO_ROOT))
|
||||||
|
|
||||||
|
# [0-9] rather than \d, and [ \t] rather than \s: \d also matches non-ASCII
|
||||||
|
# decimal digits (which int() parses), and \s matches newlines, so "##\n3.2.0"
|
||||||
|
# would otherwise read as a version heading. Keep these in step with
|
||||||
|
# test/test_version_consistency.py.
|
||||||
|
SEMVER = re.compile(r"^[0-9]+\.[0-9]+\.[0-9]+$")
|
||||||
|
HEADING = re.compile(
|
||||||
|
r"^##[ \t]+(?P<version>[0-9]+\.[0-9]+\.[0-9]+)[ \t]*$", re.MULTILINE)
|
||||||
|
|
||||||
|
|
||||||
|
def normalize(tag: str) -> str:
|
||||||
|
"""`v3.2.0` and `3.2.0` are the same release; tags here carry the `v`."""
|
||||||
|
return tag[1:] if tag.startswith("v") else tag
|
||||||
|
|
||||||
|
|
||||||
|
def newest_changelog_version(changelog: Path) -> str | None:
|
||||||
|
"""Newest version heading, or None when there is none.
|
||||||
|
|
||||||
|
Raises OSError if the file cannot be read; main() turns that into a clear
|
||||||
|
message rather than a traceback, because this runs as a release gate and a
|
||||||
|
traceback there reads as "the tooling is broken", not "your CHANGELOG is
|
||||||
|
missing".
|
||||||
|
"""
|
||||||
|
headings = HEADING.findall(changelog.read_text(encoding="utf-8"))
|
||||||
|
return headings[0] if headings else None
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> int:
|
||||||
|
parser = argparse.ArgumentParser(description=__doc__)
|
||||||
|
parser.add_argument(
|
||||||
|
"tag",
|
||||||
|
help="Release tag to check, with or without the leading 'v' (e.g. v3.2.0)",
|
||||||
|
)
|
||||||
|
args = parser.parse_args()
|
||||||
|
|
||||||
|
from src import __version__ as core_version
|
||||||
|
|
||||||
|
tag_version = normalize(args.tag)
|
||||||
|
changelog_path = REPO_ROOT / "CHANGELOG.md"
|
||||||
|
|
||||||
|
problems: list[str] = []
|
||||||
|
|
||||||
|
try:
|
||||||
|
changelog_version = newest_changelog_version(changelog_path)
|
||||||
|
except OSError as e:
|
||||||
|
print(
|
||||||
|
f"Release version check FAILED for tag {args.tag}:\n"
|
||||||
|
f" - could not read {changelog_path}: {e}\n"
|
||||||
|
f" Restore the file (git checkout -- CHANGELOG.md) and re-run.",
|
||||||
|
file=sys.stderr,
|
||||||
|
)
|
||||||
|
return 1
|
||||||
|
|
||||||
|
if not SEMVER.match(tag_version):
|
||||||
|
problems.append(
|
||||||
|
f"tag {args.tag!r} is not vX.Y.Z. Older tags (v2.5) predate this "
|
||||||
|
"check; new releases must be full semver so floors can parse them."
|
||||||
|
)
|
||||||
|
|
||||||
|
if not SEMVER.match(core_version):
|
||||||
|
problems.append(f"src.__version__ is {core_version!r}, which is not X.Y.Z")
|
||||||
|
|
||||||
|
if tag_version != core_version:
|
||||||
|
problems.append(
|
||||||
|
f"tag says {tag_version} but src.__version__ says {core_version}. "
|
||||||
|
"Bump src/__init__.py to match the tag before releasing — devices "
|
||||||
|
"report __version__, not the tag, and plugin floors compare "
|
||||||
|
"against it."
|
||||||
|
)
|
||||||
|
|
||||||
|
if changelog_version is None:
|
||||||
|
problems.append("CHANGELOG.md has no '## X.Y.Z' version heading")
|
||||||
|
elif changelog_version != core_version:
|
||||||
|
problems.append(
|
||||||
|
f"CHANGELOG.md's newest heading is {changelog_version} but "
|
||||||
|
f"src.__version__ is {core_version}. Plugin authors read the "
|
||||||
|
"CHANGELOG to pick a ledmatrix_min_version floor."
|
||||||
|
)
|
||||||
|
|
||||||
|
if problems:
|
||||||
|
print(f"Release version check FAILED for tag {args.tag}:", file=sys.stderr)
|
||||||
|
for problem in problems:
|
||||||
|
print(f" - {problem}", file=sys.stderr)
|
||||||
|
return 1
|
||||||
|
|
||||||
|
print(
|
||||||
|
f"OK: tag {args.tag}, src.__version__ {core_version}, and the CHANGELOG "
|
||||||
|
"all agree."
|
||||||
|
)
|
||||||
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
sys.exit(main())
|
||||||
+1
-1
@@ -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"
|
||||||
|
|
||||||
|
|||||||
@@ -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(
|
||||||
@@ -164,7 +169,13 @@ class Baseball(SportsCore):
|
|||||||
# Get game state information
|
# Get game state information
|
||||||
if status_state == "in":
|
if status_state == "in":
|
||||||
# For live games, get detailed state
|
# For live games, get detailed state
|
||||||
inning = game_event["status"].get(
|
# Use the competition-level `status` already validated by
|
||||||
|
# _extract_game_details_common. Real ESPN events duplicate
|
||||||
|
# status at the event top level, but MiLB events (synthesized
|
||||||
|
# from the MLB Stats API into an ESPN-like shape) populate
|
||||||
|
# only the competition-level one, so the top-level lookup
|
||||||
|
# raised a bare KeyError and dropped the event.
|
||||||
|
inning = status.get(
|
||||||
"period", 1
|
"period", 1
|
||||||
) # Get inning from status period
|
) # Get inning from status period
|
||||||
|
|
||||||
@@ -187,7 +198,7 @@ class Baseball(SportsCore):
|
|||||||
if "end" in status_detail or "end" in status_short:
|
if "end" in status_detail or "end" in status_short:
|
||||||
inning_half = "top"
|
inning_half = "top"
|
||||||
inning = (
|
inning = (
|
||||||
game_event["status"].get("period", 1) + 1
|
status.get("period", 1) + 1
|
||||||
) # Use period and increment for next inning
|
) # Use period and increment for next inning
|
||||||
if is_favorite_game:
|
if is_favorite_game:
|
||||||
self.logger.debug(
|
self.logger.debug(
|
||||||
|
|||||||
@@ -38,10 +38,17 @@ class Hockey(SportsCore):
|
|||||||
status = competition["status"]
|
status = competition["status"]
|
||||||
powerplay = False
|
powerplay = False
|
||||||
penalties = ""
|
penalties = ""
|
||||||
|
# A competitor may legitimately arrive without a "statistics"
|
||||||
|
# array (pre-game feeds, and some in-progress ones). Reading it
|
||||||
|
# unguarded raised KeyError inside the generator and dropped the
|
||||||
|
# WHOLE event, discarding valid scores and status. Default to an
|
||||||
|
# empty list so the saves/shots figures fall back to 0 instead.
|
||||||
|
home_stats = home_team.get("statistics", [])
|
||||||
|
away_stats = away_team.get("statistics", [])
|
||||||
home_team_saves = next(
|
home_team_saves = next(
|
||||||
(
|
(
|
||||||
int(c["displayValue"])
|
int(c["displayValue"])
|
||||||
for c in home_team["statistics"]
|
for c in home_stats
|
||||||
if c.get("name") == "saves"
|
if c.get("name") == "saves"
|
||||||
),
|
),
|
||||||
0,
|
0,
|
||||||
@@ -49,7 +56,7 @@ class Hockey(SportsCore):
|
|||||||
home_team_saves_per = next(
|
home_team_saves_per = next(
|
||||||
(
|
(
|
||||||
float(c["displayValue"])
|
float(c["displayValue"])
|
||||||
for c in home_team["statistics"]
|
for c in home_stats
|
||||||
if c.get("name") == "savePct"
|
if c.get("name") == "savePct"
|
||||||
),
|
),
|
||||||
0.0,
|
0.0,
|
||||||
@@ -57,7 +64,7 @@ class Hockey(SportsCore):
|
|||||||
away_team_saves = next(
|
away_team_saves = next(
|
||||||
(
|
(
|
||||||
int(c["displayValue"])
|
int(c["displayValue"])
|
||||||
for c in away_team["statistics"]
|
for c in away_stats
|
||||||
if c.get("name") == "saves"
|
if c.get("name") == "saves"
|
||||||
),
|
),
|
||||||
0,
|
0,
|
||||||
@@ -65,7 +72,7 @@ class Hockey(SportsCore):
|
|||||||
away_team_saves_per = next(
|
away_team_saves_per = next(
|
||||||
(
|
(
|
||||||
float(c["displayValue"])
|
float(c["displayValue"])
|
||||||
for c in away_team["statistics"]
|
for c in away_stats
|
||||||
if c.get("name") == "savePct"
|
if c.get("name") == "savePct"
|
||||||
),
|
),
|
||||||
0.0,
|
0.0,
|
||||||
|
|||||||
@@ -0,0 +1,17 @@
|
|||||||
|
"""Sports scoreboard base classes.
|
||||||
|
|
||||||
|
Formerly the single module ``src/base_classes/sports.py``; now a package so
|
||||||
|
capabilities can be composed instead of accumulating in one class. See
|
||||||
|
docs/SPORTS_UNIFICATION.md for the architecture. The import path is
|
||||||
|
unchanged: ``from src.base_classes.sports import SportsCore`` still works.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from .core import SportsCore
|
||||||
|
from .modes import SportsLive, SportsRecent, SportsUpcoming
|
||||||
|
|
||||||
|
__all__ = [
|
||||||
|
"SportsCore",
|
||||||
|
"SportsUpcoming",
|
||||||
|
"SportsRecent",
|
||||||
|
"SportsLive",
|
||||||
|
]
|
||||||
@@ -0,0 +1,32 @@
|
|||||||
|
"""Opt-in capabilities for the sports scoreboards.
|
||||||
|
|
||||||
|
Each module here is a feature that only *some* sports want. They are composed
|
||||||
|
by inheritance (mixins) or selected by name (strategies) — never enabled by an
|
||||||
|
``if self.<feature>_enabled:`` branch inside the base classes.
|
||||||
|
|
||||||
|
The distinction matters: hockey has no celebrations, so ``HockeyLive`` does not
|
||||||
|
inherit :class:`~.celebrations.CelebrationMixin` and the celebration code is not
|
||||||
|
in hockey's MRO at all. A bug in it cannot reach a plugin that never opted in.
|
||||||
|
|
||||||
|
See ``docs/SPORTS_UNIFICATION.md`` for the full rationale.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from .celebrations import CelebrationMixin
|
||||||
|
from .rotation import (
|
||||||
|
RotationStrategy,
|
||||||
|
SimpleRotation,
|
||||||
|
SmoothWeightedRotation,
|
||||||
|
WeightedCycleRotation,
|
||||||
|
get_rotation_strategy,
|
||||||
|
register_rotation_strategy,
|
||||||
|
)
|
||||||
|
|
||||||
|
__all__ = [
|
||||||
|
"CelebrationMixin",
|
||||||
|
"RotationStrategy",
|
||||||
|
"SimpleRotation",
|
||||||
|
"SmoothWeightedRotation",
|
||||||
|
"WeightedCycleRotation",
|
||||||
|
"get_rotation_strategy",
|
||||||
|
"register_rotation_strategy",
|
||||||
|
]
|
||||||
@@ -0,0 +1,418 @@
|
|||||||
|
"""Score / win celebration takeover — an opt-in capability.
|
||||||
|
|
||||||
|
Four of the nine scoreboards celebrate (afl, nrl, soccer, football); the other
|
||||||
|
five do not. This is a **mixin** rather than a flag inside ``SportsLive`` so the
|
||||||
|
five that do not opt in have none of this code in their MRO: a bug here cannot
|
||||||
|
reach hockey, and hockey's config never grows keys it ignores.
|
||||||
|
|
||||||
|
Usage — mix in *before* the mode class so its ``display`` runs first::
|
||||||
|
|
||||||
|
class SoccerLive(CelebrationMixin, SportsLive):
|
||||||
|
def score_phrase(self, points, team_abbr):
|
||||||
|
return secrets.choice(("GOOOOAAALLL!", f"{team_abbr} SCORES!"))
|
||||||
|
|
||||||
|
The two lineages spelled this differently (``_check_for_goal`` /
|
||||||
|
``celebrate_opponent_goals`` in the soccer lineage, ``_check_for_score`` /
|
||||||
|
``celebrate_opponent_scores`` in football) but the bodies were identical apart
|
||||||
|
from three things, each of which is a seam here rather than a branch:
|
||||||
|
|
||||||
|
* **wording** — :meth:`score_phrase`, the hook football uses to say "TOUCHDOWN"
|
||||||
|
from the points delta and soccer uses to say "GOOOOAAALLL";
|
||||||
|
* **follow-up suppression** — :attr:`COALESCE_SCORING_SEQUENCE`, on for football
|
||||||
|
where a touchdown lands as +6 then +1 a few seconds later, off elsewhere where
|
||||||
|
two quick goals are two real events;
|
||||||
|
* **team identity** — matching goes through ``_favorite_key``, so nrl can match
|
||||||
|
on team id (its abbreviations are ambiguous) without core knowing why.
|
||||||
|
|
||||||
|
The config keys are read under both spellings, so a plugin adopting the mixin
|
||||||
|
keeps working with the ``*_goals`` keys already in its published schema.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import re
|
||||||
|
import time
|
||||||
|
from typing import Any, Dict, List, Optional
|
||||||
|
|
||||||
|
from PIL import Image, ImageDraw
|
||||||
|
|
||||||
|
|
||||||
|
class CelebrationMixin:
|
||||||
|
"""Full-screen takeover when a tracked team scores or wins."""
|
||||||
|
|
||||||
|
#: Collapse increments that land while a celebration is already on screen
|
||||||
|
#: into that one celebration. True for sports where a single scoring play
|
||||||
|
#: arrives as more than one score update (football: touchdown +6, then the
|
||||||
|
#: extra point +1). False where consecutive increments are distinct events —
|
||||||
|
#: suppressing there would swallow a real goal.
|
||||||
|
COALESCE_SCORING_SEQUENCE = False
|
||||||
|
|
||||||
|
def __init__(self, *args, **kwargs):
|
||||||
|
super().__init__(*args, **kwargs)
|
||||||
|
mode_config = getattr(self, "mode_config", {}) or {}
|
||||||
|
self.celebration_enabled = mode_config.get("celebration_enabled", True)
|
||||||
|
# Coerced and floored at init: this value is compared numerically on the
|
||||||
|
# display path, where a string from a hand-edited config would raise
|
||||||
|
# TypeError outside any try block, and a zero or negative value would
|
||||||
|
# arm a celebration that can never render.
|
||||||
|
raw_duration = mode_config.get("celebration_duration", 8)
|
||||||
|
try:
|
||||||
|
self.celebration_duration = max(1.0, float(raw_duration))
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
self.logger.warning(
|
||||||
|
"[Celebrations] Unusable celebration_duration %r; using 8s. "
|
||||||
|
"Set a positive number of seconds.",
|
||||||
|
raw_duration,
|
||||||
|
)
|
||||||
|
self.celebration_duration = 8.0
|
||||||
|
# Both spellings: the soccer lineage ships `celebrate_opponent_goals`,
|
||||||
|
# football ships `celebrate_opponent_scores`. Whichever the plugin's
|
||||||
|
# schema declares is the one its users have set.
|
||||||
|
self.celebrate_opponent_scores = mode_config.get(
|
||||||
|
"celebrate_opponent_scores",
|
||||||
|
mode_config.get("celebrate_opponent_goals", False),
|
||||||
|
)
|
||||||
|
# Per-game score baselines: {game_id: {"away": int, "home": int}}
|
||||||
|
self._score_baselines: Dict[str, Dict[str, int]] = {}
|
||||||
|
# The active celebration (a game *snapshot*, so a win survives the game
|
||||||
|
# leaving live_games) or None. See _start_celebration for the shape.
|
||||||
|
self.active_celebration: Optional[Dict[str, Any]] = None
|
||||||
|
|
||||||
|
# ------------------------------------------------------------------
|
||||||
|
# Override points
|
||||||
|
# ------------------------------------------------------------------
|
||||||
|
|
||||||
|
def score_phrase(self, points: int, team_abbr: str) -> str:
|
||||||
|
"""The wording for a score celebration.
|
||||||
|
|
||||||
|
``points`` is the score delta that triggered it, which sports with
|
||||||
|
variable-value scores use to name the play. The default is deliberately
|
||||||
|
sport-neutral; every celebrating plugin overrides it.
|
||||||
|
"""
|
||||||
|
return f"{team_abbr} SCORES!"
|
||||||
|
|
||||||
|
def win_phrase(self, team_abbr: str) -> str:
|
||||||
|
"""The wording for a win celebration."""
|
||||||
|
return f"{team_abbr} WINS!"
|
||||||
|
|
||||||
|
def _is_favorite(self, key: Optional[str]) -> bool:
|
||||||
|
"""Whether ``key`` (whatever ``_favorite_key`` returns) is a favorite."""
|
||||||
|
return bool(self.favorite_teams) and key in self.favorite_teams
|
||||||
|
|
||||||
|
# ------------------------------------------------------------------
|
||||||
|
# Detection
|
||||||
|
# ------------------------------------------------------------------
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _score_to_int(score) -> Optional[int]:
|
||||||
|
"""Coerce an ESPN score value (str / int / dict) to an int, or None."""
|
||||||
|
try:
|
||||||
|
if score is None:
|
||||||
|
return None
|
||||||
|
if isinstance(score, str):
|
||||||
|
s = score.strip()
|
||||||
|
if not s:
|
||||||
|
return None
|
||||||
|
try:
|
||||||
|
return int(float(s))
|
||||||
|
except ValueError:
|
||||||
|
numbers = re.findall(r"\d+", s)
|
||||||
|
return int(numbers[0]) if numbers else None
|
||||||
|
if isinstance(score, dict):
|
||||||
|
return int(float(score.get("value", score.get("displayValue", 0))))
|
||||||
|
return int(float(score))
|
||||||
|
except (ValueError, TypeError):
|
||||||
|
return None
|
||||||
|
|
||||||
|
def _should_celebrate_for(self, game: Dict, side: str) -> bool:
|
||||||
|
"""Whether a score by ``side`` in ``game`` should trigger a celebration."""
|
||||||
|
if self._is_favorite(self._favorite_key(game, side)):
|
||||||
|
return True
|
||||||
|
if not self.favorite_teams:
|
||||||
|
# No favorites configured: the user opted to show this game, so
|
||||||
|
# celebrate any score in it.
|
||||||
|
return True
|
||||||
|
# Favorites exist but this team isn't one -> it's the opponent.
|
||||||
|
return self.celebrate_opponent_scores
|
||||||
|
|
||||||
|
def prune_score_baselines(self, live_games: List[Dict]) -> None:
|
||||||
|
"""Drop baselines for games no longer live.
|
||||||
|
|
||||||
|
Only :meth:`_check_for_win` removes entries, and it only fires for games
|
||||||
|
seen to go final. A game that vanishes from the live list any other way
|
||||||
|
— postponed, dropped by the feed, or simply still live when the board
|
||||||
|
restarts — leaves its baseline behind forever, so on a board that runs
|
||||||
|
all season the dict grows without bound.
|
||||||
|
|
||||||
|
Call this from ``update()`` with the current live set, alongside the
|
||||||
|
equivalent pruning in :meth:`SmoothWeightedRotation.next_game`.
|
||||||
|
"""
|
||||||
|
live_ids = {g.get("id") for g in live_games}
|
||||||
|
self._score_baselines = {
|
||||||
|
gid: baseline
|
||||||
|
for gid, baseline in self._score_baselines.items()
|
||||||
|
if gid in live_ids
|
||||||
|
}
|
||||||
|
|
||||||
|
def has_active_celebration(self) -> bool:
|
||||||
|
"""True while a celebration is within its display window."""
|
||||||
|
celebration = self.active_celebration
|
||||||
|
return bool(celebration) and (
|
||||||
|
time.time() - celebration["started_at"] < self.celebration_duration
|
||||||
|
)
|
||||||
|
|
||||||
|
def _check_for_score(self, game: Dict) -> None:
|
||||||
|
"""Compare a live game's score against its baseline and arm a
|
||||||
|
celebration when a celebratable team's score increases."""
|
||||||
|
if not self.celebration_enabled:
|
||||||
|
return
|
||||||
|
game_id = game.get("id")
|
||||||
|
if not game_id:
|
||||||
|
return
|
||||||
|
away = self._score_to_int(game.get("away_score"))
|
||||||
|
home = self._score_to_int(game.get("home_score"))
|
||||||
|
if away is None or home is None:
|
||||||
|
return
|
||||||
|
|
||||||
|
baseline = self._score_baselines.get(game_id)
|
||||||
|
# Always refresh the baseline: a first sighting must never celebrate (a
|
||||||
|
# game already in progress at boot would false-fire), and a decrement
|
||||||
|
# (VAR, a correction) just re-bases silently.
|
||||||
|
self._score_baselines[game_id] = {"away": away, "home": home}
|
||||||
|
if baseline is None:
|
||||||
|
return
|
||||||
|
|
||||||
|
away_delta = away - baseline["away"]
|
||||||
|
home_delta = home - baseline["home"]
|
||||||
|
if away_delta <= 0 and home_delta <= 0:
|
||||||
|
return
|
||||||
|
|
||||||
|
# One takeover per scoring sequence, where the sport has such a thing.
|
||||||
|
# The baseline is already advanced above, so nothing re-fires later.
|
||||||
|
if self.COALESCE_SCORING_SEQUENCE and self.has_active_celebration():
|
||||||
|
return
|
||||||
|
|
||||||
|
scored_side = None
|
||||||
|
points = 0
|
||||||
|
if away_delta > 0 and self._should_celebrate_for(game, "away"):
|
||||||
|
scored_side, points = "away", away_delta
|
||||||
|
if scored_side is None and home_delta > 0 and self._should_celebrate_for(
|
||||||
|
game, "home"
|
||||||
|
):
|
||||||
|
scored_side, points = "home", home_delta
|
||||||
|
if scored_side is None:
|
||||||
|
return
|
||||||
|
|
||||||
|
self._start_celebration(
|
||||||
|
game,
|
||||||
|
"score",
|
||||||
|
scored_side=scored_side,
|
||||||
|
team_abbr=game.get(f"{scored_side}_abbr", ""),
|
||||||
|
away_score=away,
|
||||||
|
home_score=home,
|
||||||
|
points=points,
|
||||||
|
)
|
||||||
|
|
||||||
|
def _check_for_win(self, game: Dict) -> None:
|
||||||
|
"""When a game we were tracking live goes final, arm a win celebration
|
||||||
|
if a favorite won. Fires at most once per game."""
|
||||||
|
if not self.celebration_enabled:
|
||||||
|
return
|
||||||
|
game_id = game.get("id")
|
||||||
|
if not game_id:
|
||||||
|
return
|
||||||
|
# Only celebrate wins for games we actually watched go live: one seen
|
||||||
|
# for the first time already-final (the board started after full time)
|
||||||
|
# has no baseline and must not fire.
|
||||||
|
if game_id not in self._score_baselines:
|
||||||
|
return
|
||||||
|
# Consume the baseline so this can only fire once.
|
||||||
|
self._score_baselines.pop(game_id, None)
|
||||||
|
|
||||||
|
away = self._score_to_int(game.get("away_score"))
|
||||||
|
home = self._score_to_int(game.get("home_score"))
|
||||||
|
if away is None or home is None:
|
||||||
|
return
|
||||||
|
|
||||||
|
if away > home:
|
||||||
|
winner_side = "away"
|
||||||
|
elif home > away:
|
||||||
|
winner_side = "home"
|
||||||
|
else:
|
||||||
|
return # draw -> no win celebration
|
||||||
|
|
||||||
|
# Wins are gated strictly on favorites: every game ends, so the
|
||||||
|
# "no favorites -> celebrate all" score fallback would be far too noisy.
|
||||||
|
if not self._is_favorite(self._favorite_key(game, winner_side)):
|
||||||
|
return
|
||||||
|
|
||||||
|
self._start_celebration(
|
||||||
|
game,
|
||||||
|
"win",
|
||||||
|
scored_side=winner_side,
|
||||||
|
team_abbr=game.get(f"{winner_side}_abbr", ""),
|
||||||
|
away_score=away,
|
||||||
|
home_score=home,
|
||||||
|
)
|
||||||
|
|
||||||
|
def _start_celebration(
|
||||||
|
self,
|
||||||
|
game: Dict,
|
||||||
|
kind: str,
|
||||||
|
scored_side: str,
|
||||||
|
team_abbr: str,
|
||||||
|
away_score: int,
|
||||||
|
home_score: int,
|
||||||
|
points: int = 0,
|
||||||
|
) -> None:
|
||||||
|
"""Arm a celebration. ``scored_side`` ('away'/'home') is the side whose
|
||||||
|
score digit gets highlighted."""
|
||||||
|
phrase = (
|
||||||
|
self.win_phrase(team_abbr)
|
||||||
|
if kind == "win"
|
||||||
|
else self.score_phrase(points, team_abbr)
|
||||||
|
)
|
||||||
|
|
||||||
|
self.active_celebration = {
|
||||||
|
"kind": kind,
|
||||||
|
"game": dict(game), # snapshot: survives the game leaving live_games
|
||||||
|
"scored_side": scored_side,
|
||||||
|
"team_abbr": team_abbr,
|
||||||
|
"away_score": away_score,
|
||||||
|
"home_score": home_score,
|
||||||
|
"started_at": time.time(),
|
||||||
|
"phrase": phrase,
|
||||||
|
}
|
||||||
|
# Pin focus to the involved game so the post-celebration scorebug
|
||||||
|
# resumes on it.
|
||||||
|
self.current_game = dict(game)
|
||||||
|
self.logger.info(
|
||||||
|
f"[Celebrations] {kind} armed: {phrase} "
|
||||||
|
f"[{game.get('away_abbr')} {away_score}-{home_score} {game.get('home_abbr')}]"
|
||||||
|
)
|
||||||
|
|
||||||
|
# ------------------------------------------------------------------
|
||||||
|
# Rendering
|
||||||
|
# ------------------------------------------------------------------
|
||||||
|
|
||||||
|
def _fit_font(self, draw, text: str, max_width: int, fonts: List):
|
||||||
|
"""The first font whose rendered ``text`` fits ``max_width``, falling
|
||||||
|
back to the last (smallest) font."""
|
||||||
|
for font in fonts:
|
||||||
|
if draw.textlength(text, font=font) <= max_width - 2:
|
||||||
|
return font
|
||||||
|
return fonts[-1]
|
||||||
|
|
||||||
|
def _draw_celebration_layout(
|
||||||
|
self, celebration: Dict, force_clear: bool = False
|
||||||
|
) -> None:
|
||||||
|
"""Render the full-screen score/win takeover."""
|
||||||
|
if force_clear:
|
||||||
|
self.display_manager.clear()
|
||||||
|
|
||||||
|
display_width = (
|
||||||
|
self.display_manager.matrix.width
|
||||||
|
if hasattr(self.display_manager, "matrix") and self.display_manager.matrix
|
||||||
|
else self.display_width
|
||||||
|
)
|
||||||
|
display_height = (
|
||||||
|
self.display_manager.matrix.height
|
||||||
|
if hasattr(self.display_manager, "matrix") and self.display_manager.matrix
|
||||||
|
else self.display_height
|
||||||
|
)
|
||||||
|
|
||||||
|
elapsed = time.time() - celebration["started_at"]
|
||||||
|
game = celebration["game"]
|
||||||
|
|
||||||
|
# Background: a brief color flash for the first ~1.2s, then black.
|
||||||
|
bg = (0, 0, 0, 255)
|
||||||
|
if elapsed < 1.2 and int(elapsed / 0.2) % 2 == 0:
|
||||||
|
bg = (12, 12, 48, 255)
|
||||||
|
main_img = Image.new("RGBA", (display_width, display_height), bg)
|
||||||
|
overlay = Image.new("RGBA", (display_width, display_height), (0, 0, 0, 0))
|
||||||
|
draw = ImageDraw.Draw(overlay)
|
||||||
|
|
||||||
|
# Logos at the edges (best-effort: a logo failure must not blank the
|
||||||
|
# celebration).
|
||||||
|
try:
|
||||||
|
center_y = display_height // 2
|
||||||
|
home_logo = self._load_and_resize_logo(
|
||||||
|
game.get("home_id"), game.get("home_abbr"),
|
||||||
|
game.get("home_logo_path"), game.get("home_logo_url"),
|
||||||
|
)
|
||||||
|
away_logo = self._load_and_resize_logo(
|
||||||
|
game.get("away_id"), game.get("away_abbr"),
|
||||||
|
game.get("away_logo_path"), game.get("away_logo_url"),
|
||||||
|
)
|
||||||
|
if home_logo:
|
||||||
|
main_img.paste(
|
||||||
|
home_logo,
|
||||||
|
(display_width - home_logo.width + 2, center_y - home_logo.height // 2),
|
||||||
|
home_logo,
|
||||||
|
)
|
||||||
|
if away_logo:
|
||||||
|
main_img.paste(
|
||||||
|
away_logo, (-2, center_y - away_logo.height // 2), away_logo
|
||||||
|
)
|
||||||
|
except Exception as e:
|
||||||
|
self.logger.debug(f"[Celebrations] Logo load failed: {e}")
|
||||||
|
|
||||||
|
# Phrase across the top, shrunk to fit the panel width.
|
||||||
|
phrase = celebration["phrase"]
|
||||||
|
phrase_font = self._fit_font(
|
||||||
|
draw, phrase, display_width, [self.fonts["time"], self.fonts["status"]]
|
||||||
|
)
|
||||||
|
phrase_width = draw.textlength(phrase, font=phrase_font)
|
||||||
|
self._draw_text_with_outline(
|
||||||
|
draw, phrase, ((display_width - phrase_width) // 2, 1), phrase_font
|
||||||
|
)
|
||||||
|
|
||||||
|
# Score centered low, with the scoring/winning side's digit pulsing in a
|
||||||
|
# highlight color so the change reads at a glance.
|
||||||
|
away_text = str(celebration["away_score"])
|
||||||
|
home_text = str(celebration["home_score"])
|
||||||
|
score_font = self.fonts["score"]
|
||||||
|
segments = [
|
||||||
|
(away_text, celebration["scored_side"] == "away"),
|
||||||
|
("-", False),
|
||||||
|
(home_text, celebration["scored_side"] == "home"),
|
||||||
|
]
|
||||||
|
total_width = sum(draw.textlength(seg, font=score_font) for seg, _ in segments)
|
||||||
|
highlight = (255, 255, 0) if int(elapsed * 4) % 2 == 0 else (255, 170, 0)
|
||||||
|
x = (display_width - total_width) // 2
|
||||||
|
y = display_height - 14
|
||||||
|
for seg, is_highlight in segments:
|
||||||
|
color = highlight if is_highlight else (255, 255, 255)
|
||||||
|
self._draw_text_with_outline(draw, seg, (int(x), y), score_font, fill=color)
|
||||||
|
x += draw.textlength(seg, font=score_font)
|
||||||
|
|
||||||
|
main_img = Image.alpha_composite(main_img, overlay).convert("RGB")
|
||||||
|
self.display_manager.image = main_img
|
||||||
|
self.display_manager.update_display()
|
||||||
|
|
||||||
|
def display(self, force_clear: bool = False) -> bool:
|
||||||
|
"""Render an active celebration as a full-screen takeover; otherwise
|
||||||
|
defer to the normal live scorebug."""
|
||||||
|
if not self.is_enabled:
|
||||||
|
return False
|
||||||
|
celebration = self.active_celebration
|
||||||
|
if celebration:
|
||||||
|
if self.has_active_celebration():
|
||||||
|
try:
|
||||||
|
self._draw_celebration_layout(celebration, force_clear)
|
||||||
|
return True
|
||||||
|
except Exception as e:
|
||||||
|
self.logger.error(
|
||||||
|
f"[Celebrations] Error drawing celebration: {e}", exc_info=True
|
||||||
|
)
|
||||||
|
# Disarm rather than retry: the same render would fail on
|
||||||
|
# every frame for the rest of the window, logging a
|
||||||
|
# traceback each time and leaving the scorebug off screen.
|
||||||
|
self.active_celebration = None
|
||||||
|
self.last_game_switch = time.time()
|
||||||
|
else:
|
||||||
|
self.active_celebration = None
|
||||||
|
# Reset the dwell so the scorebug resumes on the scoring/winning
|
||||||
|
# game for a full duration before rotation can move on.
|
||||||
|
self.last_game_switch = time.time()
|
||||||
|
return super().display(force_clear)
|
||||||
@@ -0,0 +1,246 @@
|
|||||||
|
"""Live-rotation strategies — which live game to show next.
|
||||||
|
|
||||||
|
The nine plugin copies grew three spellings of this, and the survey behind
|
||||||
|
``docs/SPORTS_UNIFICATION.md`` found they are all the *same* Smooth Weighted
|
||||||
|
Round-Robin algorithm in two shapes:
|
||||||
|
|
||||||
|
* an **incremental picker** that holds weight state across calls and answers
|
||||||
|
"what next?" one game at a time (afl / nrl / soccer's ``_swrr_advance``), and
|
||||||
|
* a **precomputed cycle** that returns a full list of game ids up front
|
||||||
|
(football / baseball / basketball's ``_build_weighted_schedule`` and hockey's
|
||||||
|
``_build_rotation_schedule``, which differ only in loop shape).
|
||||||
|
|
||||||
|
They agree *within* a cycle — SWRR is deterministic — and differ only at cycle
|
||||||
|
boundaries, where the incremental form has no seam and the precomputed form
|
||||||
|
restarts. That is a real behavioral difference, so core ships both rather than
|
||||||
|
declaring a winner, and a plugin picks one by name:
|
||||||
|
|
||||||
|
self.rotation = get_rotation_strategy("swrr", weight_for=self._live_weight)
|
||||||
|
|
||||||
|
Core never learns which sport is asking. A plugin with a genuinely novel
|
||||||
|
ordering registers its own strategy instead of core growing a branch::
|
||||||
|
|
||||||
|
register_rotation_strategy("my-order", MyRotation)
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import Callable, Dict, List, Optional, Type
|
||||||
|
|
||||||
|
|
||||||
|
def _game_id(game: Dict) -> Optional[str]:
|
||||||
|
"""The rotation key for a game, or None if it has no usable id."""
|
||||||
|
return game.get("id")
|
||||||
|
|
||||||
|
|
||||||
|
class RotationStrategy:
|
||||||
|
"""Base class for live-rotation ordering.
|
||||||
|
|
||||||
|
Subclasses implement :meth:`schedule`; :meth:`next_game` has a working
|
||||||
|
default derived from it. Strategies whose natural shape is incremental
|
||||||
|
override :meth:`next_game` instead and derive :meth:`schedule`.
|
||||||
|
|
||||||
|
:param weight_for: callable mapping a game dict to a positive integer
|
||||||
|
weight — how many turns it gets per turn of a weight-1 game. Supplied by
|
||||||
|
the host so the *favorites* policy stays with the plugin and this module
|
||||||
|
stays free of any notion of what a favorite is. Defaults to equal
|
||||||
|
weights, which makes every strategy a plain round robin.
|
||||||
|
"""
|
||||||
|
|
||||||
|
#: Name this strategy is registered under. Set by :func:`register_rotation_strategy`.
|
||||||
|
name: str = ""
|
||||||
|
|
||||||
|
#: Ceiling on a per-game weight. A cycle is ``sum(weights)`` long and each
|
||||||
|
#: step scans every game, so an unbounded weight — a misread config field,
|
||||||
|
#: say — would spin the display thread for an unbounded time. On a Pi that
|
||||||
|
#: stalls rendering outright, so the bound is clamped like the floor is.
|
||||||
|
MAX_WEIGHT = 16
|
||||||
|
|
||||||
|
def __init__(self, weight_for: Optional[Callable[[Dict], int]] = None):
|
||||||
|
self._weight_for = weight_for or (lambda game: 1)
|
||||||
|
|
||||||
|
def weights(self, games: List[Dict]) -> Dict[str, int]:
|
||||||
|
"""``{game_id: weight}`` for games that have an id, in ``games`` order.
|
||||||
|
|
||||||
|
A weight below 1 is clamped up: a zero or negative weight would starve
|
||||||
|
a game out of the rotation entirely, which no caller means to express
|
||||||
|
and which would make ``total_weight`` collapse. It is clamped down at
|
||||||
|
:attr:`MAX_WEIGHT` for the reason documented there.
|
||||||
|
"""
|
||||||
|
weights: Dict[str, int] = {}
|
||||||
|
for game in games:
|
||||||
|
gid = _game_id(game)
|
||||||
|
if gid is None:
|
||||||
|
continue
|
||||||
|
try:
|
||||||
|
weight = int(self._weight_for(game))
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
weight = 1
|
||||||
|
weights[gid] = min(self.MAX_WEIGHT, max(1, weight))
|
||||||
|
return weights
|
||||||
|
|
||||||
|
def schedule(self, games: List[Dict]) -> List[str]:
|
||||||
|
"""Game ids in display order for one cycle. Ids may repeat."""
|
||||||
|
raise NotImplementedError
|
||||||
|
|
||||||
|
def next_game(self, games: List[Dict]) -> Optional[Dict]:
|
||||||
|
"""The next game to display, or None when there is nothing to show."""
|
||||||
|
order = self.schedule(games)
|
||||||
|
if not order:
|
||||||
|
return None
|
||||||
|
by_id = {gid: g for g in games if (gid := _game_id(g)) is not None}
|
||||||
|
return by_id.get(order[0])
|
||||||
|
|
||||||
|
def reset(self) -> None:
|
||||||
|
"""Drop any accumulated state. Stateless strategies need do nothing."""
|
||||||
|
|
||||||
|
|
||||||
|
class SimpleRotation(RotationStrategy):
|
||||||
|
"""Plain round robin: every live game once per cycle, weights ignored.
|
||||||
|
|
||||||
|
The fallback for a plugin that wants strictly even rotation regardless of
|
||||||
|
favorites.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def schedule(self, games: List[Dict]) -> List[str]:
|
||||||
|
return [gid for g in games if (gid := _game_id(g)) is not None]
|
||||||
|
|
||||||
|
|
||||||
|
class WeightedCycleRotation(RotationStrategy):
|
||||||
|
"""Precomputed SWRR cycle — the football / baseball / basketball / hockey shape.
|
||||||
|
|
||||||
|
Returns a full cycle of ``sum(weights)`` ids with repeats spaced evenly
|
||||||
|
rather than clumped, highest weight scheduled first. When no game carries a
|
||||||
|
boost the cycle degenerates to a single pass in ``games`` order, which is
|
||||||
|
exactly the plain round robin it replaced.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def schedule(self, games: List[Dict]) -> List[str]:
|
||||||
|
weights = self.weights(games)
|
||||||
|
if not weights:
|
||||||
|
return []
|
||||||
|
total_weight = sum(weights.values())
|
||||||
|
if total_weight <= len(weights):
|
||||||
|
# No boost in effect — plain order, one pass. (Also the guard that
|
||||||
|
# keeps the loop below from being O(total_weight) for nothing.)
|
||||||
|
return list(weights)
|
||||||
|
|
||||||
|
current = {gid: 0 for gid in weights}
|
||||||
|
order: List[str] = []
|
||||||
|
for _ in range(total_weight):
|
||||||
|
for gid, weight in weights.items():
|
||||||
|
current[gid] += weight
|
||||||
|
picked = max(current, key=lambda gid: current[gid])
|
||||||
|
current[picked] -= total_weight
|
||||||
|
order.append(picked)
|
||||||
|
return order
|
||||||
|
|
||||||
|
|
||||||
|
class SmoothWeightedRotation(RotationStrategy):
|
||||||
|
"""Incremental SWRR — the afl / nrl / soccer shape.
|
||||||
|
|
||||||
|
Weight state persists across calls, so there is no fixed-length cycle and
|
||||||
|
therefore no clustering seam at a cycle boundary. A game seen for the first
|
||||||
|
time starts at weight 0 and receives its full weight on the next call, so a
|
||||||
|
favorite's game that has just gone live naturally wins the first pick after
|
||||||
|
it appears — "queued first on refresh" without a special-cased branch.
|
||||||
|
|
||||||
|
State for games no longer live is dropped on each call, so a long-running
|
||||||
|
board does not accumulate entries for finished games.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, weight_for: Optional[Callable[[Dict], int]] = None):
|
||||||
|
super().__init__(weight_for)
|
||||||
|
self._current: Dict[str, int] = {}
|
||||||
|
|
||||||
|
def reset(self) -> None:
|
||||||
|
self._current = {}
|
||||||
|
|
||||||
|
def next_game(self, games: List[Dict]) -> Optional[Dict]:
|
||||||
|
if not games:
|
||||||
|
return None
|
||||||
|
weights = self.weights(games)
|
||||||
|
if not weights:
|
||||||
|
return None
|
||||||
|
|
||||||
|
# Keep state only for games still live.
|
||||||
|
self._current = {
|
||||||
|
gid: value for gid, value in self._current.items() if gid in weights
|
||||||
|
}
|
||||||
|
for gid, weight in weights.items():
|
||||||
|
self._current[gid] = self._current.get(gid, 0) + weight
|
||||||
|
|
||||||
|
total_weight = sum(weights.values())
|
||||||
|
# Iterate in `games` order so ties break toward the feed's ordering,
|
||||||
|
# which is what the plugin copies did and what makes the no-boost case
|
||||||
|
# identical to a plain round robin.
|
||||||
|
ids_in_order = [gid for g in games if (gid := _game_id(g)) in weights]
|
||||||
|
best = max(ids_in_order, key=lambda gid: self._current[gid])
|
||||||
|
self._current[best] -= total_weight
|
||||||
|
return next(g for g in games if _game_id(g) == best)
|
||||||
|
|
||||||
|
def schedule(self, games: List[Dict]) -> List[str]:
|
||||||
|
"""One cycle's worth of picks, without disturbing live state.
|
||||||
|
|
||||||
|
Derived by running the picker forward on a copy, so the returned order
|
||||||
|
is exactly what repeated :meth:`next_game` calls would produce from the
|
||||||
|
current state — callers can use it to preview or log the rotation
|
||||||
|
without perturbing it.
|
||||||
|
"""
|
||||||
|
weights = self.weights(games)
|
||||||
|
if not weights:
|
||||||
|
return []
|
||||||
|
# type(self), not this class: a subclass that overrides next_game must
|
||||||
|
# be previewed through its own ordering, or the returned order is not
|
||||||
|
# the one repeated next_game calls would produce — which is exactly
|
||||||
|
# what this method promises.
|
||||||
|
preview = type(self)(self._weight_for)
|
||||||
|
preview._current = dict(self._current)
|
||||||
|
order: List[str] = []
|
||||||
|
for _ in range(sum(weights.values())):
|
||||||
|
picked = preview.next_game(games)
|
||||||
|
if picked is None:
|
||||||
|
break
|
||||||
|
order.append(_game_id(picked))
|
||||||
|
return order
|
||||||
|
|
||||||
|
|
||||||
|
_REGISTRY: Dict[str, Type[RotationStrategy]] = {}
|
||||||
|
|
||||||
|
|
||||||
|
def register_rotation_strategy(name: str, factory: Type[RotationStrategy]) -> None:
|
||||||
|
"""Register a rotation strategy under ``name``.
|
||||||
|
|
||||||
|
When a plugin needs an ordering that core does not ship, it registers its
|
||||||
|
own here instead of core growing a sport-specific branch. Re-registering a
|
||||||
|
name replaces it, so a plugin may also override a built-in for itself.
|
||||||
|
"""
|
||||||
|
if not name:
|
||||||
|
raise ValueError("rotation strategy name must be a non-empty string")
|
||||||
|
# Fail at registration, not at the first schedule() call several frames
|
||||||
|
# later, where the cause is no longer on the stack.
|
||||||
|
if not (isinstance(factory, type) and issubclass(factory, RotationStrategy)):
|
||||||
|
raise TypeError(
|
||||||
|
f"rotation strategy {name!r} must be a RotationStrategy subclass, "
|
||||||
|
f"got {factory!r}"
|
||||||
|
)
|
||||||
|
factory.name = name
|
||||||
|
_REGISTRY[name] = factory
|
||||||
|
|
||||||
|
|
||||||
|
def get_rotation_strategy(
|
||||||
|
name: str, weight_for: Optional[Callable[[Dict], int]] = None
|
||||||
|
) -> RotationStrategy:
|
||||||
|
"""Build the strategy registered under ``name``.
|
||||||
|
|
||||||
|
Falls back to ``"simple"`` for an unknown name rather than raising: the name
|
||||||
|
arrives from user config, and a typo should cost the boost, not the
|
||||||
|
scoreboard.
|
||||||
|
"""
|
||||||
|
factory = _REGISTRY.get(name) or _REGISTRY["simple"]
|
||||||
|
return factory(weight_for=weight_for)
|
||||||
|
|
||||||
|
|
||||||
|
register_rotation_strategy("simple", SimpleRotation)
|
||||||
|
register_rotation_strategy("weighted", WeightedCycleRotation)
|
||||||
|
register_rotation_strategy("swrr", SmoothWeightedRotation)
|
||||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -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
|
||||||
@@ -79,9 +79,10 @@ def _competitor(abbr, team_id, score, home_away, record="30-10-5"):
|
|||||||
"logo": None,
|
"logo": None,
|
||||||
},
|
},
|
||||||
"records": [{"summary": record}],
|
"records": [{"summary": record}],
|
||||||
# The hockey extractor iterates competitor["statistics"] and
|
# The hockey extractor reads competitor["statistics"] for shot counts;
|
||||||
# returns None for the whole event when the key is absent (see
|
# it now defaults to an empty list when the key is absent rather than
|
||||||
# test_hockey_event_without_statistics_returns_none).
|
# dropping the whole event (see
|
||||||
|
# test_hockey_event_without_statistics_still_extracts).
|
||||||
"statistics": [],
|
"statistics": [],
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -305,15 +306,25 @@ class TestExtractGameDetailsContract:
|
|||||||
assert details["home_shots"] == 0
|
assert details["home_shots"] == 0
|
||||||
assert details["away_shots"] == 0
|
assert details["away_shots"] == 0
|
||||||
|
|
||||||
def test_hockey_event_without_statistics_returns_none(self):
|
def test_hockey_event_without_statistics_still_extracts(self):
|
||||||
# PINNED AS-IS: the hockey extractor unconditionally iterates
|
# FIXED (was pinned as returning None): the hockey extractor used to
|
||||||
# competitor["statistics"]; a competitor without the key raises
|
# iterate competitor["statistics"] unguarded, so a competitor without
|
||||||
# KeyError internally and the WHOLE event is dropped (returns
|
# the key raised KeyError internally and the WHOLE event was dropped
|
||||||
# None), even though scores/status are present.
|
# despite valid scores and status. It now defaults to an empty list,
|
||||||
|
# matching the behaviour already shipped in the hockey plugin, so the
|
||||||
|
# event survives with zeroed shot counts -- the same values
|
||||||
|
# test_hockey_live_power_play_and_default_shots already expects for an
|
||||||
|
# EMPTY statistics array.
|
||||||
event = make_event("410", "in", "2026-01-15T18:30:00Z")
|
event = make_event("410", "in", "2026-01-15T18:30:00Z")
|
||||||
for comp in event["competitions"][0]["competitors"]:
|
for comp in event["competitions"][0]["competitors"]:
|
||||||
del comp["statistics"]
|
del comp["statistics"]
|
||||||
assert extract(Hockey, event) is None
|
details = extract(Hockey, event)
|
||||||
|
assert details is not None
|
||||||
|
assert details["home_abbr"] == "TB"
|
||||||
|
assert details["away_abbr"] == "DAL"
|
||||||
|
assert details["home_score"] == "3"
|
||||||
|
assert details["home_shots"] == 0
|
||||||
|
assert details["away_shots"] == 0
|
||||||
|
|
||||||
def test_baseball_live_inning_and_count(self):
|
def test_baseball_live_inning_and_count(self):
|
||||||
event = make_event(
|
event = make_event(
|
||||||
@@ -337,14 +348,35 @@ class TestExtractGameDetailsContract:
|
|||||||
assert details["status"] == "status_in_progress"
|
assert details["status"] == "status_in_progress"
|
||||||
assert details["series_summary"] == ""
|
assert details["series_summary"] == ""
|
||||||
|
|
||||||
def test_baseball_live_without_top_level_status_returns_none(self):
|
def test_baseball_live_without_top_level_status_still_extracts(self):
|
||||||
# PINNED AS-IS: for live games the baseball extractor reads
|
# FIXED (was pinned as returning None): the baseball extractor read
|
||||||
# game_event["status"] (the event TOP-LEVEL status, not the
|
# game_event["status"] -- the event TOP-LEVEL status -- for the
|
||||||
# competition status) for the inning; an otherwise-valid live
|
# inning, so an otherwise-valid live event lacking that duplicate key
|
||||||
# event lacking that duplicate key is dropped entirely.
|
# was dropped entirely. Real ESPN events carry status in both places,
|
||||||
event = make_event("412", "in", "2026-07-16T23:05:00Z")
|
# but MiLB events (synthesized from the MLB Stats API into an
|
||||||
|
# ESPN-like shape) populate only the competition-level one. It now
|
||||||
|
# reads the competition-level `status` that
|
||||||
|
# _extract_game_details_common has already validated, so it can never
|
||||||
|
# be missing at that point.
|
||||||
|
event = make_event("412", "in", "2026-07-16T23:05:00Z", period=7)
|
||||||
del event["status"]
|
del event["status"]
|
||||||
assert extract(Baseball, event) is None
|
details = extract(Baseball, event)
|
||||||
|
assert details is not None
|
||||||
|
assert details["inning"] == 7
|
||||||
|
|
||||||
|
def test_baseball_live_without_top_level_status_extracts_for_favorites(self):
|
||||||
|
# The favourite-team branch logs the status payload for diagnostics and
|
||||||
|
# read the same event top-level key the test above proves can be
|
||||||
|
# absent. So the identical MiLB event that extracts fine for a
|
||||||
|
# non-favourite raised KeyError and was dropped once the team WAS a
|
||||||
|
# favourite -- the worst shape for the bug, since it only hit the games
|
||||||
|
# the user cared most about, and only on the diagnostic path that was
|
||||||
|
# supposed to help debug them.
|
||||||
|
event = make_event("413", "in", "2026-07-16T23:05:00Z", period=7)
|
||||||
|
del event["status"]
|
||||||
|
details = extract(Baseball, event, favorites=["TB"])
|
||||||
|
assert details is not None
|
||||||
|
assert details["inning"] == 7
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
@@ -400,7 +432,7 @@ def build_manager(monkeypatch, tmp_path):
|
|||||||
monkeypatch.setattr(
|
monkeypatch.setattr(
|
||||||
SportsCore, "_initialize_logo_dir", lambda self, configured: tmp_path)
|
SportsCore, "_initialize_logo_dir", lambda self, configured: tmp_path)
|
||||||
monkeypatch.setattr(
|
monkeypatch.setattr(
|
||||||
"src.base_classes.sports.get_background_service",
|
"src.base_classes.sports.core.get_background_service",
|
||||||
lambda *args, **kwargs: MagicMock())
|
lambda *args, **kwargs: MagicMock())
|
||||||
|
|
||||||
# Rig requests.Session BEFORE any manager is built. Construction creates
|
# Rig requests.Session BEFORE any manager is built. Construction creates
|
||||||
|
|||||||
@@ -0,0 +1,880 @@
|
|||||||
|
"""Tests for the opt-in sports capabilities (phase B2).
|
||||||
|
|
||||||
|
Two properties matter beyond "the code works":
|
||||||
|
|
||||||
|
1. **Opting out is structural.** A mode class that does not mix in
|
||||||
|
``CelebrationMixin`` must have none of its attributes or methods — not
|
||||||
|
merely a disabled flag. ``TestOptOutIsStructural`` asserts that directly,
|
||||||
|
because it is the property the whole mixin design exists to buy.
|
||||||
|
|
||||||
|
2. **The promoted behavior matches the plugin copies.** These bodies came from
|
||||||
|
afl/soccer/nrl (goal dialect) and football (score dialect); the tests pin
|
||||||
|
the reconciled behavior of both, including the three seams where the
|
||||||
|
lineages genuinely disagreed.
|
||||||
|
|
||||||
|
See docs/SPORTS_UNIFICATION.md.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import sys
|
||||||
|
import time
|
||||||
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
sys.modules.setdefault("rgbmatrix", MagicMock())
|
||||||
|
|
||||||
|
from src.base_classes.sports.capabilities import ( # noqa: E402
|
||||||
|
CelebrationMixin,
|
||||||
|
RotationStrategy,
|
||||||
|
SimpleRotation,
|
||||||
|
SmoothWeightedRotation,
|
||||||
|
WeightedCycleRotation,
|
||||||
|
get_rotation_strategy,
|
||||||
|
register_rotation_strategy,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def game(gid, home="HOM", away="AWY", home_score=0, away_score=0, **extra):
|
||||||
|
g = {
|
||||||
|
"id": gid,
|
||||||
|
"home_abbr": home,
|
||||||
|
"away_abbr": away,
|
||||||
|
"home_id": f"{gid}-h",
|
||||||
|
"away_id": f"{gid}-a",
|
||||||
|
"home_score": home_score,
|
||||||
|
"away_score": away_score,
|
||||||
|
}
|
||||||
|
g.update(extra)
|
||||||
|
return g
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Rotation strategies
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def boost(favorites, factor=3):
|
||||||
|
"""A weight_for callable of the shape the plugins supply."""
|
||||||
|
return lambda g: factor if g.get("home_abbr") in favorites else 1
|
||||||
|
|
||||||
|
|
||||||
|
class TestRegistry:
|
||||||
|
@pytest.mark.parametrize("name,cls", [
|
||||||
|
("simple", SimpleRotation),
|
||||||
|
("weighted", WeightedCycleRotation),
|
||||||
|
("swrr", SmoothWeightedRotation),
|
||||||
|
])
|
||||||
|
def test_builtin_names_resolve(self, name, cls):
|
||||||
|
assert isinstance(get_rotation_strategy(name), cls)
|
||||||
|
|
||||||
|
def test_unknown_name_falls_back_to_simple(self):
|
||||||
|
"""The name comes from user config; a typo should cost the boost, not
|
||||||
|
the scoreboard."""
|
||||||
|
assert isinstance(get_rotation_strategy("typo"), SimpleRotation)
|
||||||
|
|
||||||
|
def test_a_plugin_can_register_its_own(self):
|
||||||
|
class MyRotation(SimpleRotation):
|
||||||
|
pass
|
||||||
|
|
||||||
|
register_rotation_strategy("test-only", MyRotation)
|
||||||
|
try:
|
||||||
|
assert isinstance(get_rotation_strategy("test-only"), MyRotation)
|
||||||
|
assert MyRotation.name == "test-only"
|
||||||
|
finally:
|
||||||
|
from src.base_classes.sports.capabilities import rotation
|
||||||
|
rotation._REGISTRY.pop("test-only", None)
|
||||||
|
|
||||||
|
def test_empty_name_is_rejected(self):
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
register_rotation_strategy("", SimpleRotation)
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("bad", [
|
||||||
|
SimpleRotation(), # an instance, not the class
|
||||||
|
str, # unrelated class
|
||||||
|
lambda **kw: None, # a factory function
|
||||||
|
])
|
||||||
|
def test_a_non_strategy_factory_is_rejected(self, bad):
|
||||||
|
"""Fail at registration, not several frames later inside schedule(),
|
||||||
|
where the cause is no longer on the stack."""
|
||||||
|
with pytest.raises(TypeError):
|
||||||
|
register_rotation_strategy("bad-factory", bad)
|
||||||
|
|
||||||
|
def test_weight_for_is_optional(self):
|
||||||
|
"""Default weights are equal, so every strategy degenerates to a plain
|
||||||
|
round robin — the pre-boost behavior."""
|
||||||
|
games = [game("a"), game("b"), game("c")]
|
||||||
|
for name in ("simple", "weighted", "swrr"):
|
||||||
|
assert get_rotation_strategy(name).schedule(games) == ["a", "b", "c"]
|
||||||
|
|
||||||
|
|
||||||
|
class TestWeights:
|
||||||
|
def test_games_without_an_id_are_skipped(self):
|
||||||
|
strategy = get_rotation_strategy("weighted")
|
||||||
|
assert strategy.weights([game("a"), {"home_abbr": "X"}]) == {"a": 1}
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("bad", [0, -5])
|
||||||
|
def test_non_positive_weights_are_clamped_to_one(self, bad):
|
||||||
|
"""A zero weight would starve the game out of the rotation entirely and
|
||||||
|
collapse total_weight — no caller means that."""
|
||||||
|
strategy = get_rotation_strategy("weighted", weight_for=lambda g: bad)
|
||||||
|
assert strategy.weights([game("a")]) == {"a": 1}
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("bad", [None, "three", object()])
|
||||||
|
def test_unusable_weights_fall_back_to_one(self, bad):
|
||||||
|
strategy = get_rotation_strategy("weighted", weight_for=lambda g: bad)
|
||||||
|
assert strategy.weights([game("a")]) == {"a": 1}
|
||||||
|
|
||||||
|
def test_huge_weights_are_clamped(self):
|
||||||
|
"""A cycle is sum(weights) long and each step scans every game, so an
|
||||||
|
unbounded weight from a misread config spins the display thread."""
|
||||||
|
strategy = get_rotation_strategy("weighted", weight_for=lambda g: 10_000)
|
||||||
|
assert strategy.weights([game("a")]) == {"a": RotationStrategy.MAX_WEIGHT}
|
||||||
|
|
||||||
|
def test_a_clamped_cycle_stays_bounded(self):
|
||||||
|
strategy = get_rotation_strategy("weighted", weight_for=lambda g: 10_000)
|
||||||
|
order = strategy.schedule([game("a"), game("b")])
|
||||||
|
assert len(order) == 2 * RotationStrategy.MAX_WEIGHT
|
||||||
|
|
||||||
|
|
||||||
|
class TestSimpleRotation:
|
||||||
|
def test_one_pass_in_feed_order(self):
|
||||||
|
games = [game("a"), game("b"), game("c")]
|
||||||
|
assert SimpleRotation().schedule(games) == ["a", "b", "c"]
|
||||||
|
|
||||||
|
def test_weights_are_ignored(self):
|
||||||
|
games = [game("a", home="FAV"), game("b")]
|
||||||
|
strategy = SimpleRotation(weight_for=boost({"FAV"}, 5))
|
||||||
|
assert strategy.schedule(games) == ["a", "b"]
|
||||||
|
|
||||||
|
def test_empty(self):
|
||||||
|
assert SimpleRotation().schedule([]) == []
|
||||||
|
assert SimpleRotation().next_game([]) is None
|
||||||
|
|
||||||
|
|
||||||
|
class TestWeightedCycleRotation:
|
||||||
|
def test_no_boost_is_a_single_pass(self):
|
||||||
|
games = [game("a"), game("b"), game("c")]
|
||||||
|
strategy = WeightedCycleRotation(weight_for=boost({"NONE"}))
|
||||||
|
assert strategy.schedule(games) == ["a", "b", "c"]
|
||||||
|
|
||||||
|
def test_favorite_gets_boost_many_slots(self):
|
||||||
|
games = [game("a", home="FAV"), game("b")]
|
||||||
|
order = WeightedCycleRotation(weight_for=boost({"FAV"}, 3)).schedule(games)
|
||||||
|
assert len(order) == 4
|
||||||
|
assert order.count("a") == 3
|
||||||
|
assert order.count("b") == 1
|
||||||
|
|
||||||
|
def test_repeats_are_spaced_not_clumped(self):
|
||||||
|
"""The point of SWRR over naive repetition: 'aaab' is what we must NOT
|
||||||
|
produce."""
|
||||||
|
games = [game("a", home="FAV"), game("b")]
|
||||||
|
order = WeightedCycleRotation(weight_for=boost({"FAV"}, 3)).schedule(games)
|
||||||
|
assert order != ["a", "a", "a", "b"]
|
||||||
|
assert order[0] == "a", "highest weight is scheduled first"
|
||||||
|
|
||||||
|
def test_is_stateless_across_calls(self):
|
||||||
|
games = [game("a", home="FAV"), game("b")]
|
||||||
|
strategy = WeightedCycleRotation(weight_for=boost({"FAV"}, 3))
|
||||||
|
assert strategy.schedule(games) == strategy.schedule(games)
|
||||||
|
|
||||||
|
def test_next_game_returns_the_first_of_the_cycle(self):
|
||||||
|
games = [game("a"), game("b", home="FAV")]
|
||||||
|
strategy = WeightedCycleRotation(weight_for=boost({"FAV"}, 4))
|
||||||
|
assert strategy.next_game(games)["id"] == "b"
|
||||||
|
|
||||||
|
def test_empty(self):
|
||||||
|
assert WeightedCycleRotation().schedule([]) == []
|
||||||
|
|
||||||
|
|
||||||
|
class TestSmoothWeightedRotation:
|
||||||
|
def test_no_boost_is_plain_round_robin(self):
|
||||||
|
games = [game("a"), game("b"), game("c")]
|
||||||
|
strategy = SmoothWeightedRotation()
|
||||||
|
assert [strategy.next_game(games)["id"] for _ in range(6)] == [
|
||||||
|
"a", "b", "c", "a", "b", "c"]
|
||||||
|
|
||||||
|
def test_favorite_wins_the_share_over_a_long_run(self):
|
||||||
|
games = [game("a", home="FAV"), game("b")]
|
||||||
|
strategy = SmoothWeightedRotation(weight_for=boost({"FAV"}, 3))
|
||||||
|
picks = [strategy.next_game(games)["id"] for _ in range(40)]
|
||||||
|
assert picks.count("a") == 30
|
||||||
|
assert picks.count("b") == 10
|
||||||
|
|
||||||
|
def test_no_clustering_seam_across_cycle_boundaries(self):
|
||||||
|
"""The property that motivates keeping this strategy separate from the
|
||||||
|
precomputed one: state persists, so there is no restart every N picks
|
||||||
|
and therefore no place where repeats bunch up."""
|
||||||
|
games = [game("a", home="FAV"), game("b")]
|
||||||
|
strategy = SmoothWeightedRotation(weight_for=boost({"FAV"}, 3))
|
||||||
|
picks = [strategy.next_game(games)["id"] for _ in range(40)]
|
||||||
|
assert "aaaa" not in "".join(picks)
|
||||||
|
|
||||||
|
def test_a_new_favorite_is_queued_first(self):
|
||||||
|
"""A favorite's game that has just gone live starts at weight 0, gets
|
||||||
|
its full weight on the next call, and so wins the first pick after it
|
||||||
|
appears — without a special-cased branch."""
|
||||||
|
games = [game("a"), game("b")]
|
||||||
|
strategy = SmoothWeightedRotation(weight_for=boost({"FAV"}, 5))
|
||||||
|
for _ in range(3):
|
||||||
|
strategy.next_game(games)
|
||||||
|
games.append(game("c", home="FAV"))
|
||||||
|
assert strategy.next_game(games)["id"] == "c"
|
||||||
|
|
||||||
|
def test_state_for_games_no_longer_live_is_dropped(self):
|
||||||
|
games = [game("a"), game("b")]
|
||||||
|
strategy = SmoothWeightedRotation()
|
||||||
|
strategy.next_game(games)
|
||||||
|
strategy.next_game([game("a")])
|
||||||
|
assert set(strategy._current) == {"a"}
|
||||||
|
|
||||||
|
def test_reset_clears_state(self):
|
||||||
|
games = [game("a"), game("b")]
|
||||||
|
strategy = SmoothWeightedRotation()
|
||||||
|
strategy.next_game(games)
|
||||||
|
strategy.reset()
|
||||||
|
assert strategy._current == {}
|
||||||
|
assert strategy.next_game(games)["id"] == "a"
|
||||||
|
|
||||||
|
def test_schedule_previews_without_perturbing_state(self):
|
||||||
|
games = [game("a", home="FAV"), game("b")]
|
||||||
|
strategy = SmoothWeightedRotation(weight_for=boost({"FAV"}, 3))
|
||||||
|
preview = strategy.schedule(games)
|
||||||
|
actual = [strategy.next_game(games)["id"] for _ in range(len(preview))]
|
||||||
|
assert preview == actual
|
||||||
|
|
||||||
|
def test_preview_uses_the_subclass_ordering(self):
|
||||||
|
"""schedule() promises the order repeated next_game calls produce. Built
|
||||||
|
from the base class, a subclass that overrides next_game gets a preview
|
||||||
|
of the wrong algorithm."""
|
||||||
|
class Reversed(SmoothWeightedRotation):
|
||||||
|
def next_game(self, games):
|
||||||
|
return super().next_game(list(reversed(games)))
|
||||||
|
|
||||||
|
games = [game("a"), game("b"), game("c")]
|
||||||
|
strategy = Reversed()
|
||||||
|
preview = strategy.schedule(games)
|
||||||
|
actual = [strategy.next_game(games)["id"] for _ in range(len(preview))]
|
||||||
|
assert preview == actual
|
||||||
|
|
||||||
|
def test_empty(self):
|
||||||
|
assert SmoothWeightedRotation().next_game([]) is None
|
||||||
|
assert SmoothWeightedRotation().schedule([]) == []
|
||||||
|
|
||||||
|
def test_games_without_ids_are_ignored(self):
|
||||||
|
assert SmoothWeightedRotation().next_game([{"home_abbr": "X"}]) is None
|
||||||
|
|
||||||
|
|
||||||
|
class TestStrategiesAgreeWithinACycle:
|
||||||
|
"""The survey's core finding: the 'three dialects' are one algorithm. They
|
||||||
|
must produce the same order within a cycle; they differ only at the
|
||||||
|
boundary, which is why both shapes survive."""
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("factor", [2, 3, 5])
|
||||||
|
def test_first_cycle_matches(self, factor):
|
||||||
|
games = [game("a", home="FAV"), game("b"), game("c")]
|
||||||
|
weight_for = boost({"FAV"}, factor)
|
||||||
|
assert (SmoothWeightedRotation(weight_for=weight_for).schedule(games)
|
||||||
|
== WeightedCycleRotation(weight_for=weight_for).schedule(games))
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Differential: core strategies vs. the plugin implementations they replace
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
BOOST = 3
|
||||||
|
|
||||||
|
|
||||||
|
def _is_fav(g):
|
||||||
|
return g.get("home_abbr") == "FAV"
|
||||||
|
|
||||||
|
|
||||||
|
def _weight_for(g):
|
||||||
|
return BOOST if _is_fav(g) else 1
|
||||||
|
|
||||||
|
|
||||||
|
class _PluginSwrr:
|
||||||
|
"""afl / nrl / soccer ``_swrr_advance``, transcribed verbatim."""
|
||||||
|
|
||||||
|
favorite_live_boost = BOOST
|
||||||
|
|
||||||
|
def _is_favorite_game(self, g):
|
||||||
|
return _is_fav(g)
|
||||||
|
|
||||||
|
def advance(self, games):
|
||||||
|
if not games:
|
||||||
|
return None
|
||||||
|
weights = {}
|
||||||
|
for g in games:
|
||||||
|
gid = g.get("id")
|
||||||
|
if gid is None:
|
||||||
|
continue
|
||||||
|
weights[gid] = self.favorite_live_boost if self._is_favorite_game(g) else 1
|
||||||
|
if not weights:
|
||||||
|
return None
|
||||||
|
if not hasattr(self, "_swrr_weights"):
|
||||||
|
self._swrr_weights = {}
|
||||||
|
self._swrr_weights = {
|
||||||
|
gid: w for gid, w in self._swrr_weights.items() if gid in weights}
|
||||||
|
for gid, w in weights.items():
|
||||||
|
self._swrr_weights[gid] = self._swrr_weights.get(gid, 0) + w
|
||||||
|
total_weight = sum(weights.values())
|
||||||
|
ids_in_order = [g.get("id") for g in games if g.get("id") in weights]
|
||||||
|
best_gid = max(ids_in_order, key=lambda gid: self._swrr_weights[gid])
|
||||||
|
self._swrr_weights[best_gid] -= total_weight
|
||||||
|
return next(g for g in games if g.get("id") == best_gid)
|
||||||
|
|
||||||
|
|
||||||
|
def _plugin_weighted_schedule(games):
|
||||||
|
"""football / baseball / basketball ``_build_weighted_schedule``, verbatim."""
|
||||||
|
if not games:
|
||||||
|
return []
|
||||||
|
weights = {g["id"]: (BOOST if _is_fav(g) else 1) for g in games}
|
||||||
|
total_weight = sum(weights.values())
|
||||||
|
if total_weight <= len(games):
|
||||||
|
return [g["id"] for g in games]
|
||||||
|
current_weight = {gid: 0 for gid in weights}
|
||||||
|
schedule = []
|
||||||
|
for _ in range(total_weight):
|
||||||
|
for gid in weights:
|
||||||
|
current_weight[gid] += weights[gid]
|
||||||
|
picked = max(current_weight, key=lambda gid: current_weight[gid])
|
||||||
|
current_weight[picked] -= total_weight
|
||||||
|
schedule.append(picked)
|
||||||
|
return schedule
|
||||||
|
|
||||||
|
|
||||||
|
def _plugin_rotation_schedule(games):
|
||||||
|
"""hockey ``_build_rotation_schedule``, transcribed verbatim."""
|
||||||
|
weights = [(g["id"], BOOST if _is_fav(g) else 1) for g in games]
|
||||||
|
total_weight = sum(w for _, w in weights)
|
||||||
|
if not weights or total_weight <= 0:
|
||||||
|
return [g["id"] for g in games]
|
||||||
|
current_weights = {gid: 0 for gid, _ in weights}
|
||||||
|
schedule = []
|
||||||
|
for _ in range(total_weight):
|
||||||
|
best_id, best_current = None, None
|
||||||
|
for gid, w in weights:
|
||||||
|
current_weights[gid] += w
|
||||||
|
if best_current is None or current_weights[gid] > best_current:
|
||||||
|
best_id, best_current = gid, current_weights[gid]
|
||||||
|
current_weights[best_id] -= total_weight
|
||||||
|
schedule.append(best_id)
|
||||||
|
return schedule
|
||||||
|
|
||||||
|
|
||||||
|
def _cases():
|
||||||
|
"""Every live-game shape up to 4 games: each either a favorite or not.
|
||||||
|
|
||||||
|
Exhaustive rather than random so the gate is deterministic — a rotation
|
||||||
|
regression must fail the same way on every run.
|
||||||
|
"""
|
||||||
|
import itertools
|
||||||
|
for size in range(1, 5):
|
||||||
|
for flags in itertools.product(("FAV", "OTH"), repeat=size):
|
||||||
|
yield [game(f"g{i}", home=abbr) for i, abbr in enumerate(flags)]
|
||||||
|
|
||||||
|
|
||||||
|
class TestMatchesThePluginImplementations:
|
||||||
|
"""The promotion is only safe if these reproduce the plugin copies exactly.
|
||||||
|
|
||||||
|
B5 deletes the bundled copies on the strength of this: each core strategy is
|
||||||
|
checked against the verbatim source it replaces, over every live-game shape
|
||||||
|
up to four games.
|
||||||
|
"""
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("games", list(_cases()))
|
||||||
|
def test_swrr_matches_the_incremental_plugin_picker(self, games):
|
||||||
|
plugin = _PluginSwrr()
|
||||||
|
core = SmoothWeightedRotation(weight_for=_weight_for)
|
||||||
|
# 60 picks: long enough to cross many cycle boundaries, where a
|
||||||
|
# state-handling divergence would show up.
|
||||||
|
assert ([plugin.advance(games)["id"] for _ in range(60)]
|
||||||
|
== [core.next_game(games)["id"] for _ in range(60)])
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("games", list(_cases()))
|
||||||
|
def test_weighted_matches_the_football_lineage(self, games):
|
||||||
|
assert (_plugin_weighted_schedule(games)
|
||||||
|
== WeightedCycleRotation(weight_for=_weight_for).schedule(games))
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("games", list(_cases()))
|
||||||
|
def test_weighted_matches_hockeys_loop_shape(self, games):
|
||||||
|
assert (_plugin_rotation_schedule(games)
|
||||||
|
== WeightedCycleRotation(weight_for=_weight_for).schedule(games))
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Celebrations
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class _FakeLive:
|
||||||
|
"""Stand-in for SportsLive: just the surface the mixin touches."""
|
||||||
|
|
||||||
|
def __init__(self, mode_config=None, favorite_teams=None):
|
||||||
|
self.mode_config = mode_config or {}
|
||||||
|
self.favorite_teams = favorite_teams or []
|
||||||
|
self.logger = MagicMock()
|
||||||
|
self.display_manager = MagicMock()
|
||||||
|
self.is_enabled = True
|
||||||
|
self.current_game = None
|
||||||
|
self.last_game_switch = 0
|
||||||
|
self.display_calls = []
|
||||||
|
|
||||||
|
def _favorite_key(self, game, side):
|
||||||
|
return game.get(f"{side}_abbr")
|
||||||
|
|
||||||
|
def display(self, force_clear=False):
|
||||||
|
self.display_calls.append(force_clear)
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
class _Celebrating(CelebrationMixin, _FakeLive):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class _Coalescing(CelebrationMixin, _FakeLive):
|
||||||
|
COALESCE_SCORING_SEQUENCE = True
|
||||||
|
|
||||||
|
def score_phrase(self, points, team_abbr):
|
||||||
|
return "TOUCHDOWN!" if points >= 6 else f"{team_abbr} FIELD GOAL!"
|
||||||
|
|
||||||
|
|
||||||
|
class _ById(CelebrationMixin, _FakeLive):
|
||||||
|
"""The nrl shape: ambiguous abbreviations, so favorites match on team id."""
|
||||||
|
|
||||||
|
def _favorite_key(self, game, side):
|
||||||
|
return game.get(f"{side}_id")
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def celebrating():
|
||||||
|
def _build(cls=_Celebrating, mode_config=None, favorites=None):
|
||||||
|
return cls(mode_config=mode_config, favorite_teams=favorites)
|
||||||
|
return _build
|
||||||
|
|
||||||
|
|
||||||
|
class TestOptOutIsStructural:
|
||||||
|
"""The property the mixin design exists to buy: a class that does not opt in
|
||||||
|
has none of this code — not a disabled flag, not an unused attribute."""
|
||||||
|
|
||||||
|
def test_a_non_celebrating_class_has_no_celebration_surface(self):
|
||||||
|
plain = _FakeLive()
|
||||||
|
for attribute in ("active_celebration", "_score_baselines",
|
||||||
|
"celebration_enabled", "celebration_duration",
|
||||||
|
"_check_for_score", "_check_for_win",
|
||||||
|
"has_active_celebration", "_draw_celebration_layout"):
|
||||||
|
assert not hasattr(plain, attribute), (
|
||||||
|
f"{attribute} leaked onto a class that never opted in")
|
||||||
|
|
||||||
|
def test_the_mixin_is_absent_from_a_non_celebrating_mro(self):
|
||||||
|
assert CelebrationMixin not in _FakeLive.__mro__
|
||||||
|
assert CelebrationMixin in _Celebrating.__mro__
|
||||||
|
|
||||||
|
def test_mixin_does_not_require_the_base_to_know_about_it(self):
|
||||||
|
"""SportsLive must carry no celebration hooks — that would be the
|
||||||
|
god-class shape the mixin replaces."""
|
||||||
|
from src.base_classes.sports import SportsLive
|
||||||
|
source = __import__("inspect").getsource(SportsLive)
|
||||||
|
assert "celebration" not in source.lower()
|
||||||
|
|
||||||
|
|
||||||
|
class TestCelebrationConfig:
|
||||||
|
def test_defaults(self, celebrating):
|
||||||
|
manager = celebrating()
|
||||||
|
assert manager.celebration_enabled is True
|
||||||
|
assert manager.celebration_duration == 8
|
||||||
|
assert manager.celebrate_opponent_scores is False
|
||||||
|
assert manager.active_celebration is None
|
||||||
|
|
||||||
|
def test_reads_the_goal_spelling_of_the_opponent_key(self, celebrating):
|
||||||
|
"""The soccer lineage's published schema says `celebrate_opponent_goals`;
|
||||||
|
adopting the mixin must not silently reset users' setting."""
|
||||||
|
manager = celebrating(mode_config={"celebrate_opponent_goals": True})
|
||||||
|
assert manager.celebrate_opponent_scores is True
|
||||||
|
|
||||||
|
def test_reads_the_score_spelling_of_the_opponent_key(self, celebrating):
|
||||||
|
manager = celebrating(mode_config={"celebrate_opponent_scores": True})
|
||||||
|
assert manager.celebrate_opponent_scores is True
|
||||||
|
|
||||||
|
def test_score_spelling_wins_when_both_are_present(self, celebrating):
|
||||||
|
manager = celebrating(mode_config={"celebrate_opponent_scores": False,
|
||||||
|
"celebrate_opponent_goals": True})
|
||||||
|
assert manager.celebrate_opponent_scores is False
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("bad", ["eight", None, {}, []])
|
||||||
|
def test_unusable_duration_falls_back(self, celebrating, bad):
|
||||||
|
"""The duration is compared numerically on the display path, outside
|
||||||
|
any try block — a string from a hand-edited config would propagate a
|
||||||
|
TypeError straight out of display()."""
|
||||||
|
manager = celebrating(mode_config={"celebration_duration": bad})
|
||||||
|
assert manager.celebration_duration == 8.0
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("bad", [0, -5])
|
||||||
|
def test_non_positive_duration_is_floored(self, celebrating, bad):
|
||||||
|
"""Zero or negative would arm a celebration that can never render."""
|
||||||
|
manager = celebrating(mode_config={"celebration_duration": bad})
|
||||||
|
assert manager.celebration_duration == 1.0
|
||||||
|
|
||||||
|
def test_numeric_string_duration_is_accepted(self, celebrating):
|
||||||
|
assert celebrating(
|
||||||
|
mode_config={"celebration_duration": "12"}).celebration_duration == 12.0
|
||||||
|
|
||||||
|
|
||||||
|
class TestScoreDetection:
|
||||||
|
def test_first_sighting_never_celebrates(self, celebrating):
|
||||||
|
"""A game already in progress at boot must not false-fire."""
|
||||||
|
manager = celebrating()
|
||||||
|
manager._check_for_score(game("g1", home_score=3, away_score=1))
|
||||||
|
assert manager.active_celebration is None
|
||||||
|
assert manager._score_baselines["g1"] == {"away": 1, "home": 3}
|
||||||
|
|
||||||
|
def test_increment_arms_a_celebration(self, celebrating):
|
||||||
|
manager = celebrating()
|
||||||
|
manager._check_for_score(game("g1", home_score=0, away_score=0))
|
||||||
|
manager._check_for_score(game("g1", home_score=1, away_score=0))
|
||||||
|
assert manager.active_celebration["kind"] == "score"
|
||||||
|
assert manager.active_celebration["scored_side"] == "home"
|
||||||
|
|
||||||
|
def test_no_change_does_not_fire(self, celebrating):
|
||||||
|
manager = celebrating()
|
||||||
|
manager._check_for_score(game("g1", home_score=2))
|
||||||
|
manager._check_for_score(game("g1", home_score=2))
|
||||||
|
assert manager.active_celebration is None
|
||||||
|
|
||||||
|
def test_decrement_rebases_silently(self, celebrating):
|
||||||
|
"""A disallowed goal / correction must not celebrate, and must not leave
|
||||||
|
a stale baseline that fires on the way back up."""
|
||||||
|
manager = celebrating()
|
||||||
|
manager._check_for_score(game("g1", home_score=2))
|
||||||
|
manager._check_for_score(game("g1", home_score=1))
|
||||||
|
assert manager.active_celebration is None
|
||||||
|
assert manager._score_baselines["g1"]["home"] == 1
|
||||||
|
|
||||||
|
def test_disabled_never_fires(self, celebrating):
|
||||||
|
manager = celebrating(mode_config={"celebration_enabled": False})
|
||||||
|
manager._check_for_score(game("g1", home_score=0))
|
||||||
|
manager._check_for_score(game("g1", home_score=1))
|
||||||
|
assert manager.active_celebration is None
|
||||||
|
|
||||||
|
def test_game_without_an_id_is_ignored(self, celebrating):
|
||||||
|
manager = celebrating()
|
||||||
|
manager._check_for_score({"home_score": 1, "away_score": 0})
|
||||||
|
assert manager.active_celebration is None
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("score", [None, "", "not-a-number-at-all"])
|
||||||
|
def test_unusable_scores_are_ignored(self, celebrating, score):
|
||||||
|
manager = celebrating()
|
||||||
|
manager._check_for_score(game("g1", home_score=score))
|
||||||
|
assert manager._score_baselines == {}
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("raw,expected", [
|
||||||
|
("7", 7), (7, 7), (7.0, 7), (" 7 ", 7), ("7 (SO)", 7),
|
||||||
|
({"value": 7}, 7), ({"displayValue": "7"}, 7),
|
||||||
|
])
|
||||||
|
def test_score_coercion(self, raw, expected):
|
||||||
|
assert CelebrationMixin._score_to_int(raw) == expected
|
||||||
|
|
||||||
|
def test_away_side_is_detected(self, celebrating):
|
||||||
|
manager = celebrating()
|
||||||
|
manager._check_for_score(game("g1", away_score=0))
|
||||||
|
manager._check_for_score(game("g1", away_score=1))
|
||||||
|
assert manager.active_celebration["scored_side"] == "away"
|
||||||
|
|
||||||
|
|
||||||
|
class TestWhoGetsCelebrated:
|
||||||
|
def test_no_favorites_celebrates_everyone(self, celebrating):
|
||||||
|
"""The user opted to show this game at all, so any score in it counts."""
|
||||||
|
manager = celebrating(favorites=[])
|
||||||
|
manager._check_for_score(game("g1", home="XXX", home_score=0))
|
||||||
|
manager._check_for_score(game("g1", home="XXX", home_score=1))
|
||||||
|
assert manager.active_celebration is not None
|
||||||
|
|
||||||
|
def test_favorite_scores(self, celebrating):
|
||||||
|
manager = celebrating(favorites=["FAV"])
|
||||||
|
manager._check_for_score(game("g1", home="FAV", home_score=0))
|
||||||
|
manager._check_for_score(game("g1", home="FAV", home_score=1))
|
||||||
|
assert manager.active_celebration is not None
|
||||||
|
|
||||||
|
def test_opponent_suppressed_by_default(self, celebrating):
|
||||||
|
manager = celebrating(favorites=["FAV"])
|
||||||
|
manager._check_for_score(game("g1", home="OPP", away="FAV", home_score=0))
|
||||||
|
manager._check_for_score(game("g1", home="OPP", away="FAV", home_score=1))
|
||||||
|
assert manager.active_celebration is None
|
||||||
|
|
||||||
|
def test_opponent_celebrated_when_opted_in(self, celebrating):
|
||||||
|
manager = celebrating(mode_config={"celebrate_opponent_scores": True},
|
||||||
|
favorites=["FAV"])
|
||||||
|
manager._check_for_score(game("g1", home="OPP", away="FAV", home_score=0))
|
||||||
|
manager._check_for_score(game("g1", home="OPP", away="FAV", home_score=1))
|
||||||
|
assert manager.active_celebration is not None
|
||||||
|
|
||||||
|
def test_matching_goes_through_the_favorite_key_seam(self, celebrating):
|
||||||
|
"""nrl matches on team id because its abbreviations are ambiguous
|
||||||
|
('NEW' is both Newcastle and New Zealand). Core must not care why."""
|
||||||
|
manager = celebrating(_ById, favorites=["g1-h"])
|
||||||
|
manager._check_for_score(game("g1", home="NEW", home_score=0))
|
||||||
|
manager._check_for_score(game("g1", home="NEW", home_score=1))
|
||||||
|
assert manager.active_celebration is not None
|
||||||
|
|
||||||
|
def test_favorite_key_seam_also_excludes(self, celebrating):
|
||||||
|
manager = celebrating(_ById, favorites=["someone-else"])
|
||||||
|
manager._check_for_score(game("g1", home="NEW", home_score=0))
|
||||||
|
manager._check_for_score(game("g1", home="NEW", home_score=1))
|
||||||
|
assert manager.active_celebration is None
|
||||||
|
|
||||||
|
|
||||||
|
class TestPhrasing:
|
||||||
|
def test_default_phrase_is_sport_neutral(self, celebrating):
|
||||||
|
manager = celebrating()
|
||||||
|
manager._check_for_score(game("g1", home="HOM", home_score=0))
|
||||||
|
manager._check_for_score(game("g1", home="HOM", home_score=1))
|
||||||
|
assert manager.active_celebration["phrase"] == "HOM SCORES!"
|
||||||
|
|
||||||
|
def test_score_phrase_hook_sees_the_points_delta(self, celebrating):
|
||||||
|
manager = celebrating(_Coalescing)
|
||||||
|
manager._check_for_score(game("g1", home_score=0))
|
||||||
|
manager._check_for_score(game("g1", home_score=6))
|
||||||
|
assert manager.active_celebration["phrase"] == "TOUCHDOWN!"
|
||||||
|
|
||||||
|
def test_score_phrase_hook_distinguishes_smaller_plays(self, celebrating):
|
||||||
|
manager = celebrating(_Coalescing)
|
||||||
|
manager._check_for_score(game("g1", home_score=0))
|
||||||
|
manager._check_for_score(game("g1", home_score=3))
|
||||||
|
assert manager.active_celebration["phrase"] == "HOM FIELD GOAL!"
|
||||||
|
|
||||||
|
def test_win_phrase(self, celebrating):
|
||||||
|
manager = celebrating(favorites=["HOM"])
|
||||||
|
manager._check_for_score(game("g1", home_score=1))
|
||||||
|
manager._check_for_win(game("g1", home_score=2, away_score=1))
|
||||||
|
assert manager.active_celebration["phrase"] == "HOM WINS!"
|
||||||
|
|
||||||
|
|
||||||
|
class TestCoalescing:
|
||||||
|
def test_off_by_default_two_goals_are_two_celebrations(self, celebrating):
|
||||||
|
"""Soccer/afl/nrl: consecutive increments are distinct events, so
|
||||||
|
suppressing the second would swallow a real goal."""
|
||||||
|
manager = celebrating()
|
||||||
|
manager._check_for_score(game("g1", home_score=0))
|
||||||
|
manager._check_for_score(game("g1", home_score=1))
|
||||||
|
first = manager.active_celebration["started_at"]
|
||||||
|
manager._check_for_score(game("g1", home_score=2))
|
||||||
|
assert manager.active_celebration["started_at"] != first
|
||||||
|
assert manager.active_celebration["home_score"] == 2
|
||||||
|
|
||||||
|
def test_on_suppresses_the_extra_point_follow_up(self, celebrating):
|
||||||
|
"""Football: a touchdown lands as +6, then +1 seconds later. One
|
||||||
|
takeover per scoring sequence."""
|
||||||
|
manager = celebrating(_Coalescing)
|
||||||
|
manager._check_for_score(game("g1", home_score=0))
|
||||||
|
manager._check_for_score(game("g1", home_score=6))
|
||||||
|
armed = manager.active_celebration
|
||||||
|
manager._check_for_score(game("g1", home_score=7))
|
||||||
|
assert manager.active_celebration is armed
|
||||||
|
|
||||||
|
def test_suppression_still_advances_the_baseline(self, celebrating):
|
||||||
|
"""Nothing may re-fire once the window closes."""
|
||||||
|
manager = celebrating(_Coalescing)
|
||||||
|
manager._check_for_score(game("g1", home_score=0))
|
||||||
|
manager._check_for_score(game("g1", home_score=6))
|
||||||
|
manager._check_for_score(game("g1", home_score=7))
|
||||||
|
assert manager._score_baselines["g1"]["home"] == 7
|
||||||
|
|
||||||
|
|
||||||
|
class TestWinDetection:
|
||||||
|
def test_win_requires_a_baseline(self, celebrating):
|
||||||
|
"""A game seen for the first time already-final (board started after
|
||||||
|
full time) must not fire."""
|
||||||
|
manager = celebrating(favorites=["HOM"])
|
||||||
|
manager._check_for_win(game("g1", home_score=3, away_score=1))
|
||||||
|
assert manager.active_celebration is None
|
||||||
|
|
||||||
|
def test_win_fires_once_only(self, celebrating):
|
||||||
|
manager = celebrating(favorites=["HOM"])
|
||||||
|
manager._check_for_score(game("g1", home_score=1))
|
||||||
|
manager._check_for_win(game("g1", home_score=3, away_score=1))
|
||||||
|
manager.active_celebration = None
|
||||||
|
manager._check_for_win(game("g1", home_score=3, away_score=1))
|
||||||
|
assert manager.active_celebration is None
|
||||||
|
|
||||||
|
def test_draw_does_not_celebrate(self, celebrating):
|
||||||
|
manager = celebrating(favorites=["HOM"])
|
||||||
|
manager._check_for_score(game("g1", home_score=1))
|
||||||
|
manager._check_for_win(game("g1", home_score=2, away_score=2))
|
||||||
|
assert manager.active_celebration is None
|
||||||
|
|
||||||
|
def test_win_is_gated_strictly_on_favorites(self, celebrating):
|
||||||
|
"""Unlike scores, a win with no favorites configured does NOT celebrate:
|
||||||
|
every game ends, so the fallback would be constant noise."""
|
||||||
|
manager = celebrating(favorites=[])
|
||||||
|
manager._check_for_score(game("g1", home_score=1))
|
||||||
|
manager._check_for_win(game("g1", home_score=3, away_score=1))
|
||||||
|
assert manager.active_celebration is None
|
||||||
|
|
||||||
|
def test_losing_favorite_does_not_celebrate(self, celebrating):
|
||||||
|
manager = celebrating(favorites=["HOM"])
|
||||||
|
manager._check_for_score(game("g1", home_score=1))
|
||||||
|
manager._check_for_win(game("g1", home_score=1, away_score=4))
|
||||||
|
assert manager.active_celebration is None
|
||||||
|
|
||||||
|
def test_away_favorite_wins(self, celebrating):
|
||||||
|
manager = celebrating(favorites=["AWY"])
|
||||||
|
manager._check_for_score(game("g1", away_score=1))
|
||||||
|
manager._check_for_win(game("g1", home_score=1, away_score=4))
|
||||||
|
assert manager.active_celebration["scored_side"] == "away"
|
||||||
|
|
||||||
|
|
||||||
|
class TestCelebrationSnapshot:
|
||||||
|
def test_the_game_is_snapshotted_not_referenced(self, celebrating):
|
||||||
|
"""A win must survive the game leaving live_games."""
|
||||||
|
manager = celebrating()
|
||||||
|
live = game("g1", home_score=0)
|
||||||
|
manager._check_for_score(live)
|
||||||
|
live = game("g1", home_score=1)
|
||||||
|
manager._check_for_score(live)
|
||||||
|
live["home_abbr"] = "MUTATED"
|
||||||
|
assert manager.active_celebration["game"]["home_abbr"] == "HOM"
|
||||||
|
|
||||||
|
def test_focus_is_pinned_to_the_involved_game(self, celebrating):
|
||||||
|
manager = celebrating()
|
||||||
|
manager._check_for_score(game("g1", home_score=0))
|
||||||
|
manager._check_for_score(game("g1", home_score=1))
|
||||||
|
assert manager.current_game["id"] == "g1"
|
||||||
|
|
||||||
|
|
||||||
|
class TestDisplayTakeover:
|
||||||
|
def test_no_celebration_defers_to_the_scorebug(self, celebrating):
|
||||||
|
manager = celebrating()
|
||||||
|
assert manager.display(force_clear=True) is True
|
||||||
|
assert manager.display_calls == [True]
|
||||||
|
|
||||||
|
def test_active_celebration_takes_over(self, celebrating):
|
||||||
|
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()
|
||||||
|
assert manager.display() is True
|
||||||
|
assert manager.display_calls == [], "the scorebug must not also render"
|
||||||
|
manager._draw_celebration_layout.assert_called_once()
|
||||||
|
|
||||||
|
def test_expired_celebration_clears_and_defers(self, celebrating):
|
||||||
|
# A short-but-valid duration: celebration_duration is clamped to a 1.0s
|
||||||
|
# floor, so a config of 0 does NOT expire on the next frame. Backdate
|
||||||
|
# started_at past the window to exercise the real expiry branch —
|
||||||
|
# otherwise the celebration is still active and this only passes
|
||||||
|
# because _draw_celebration_layout happens to raise in the harness
|
||||||
|
# (that path is covered by test_a_render_failure_falls_through).
|
||||||
|
manager = celebrating(mode_config={"celebration_duration": 1})
|
||||||
|
manager._check_for_score(game("g1", home_score=0))
|
||||||
|
manager._check_for_score(game("g1", home_score=1))
|
||||||
|
manager.active_celebration["started_at"] = time.time() - 2 # past the 1s window
|
||||||
|
# Mock the layout so a render can't raise: otherwise the exception
|
||||||
|
# branch clears the celebration too, and this test would pass whether
|
||||||
|
# or not expiry actually fired. An expired celebration must NOT render.
|
||||||
|
manager._draw_celebration_layout = MagicMock()
|
||||||
|
assert manager.display() is True
|
||||||
|
manager._draw_celebration_layout.assert_not_called()
|
||||||
|
assert manager.active_celebration is None
|
||||||
|
assert manager.display_calls == [False]
|
||||||
|
|
||||||
|
def test_expiry_resets_the_dwell_clock(self, celebrating):
|
||||||
|
"""So the scorebug resumes on the scoring game for a full duration
|
||||||
|
before rotation can move on."""
|
||||||
|
manager = celebrating(mode_config={"celebration_duration": 1})
|
||||||
|
manager._check_for_score(game("g1", home_score=0))
|
||||||
|
manager._check_for_score(game("g1", home_score=1))
|
||||||
|
manager.active_celebration["started_at"] = time.time() - 2 # past the 1s window
|
||||||
|
manager._draw_celebration_layout = MagicMock() # expiry, not a render failure
|
||||||
|
manager.display()
|
||||||
|
manager._draw_celebration_layout.assert_not_called()
|
||||||
|
assert manager.last_game_switch > 0
|
||||||
|
|
||||||
|
def test_a_render_failure_falls_through_to_the_scorebug(self, celebrating):
|
||||||
|
"""A broken celebration must never blank the display."""
|
||||||
|
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"))
|
||||||
|
assert manager.display() is True
|
||||||
|
assert manager.display_calls == [False]
|
||||||
|
|
||||||
|
def test_a_render_failure_disarms_rather_than_retrying(self, celebrating):
|
||||||
|
"""Left armed, the same render fails on every frame for the rest of the
|
||||||
|
window — a traceback per frame, and no scorebug."""
|
||||||
|
manager = celebrating()
|
||||||
|
manager._check_for_score(game("g1", home_score=0))
|
||||||
|
manager._check_for_score(game("g1", home_score=1))
|
||||||
|
manager._draw_celebration_layout = MagicMock(side_effect=RuntimeError("boom"))
|
||||||
|
manager.display()
|
||||||
|
assert manager.active_celebration is None
|
||||||
|
manager.display()
|
||||||
|
assert manager._draw_celebration_layout.call_count == 1
|
||||||
|
|
||||||
|
|
||||||
|
class TestBaselinePruning:
|
||||||
|
"""`_score_baselines` gains an entry per game and only _check_for_win ever
|
||||||
|
removed one, so a board running all season grows the dict without bound."""
|
||||||
|
|
||||||
|
def test_prunes_games_no_longer_live(self, celebrating):
|
||||||
|
manager = celebrating()
|
||||||
|
for gid in ("g1", "g2", "g3"):
|
||||||
|
manager._check_for_score(game(gid, home_score=1))
|
||||||
|
manager.prune_score_baselines([game("g2")])
|
||||||
|
assert set(manager._score_baselines) == {"g2"}
|
||||||
|
|
||||||
|
def test_keeps_every_still_live_game(self, celebrating):
|
||||||
|
manager = celebrating()
|
||||||
|
for gid in ("g1", "g2"):
|
||||||
|
manager._check_for_score(game(gid, home_score=1))
|
||||||
|
manager.prune_score_baselines([game("g1"), game("g2")])
|
||||||
|
assert set(manager._score_baselines) == {"g1", "g2"}
|
||||||
|
|
||||||
|
def test_empty_live_set_clears_everything(self, celebrating):
|
||||||
|
manager = celebrating()
|
||||||
|
manager._check_for_score(game("g1", home_score=1))
|
||||||
|
manager.prune_score_baselines([])
|
||||||
|
assert manager._score_baselines == {}
|
||||||
|
|
||||||
|
def test_pruning_does_not_disturb_a_surviving_baseline(self, celebrating):
|
||||||
|
manager = celebrating()
|
||||||
|
manager._check_for_score(game("g1", home_score=2))
|
||||||
|
manager.prune_score_baselines([game("g1")])
|
||||||
|
manager._check_for_score(game("g1", home_score=3))
|
||||||
|
assert manager.active_celebration is not None, (
|
||||||
|
"pruning must not drop a live game's baseline and re-trigger the "
|
||||||
|
"first-sighting suppression")
|
||||||
|
|
||||||
|
def test_disabled_manager_renders_nothing(self, celebrating):
|
||||||
|
manager = celebrating()
|
||||||
|
manager.is_enabled = False
|
||||||
|
assert manager.display() is False
|
||||||
|
|
||||||
|
|
||||||
|
class TestFitFont:
|
||||||
|
def test_returns_the_first_font_that_fits(self, celebrating):
|
||||||
|
manager = celebrating()
|
||||||
|
draw = MagicMock()
|
||||||
|
draw.textlength.side_effect = [100, 20]
|
||||||
|
big, small = MagicMock(), MagicMock()
|
||||||
|
assert manager._fit_font(draw, "GOAL", 50, [big, small]) is small
|
||||||
|
|
||||||
|
def test_falls_back_to_the_smallest(self, celebrating):
|
||||||
|
manager = celebrating()
|
||||||
|
draw = MagicMock()
|
||||||
|
draw.textlength.return_value = 999
|
||||||
|
big, small = MagicMock(), MagicMock()
|
||||||
|
assert manager._fit_font(draw, "GOAL", 50, [big, small]) is small
|
||||||
|
|
||||||
|
|
||||||
|
class TestCapabilityExports:
|
||||||
|
@pytest.mark.parametrize("name", [
|
||||||
|
"CelebrationMixin", "RotationStrategy", "SimpleRotation",
|
||||||
|
"SmoothWeightedRotation", "WeightedCycleRotation",
|
||||||
|
"get_rotation_strategy", "register_rotation_strategy",
|
||||||
|
])
|
||||||
|
def test_public_name_is_importable(self, name):
|
||||||
|
"""Plugins import these behind a guarded fallback; the names are the
|
||||||
|
contract."""
|
||||||
|
import src.base_classes.sports.capabilities as capabilities
|
||||||
|
assert hasattr(capabilities, name)
|
||||||
|
|
||||||
|
def test_rotation_strategy_base_requires_a_schedule(self):
|
||||||
|
with pytest.raises(NotImplementedError):
|
||||||
|
RotationStrategy().schedule([game("a")])
|
||||||
@@ -0,0 +1,613 @@
|
|||||||
|
"""Tests for the methods promoted onto SportsCore from the nine bundled
|
||||||
|
plugin copies of ``sports.py`` (phase B1 of docs/SPORTS_UNIFICATION.md).
|
||||||
|
|
||||||
|
Three methods and their seams land here:
|
||||||
|
|
||||||
|
- ``cleanup()`` — byte-identical in all nine copies. The tests pin the
|
||||||
|
ordering (session close, then caches, then the completion log) and the
|
||||||
|
deliberate *omission*: the process-wide background service must never be
|
||||||
|
shut down by one unloading plugin.
|
||||||
|
- ``_get_layout_offset()`` — football's resolver-backed variant, with the
|
||||||
|
classic inline config read as the fallback used by every plugin that
|
||||||
|
doesn't hand core a ``_config_schema_path()``.
|
||||||
|
- ``_load_custom_font_from_element_config()`` — baseball's body (the only
|
||||||
|
copy that handles BDF strikes correctly) under hockey's wider signature,
|
||||||
|
resolving font files through the ``_font_root()`` seam instead of the
|
||||||
|
process cwd.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import ast
|
||||||
|
import json
|
||||||
|
import logging
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
from pathlib import Path
|
||||||
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from PIL import Image, ImageFont
|
||||||
|
|
||||||
|
# src.base_classes.sports transitively imports the hardware matrix driver;
|
||||||
|
# stub it so these tests can import the sports base classes off-device.
|
||||||
|
sys.modules.setdefault("rgbmatrix", MagicMock())
|
||||||
|
|
||||||
|
from src.base_classes.sports import SportsCore
|
||||||
|
|
||||||
|
LOGGER = logging.getLogger("test_sports_core_promotions")
|
||||||
|
|
||||||
|
CORE_ROOT = Path(__file__).resolve().parents[1]
|
||||||
|
FONTS_DIR = CORE_ROOT / "assets" / "fonts"
|
||||||
|
TTF_NAME = "PressStart2P-Regular.ttf"
|
||||||
|
BDF_NAME = "5x7.bdf" # a BDF whose only valid strike is 7px
|
||||||
|
BDF_NATIVE_SIZE = 7
|
||||||
|
|
||||||
|
|
||||||
|
class _StubSports(SportsCore):
|
||||||
|
"""Minimal concrete SportsCore — the abstract methods are never called
|
||||||
|
by anything under test here."""
|
||||||
|
|
||||||
|
def _fetch_data(self):
|
||||||
|
return None
|
||||||
|
|
||||||
|
def _extract_game_details(self, game_event):
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def build(monkeypatch, tmp_path):
|
||||||
|
"""Factory for real SportsCore instances: logo dir redirected to tmp and
|
||||||
|
the process-wide background service replaced with a MagicMock so the
|
||||||
|
tests can assert nothing ever calls it."""
|
||||||
|
monkeypatch.setattr(
|
||||||
|
SportsCore, "_initialize_logo_dir", lambda self, configured: tmp_path)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"src.base_classes.sports.core.get_background_service",
|
||||||
|
lambda *args, **kwargs: MagicMock())
|
||||||
|
|
||||||
|
def _build(config=None, cls=_StubSports):
|
||||||
|
display_manager = MagicMock()
|
||||||
|
display_manager.matrix.width = 128
|
||||||
|
display_manager.matrix.height = 32
|
||||||
|
display_manager.width = 128
|
||||||
|
display_manager.height = 32
|
||||||
|
display_manager.image = Image.new("RGB", (128, 32))
|
||||||
|
cache_manager = MagicMock()
|
||||||
|
cache_manager.cache_dir = str(tmp_path)
|
||||||
|
return cls(config if config is not None else {"timezone": "UTC"},
|
||||||
|
display_manager, cache_manager, LOGGER, "nhl")
|
||||||
|
|
||||||
|
return _build
|
||||||
|
|
||||||
|
|
||||||
|
def probe(config=None):
|
||||||
|
"""Unbound-call stand-in for hosts we don't need a full instance for
|
||||||
|
(same pattern as make_probe in test_sports_base_characterization)."""
|
||||||
|
host = MagicMock()
|
||||||
|
host.logger = LOGGER
|
||||||
|
host.config = config if config is not None else {}
|
||||||
|
host._font_cache = {}
|
||||||
|
host._bdf_native_size_cache = {}
|
||||||
|
host._config_schema_path.return_value = None
|
||||||
|
host._font_root.side_effect = lambda: SportsCore._font_root(host)
|
||||||
|
host._resolve_font_path.side_effect = (
|
||||||
|
lambda name: SportsCore._resolve_font_path(host, name))
|
||||||
|
return host
|
||||||
|
|
||||||
|
|
||||||
|
def offset(host, element, axis, default=0):
|
||||||
|
return SportsCore._get_layout_offset(host, element, axis, default)
|
||||||
|
|
||||||
|
|
||||||
|
def load_font(host, *args, **kwargs):
|
||||||
|
return SportsCore._load_custom_font_from_element_config(host, *args, **kwargs)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 1. cleanup()
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestCleanup:
|
||||||
|
def test_closes_session_and_clears_all_caches(self, build):
|
||||||
|
manager = build()
|
||||||
|
session = MagicMock()
|
||||||
|
manager.session = session
|
||||||
|
manager._logo_cache["TB"] = object()
|
||||||
|
manager._font_cache[("PressStart2P-Regular.ttf", 8)] = object()
|
||||||
|
manager._bdf_native_size_cache["assets/fonts/5x7.bdf"] = 7
|
||||||
|
|
||||||
|
manager.cleanup()
|
||||||
|
|
||||||
|
session.close.assert_called_once_with()
|
||||||
|
assert manager._logo_cache == {}
|
||||||
|
# Promoted alongside the font loader: these hold PIL faces and are
|
||||||
|
# an unbounded leak across enable/disable cycles if never released.
|
||||||
|
assert manager._font_cache == {}
|
||||||
|
assert manager._bdf_native_size_cache == {}
|
||||||
|
|
||||||
|
def test_second_cleanup_is_a_noop(self, build):
|
||||||
|
manager = build()
|
||||||
|
manager.session = MagicMock()
|
||||||
|
manager._logo_cache["TB"] = object()
|
||||||
|
manager._font_cache[("x", 8)] = object()
|
||||||
|
|
||||||
|
manager.cleanup()
|
||||||
|
manager.cleanup() # must not raise on already-released state
|
||||||
|
|
||||||
|
assert manager._logo_cache == {}
|
||||||
|
assert manager._font_cache == {}
|
||||||
|
assert manager._bdf_native_size_cache == {}
|
||||||
|
assert manager.session.close.call_count == 2
|
||||||
|
|
||||||
|
def test_does_not_shut_down_the_shared_background_service(self, build):
|
||||||
|
# get_background_service() hands out a PROCESS-WIDE singleton shared
|
||||||
|
# by every scoreboard. One plugin unloading must not stop background
|
||||||
|
# fetching for the other eight — cleanup() touches it not at all.
|
||||||
|
manager = build()
|
||||||
|
service = manager.background_service
|
||||||
|
manager.session = MagicMock()
|
||||||
|
|
||||||
|
manager.cleanup()
|
||||||
|
|
||||||
|
assert service.shutdown.called is False
|
||||||
|
assert service.stop.called is False
|
||||||
|
assert service.method_calls == [], (
|
||||||
|
"cleanup() called into the shared background service: "
|
||||||
|
f"{service.method_calls}")
|
||||||
|
|
||||||
|
def test_completion_is_logged_even_when_session_close_raises(self, build, caplog):
|
||||||
|
manager = build()
|
||||||
|
manager.session = MagicMock()
|
||||||
|
manager.session.close.side_effect = RuntimeError("socket already gone")
|
||||||
|
manager._logo_cache["TB"] = object()
|
||||||
|
|
||||||
|
with caplog.at_level(logging.DEBUG, logger=LOGGER.name):
|
||||||
|
manager.cleanup()
|
||||||
|
|
||||||
|
messages = [r.message for r in caplog.records]
|
||||||
|
assert any("Error closing session" in m for m in messages)
|
||||||
|
# Ordering is load-bearing: the caches still get cleared and the
|
||||||
|
# completion log still fires after a failed close.
|
||||||
|
assert manager._logo_cache == {}
|
||||||
|
assert any("cleanup completed" in m for m in messages)
|
||||||
|
|
||||||
|
def test_tolerates_missing_attributes(self):
|
||||||
|
# The hasattr guards exist so a partially constructed instance (an
|
||||||
|
# __init__ that raised) can still be cleaned up.
|
||||||
|
host = MagicMock(spec=["logger"])
|
||||||
|
host.logger = LOGGER
|
||||||
|
SportsCore.cleanup(host)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 2. _get_layout_offset() + the _config_schema_path() seam
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def layout_config(element, axis, value):
|
||||||
|
return {"customization": {"layout": {element: {axis: value}}}}
|
||||||
|
|
||||||
|
|
||||||
|
class TestLayoutOffsetClassicPath:
|
||||||
|
"""The default path: _config_schema_path() returns None, so offsets come
|
||||||
|
from the inline customization.layout read every plugin ships today."""
|
||||||
|
|
||||||
|
def test_config_schema_path_defaults_to_none(self, build):
|
||||||
|
manager = build()
|
||||||
|
assert manager._config_schema_path() is None
|
||||||
|
|
||||||
|
def test_reads_configured_int(self):
|
||||||
|
host = probe(layout_config("home_logo", "x_offset", 5))
|
||||||
|
assert offset(host, "home_logo", "x_offset") == 5
|
||||||
|
|
||||||
|
def test_float_is_truncated_to_int(self):
|
||||||
|
host = probe(layout_config("score", "y_offset", 2.9))
|
||||||
|
result = offset(host, "score", "y_offset")
|
||||||
|
assert result == 2 and isinstance(result, int)
|
||||||
|
|
||||||
|
def test_numeric_string_is_coerced(self):
|
||||||
|
host = probe(layout_config("score", "x_offset", "-3.5"))
|
||||||
|
assert offset(host, "score", "x_offset") == -3
|
||||||
|
|
||||||
|
def test_unconfigured_element_and_axis_use_default(self):
|
||||||
|
host = probe(layout_config("score", "x_offset", 5))
|
||||||
|
assert offset(host, "status_text", "x_offset", 7) == 7
|
||||||
|
assert offset(host, "score", "y_offset", -1) == -1
|
||||||
|
assert offset(probe(), "score", "x_offset", 4) == 4
|
||||||
|
|
||||||
|
def test_non_numeric_string_degrades_to_default(self):
|
||||||
|
host = probe(layout_config("score", "x_offset", "left"))
|
||||||
|
assert offset(host, "score", "x_offset", 3) == 3
|
||||||
|
|
||||||
|
def test_unsupported_type_degrades_to_default(self):
|
||||||
|
host = probe(layout_config("score", "x_offset", {"nested": 1}))
|
||||||
|
assert offset(host, "score", "x_offset", 2) == 2
|
||||||
|
host = probe(layout_config("score", "x_offset", None))
|
||||||
|
assert offset(host, "score", "x_offset", 2) == 2
|
||||||
|
|
||||||
|
def test_broken_config_object_degrades_to_default(self):
|
||||||
|
host = probe()
|
||||||
|
host.config = "not a dict"
|
||||||
|
assert offset(host, "score", "x_offset", 6) == 6
|
||||||
|
|
||||||
|
def test_boolean_counts_as_one(self):
|
||||||
|
# PINNED AS-IS: the classic read predates the resolver and treats a
|
||||||
|
# bool as its int value (True -> 1). See the resolver test below for
|
||||||
|
# the stricter, more correct handling.
|
||||||
|
host = probe(layout_config("score", "x_offset", True))
|
||||||
|
assert offset(host, "score", "x_offset", 4) == 1
|
||||||
|
|
||||||
|
|
||||||
|
class TestLayoutOffsetResolverPath:
|
||||||
|
"""When a plugin supplies its config_schema.json, offsets resolve through
|
||||||
|
src.element_style instead."""
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def schema_path(self, tmp_path):
|
||||||
|
path = tmp_path / "config_schema.json"
|
||||||
|
path.write_text(json.dumps({
|
||||||
|
"type": "object",
|
||||||
|
"properties": {
|
||||||
|
"customization": {
|
||||||
|
"type": "object",
|
||||||
|
"properties": {
|
||||||
|
"layout": {"type": "object", "properties": {}},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}))
|
||||||
|
return str(path)
|
||||||
|
|
||||||
|
def host(self, schema_path, config):
|
||||||
|
host = probe(config)
|
||||||
|
host._config_schema_path.return_value = schema_path
|
||||||
|
del host._style_resolver_cached # MagicMock auto-attrs otherwise
|
||||||
|
host._style_resolver_cached = None
|
||||||
|
return host
|
||||||
|
|
||||||
|
def test_reads_configured_offsets(self, schema_path):
|
||||||
|
host = self.host(schema_path, layout_config("home_logo", "x_offset", 5))
|
||||||
|
assert offset(host, "home_logo", "x_offset") == 5
|
||||||
|
|
||||||
|
def test_numeric_string_is_coerced(self, schema_path):
|
||||||
|
host = self.host(schema_path, layout_config("score", "x_offset", "-3.5"))
|
||||||
|
assert offset(host, "score", "x_offset") == -3
|
||||||
|
|
||||||
|
def test_missing_value_uses_default(self, schema_path):
|
||||||
|
host = self.host(schema_path, layout_config("score", "x_offset", 5))
|
||||||
|
assert offset(host, "score", "y_offset", 9) == 9
|
||||||
|
|
||||||
|
def test_bad_input_degrades_to_default(self, schema_path):
|
||||||
|
host = self.host(schema_path, layout_config("score", "x_offset", "left"))
|
||||||
|
assert offset(host, "score", "x_offset", 3) == 3
|
||||||
|
host = self.host(schema_path, {"customization": {"layout": "nope"}})
|
||||||
|
assert offset(host, "score", "x_offset", 3) == 3
|
||||||
|
|
||||||
|
def test_boolean_is_rejected_unlike_the_classic_path(self, schema_path):
|
||||||
|
# The intended behavior difference: a bool is not a pixel offset, so
|
||||||
|
# the resolver returns the default where the classic read returns 1.
|
||||||
|
host = self.host(schema_path, layout_config("score", "x_offset", True))
|
||||||
|
assert offset(host, "score", "x_offset", 4) == 4
|
||||||
|
|
||||||
|
def test_resolver_is_cached_and_rebuilt_when_config_is_swapped(self, schema_path):
|
||||||
|
host = self.host(schema_path, layout_config("score", "x_offset", 5))
|
||||||
|
assert offset(host, "score", "x_offset") == 5
|
||||||
|
first = host._style_resolver_cached
|
||||||
|
assert offset(host, "score", "x_offset") == 5
|
||||||
|
assert host._style_resolver_cached is first
|
||||||
|
|
||||||
|
# on_config_change swaps the dict object; the resolver must follow.
|
||||||
|
host.config = layout_config("score", "x_offset", 11)
|
||||||
|
assert offset(host, "score", "x_offset") == 11
|
||||||
|
assert host._style_resolver_cached is not first
|
||||||
|
|
||||||
|
def test_missing_schema_file_still_resolves_offsets(self, tmp_path):
|
||||||
|
# Offsets don't depend on schema defaults, so an unreadable schema
|
||||||
|
# must not cost the plugin its layout customization.
|
||||||
|
host = self.host(str(tmp_path / "absent.json"),
|
||||||
|
layout_config("score", "x_offset", 5))
|
||||||
|
assert offset(host, "score", "x_offset") == 5
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 3. _load_custom_font_from_element_config() + the _font_root() seam
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestFontRootSeam:
|
||||||
|
def test_default_font_root_is_the_core_install_root(self, build):
|
||||||
|
manager = build()
|
||||||
|
assert Path(manager._font_root()) == CORE_ROOT
|
||||||
|
assert (Path(manager._font_root()) / "assets" / "fonts").is_dir()
|
||||||
|
|
||||||
|
def test_resolve_font_path_honors_an_overridden_root(self, tmp_path):
|
||||||
|
fonts = tmp_path / "assets" / "fonts"
|
||||||
|
fonts.mkdir(parents=True)
|
||||||
|
(fonts / "Bundled.ttf").write_bytes(b"not really a font")
|
||||||
|
host = probe()
|
||||||
|
host._font_root.side_effect = lambda: str(tmp_path)
|
||||||
|
assert SportsCore._resolve_font_path(host, "Bundled.ttf") == str(
|
||||||
|
fonts / "Bundled.ttf")
|
||||||
|
|
||||||
|
def test_unknown_font_returns_the_familiar_relative_path(self):
|
||||||
|
host = probe()
|
||||||
|
assert SportsCore._resolve_font_path(host, "Nope.ttf") == os.path.join(
|
||||||
|
"assets", "fonts", "Nope.ttf")
|
||||||
|
|
||||||
|
|
||||||
|
class TestFontLoaderSignature:
|
||||||
|
"""Hockey's signature is the only safe superset: basketball's positional
|
||||||
|
``default_font: str`` blows up on an explicit None."""
|
||||||
|
|
||||||
|
def test_two_arg_call(self):
|
||||||
|
font = load_font(probe(), {"font": TTF_NAME, "font_size": 10})
|
||||||
|
assert isinstance(font, ImageFont.FreeTypeFont)
|
||||||
|
assert font.size == 10
|
||||||
|
|
||||||
|
def test_default_size_is_used_when_config_omits_it(self):
|
||||||
|
assert load_font(probe(), {}, 12).size == 12
|
||||||
|
|
||||||
|
def test_three_positional_args(self):
|
||||||
|
font = load_font(probe(), {}, 6, "4x6-font.ttf")
|
||||||
|
assert isinstance(font, ImageFont.FreeTypeFont)
|
||||||
|
assert font.size == 6
|
||||||
|
assert font.path.endswith("4x6-font.ttf")
|
||||||
|
|
||||||
|
def test_explicit_default_font_none(self):
|
||||||
|
# The regression this signature guards: os.path.join(..., None).
|
||||||
|
font = load_font(probe(), {"font_size": 9}, default_font=None)
|
||||||
|
assert isinstance(font, ImageFont.FreeTypeFont)
|
||||||
|
assert font.path.endswith(TTF_NAME)
|
||||||
|
|
||||||
|
def test_config_font_wins_over_default_font(self):
|
||||||
|
font = load_font(probe(), {"font": TTF_NAME}, 8, "4x6-font.ttf")
|
||||||
|
assert font.path.endswith(TTF_NAME)
|
||||||
|
|
||||||
|
def test_string_font_size_is_coerced(self):
|
||||||
|
assert load_font(probe(), {"font": TTF_NAME, "font_size": "11"}).size == 11
|
||||||
|
|
||||||
|
|
||||||
|
class TestFontLoaderBehavior:
|
||||||
|
def test_family_alias_resolves_through_the_font_manager_catalog(self):
|
||||||
|
# "press_start" is a FontManager catalog family, not a filename; the
|
||||||
|
# promoted loader must not carry its own duplicate alias table.
|
||||||
|
host = probe()
|
||||||
|
font = load_font(host, {"font": "press_start", "font_size": 8})
|
||||||
|
assert font.path.endswith(TTF_NAME)
|
||||||
|
assert ("PressStart2P-Regular.ttf", 8) in host._font_cache
|
||||||
|
|
||||||
|
def test_memo_cache_returns_the_same_face(self):
|
||||||
|
host = probe()
|
||||||
|
first = load_font(host, {"font": TTF_NAME, "font_size": 8})
|
||||||
|
second = load_font(host, {"font": TTF_NAME, "font_size": 8})
|
||||||
|
assert first is second
|
||||||
|
assert len(host._font_cache) == 1
|
||||||
|
# A different size is a different face.
|
||||||
|
assert load_font(host, {"font": TTF_NAME, "font_size": 9}) is not first
|
||||||
|
assert len(host._font_cache) == 2
|
||||||
|
|
||||||
|
def test_bdf_loads_at_its_native_strike_when_the_request_misses(self):
|
||||||
|
# BDF is a fixed-size bitmap format: FreeType raises "invalid pixel
|
||||||
|
# size" for anything but the file's own strike. Baseball's retry is
|
||||||
|
# the only copy that gets this right.
|
||||||
|
host = probe()
|
||||||
|
font = load_font(host, {"font": BDF_NAME, "font_size": 8})
|
||||||
|
assert isinstance(font, ImageFont.FreeTypeFont)
|
||||||
|
assert font.size == BDF_NATIVE_SIZE
|
||||||
|
assert font.path.endswith(BDF_NAME)
|
||||||
|
assert set(host._bdf_native_size_cache.values()) == {BDF_NATIVE_SIZE}
|
||||||
|
# The retried face is memoized under the REQUESTED size.
|
||||||
|
assert host._font_cache[(BDF_NAME, 8)] is font
|
||||||
|
|
||||||
|
def test_bdf_at_its_native_size_needs_no_retry(self):
|
||||||
|
host = probe()
|
||||||
|
font = load_font(host, {"font": BDF_NAME, "font_size": BDF_NATIVE_SIZE})
|
||||||
|
assert font.size == BDF_NATIVE_SIZE
|
||||||
|
assert host._bdf_native_size_cache == {}
|
||||||
|
|
||||||
|
def test_bdf_strike_lookup_is_memoized(self, monkeypatch):
|
||||||
|
calls = []
|
||||||
|
real = SportsCore.__module__
|
||||||
|
|
||||||
|
def counting(path):
|
||||||
|
calls.append(path)
|
||||||
|
from src.font_manager import FontManager
|
||||||
|
return FontManager._read_bdf_native_size(path)
|
||||||
|
|
||||||
|
monkeypatch.setattr(f"{real}._read_bdf_native_size", counting)
|
||||||
|
host = probe()
|
||||||
|
load_font(host, {"font": BDF_NAME, "font_size": 8})
|
||||||
|
host._font_cache.clear() # force the load path again
|
||||||
|
load_font(host, {"font": BDF_NAME, "font_size": 8})
|
||||||
|
assert len(calls) == 1
|
||||||
|
|
||||||
|
def test_missing_font_falls_back_and_caches_the_fallback(self, caplog):
|
||||||
|
host = probe()
|
||||||
|
with caplog.at_level(logging.WARNING, logger=LOGGER.name):
|
||||||
|
font = load_font(host, {"font": "DoesNotExist.ttf", "font_size": 8})
|
||||||
|
assert isinstance(font, ImageFont.FreeTypeFont)
|
||||||
|
assert font.path.endswith(TTF_NAME)
|
||||||
|
assert any("Font file not found" in r.message for r in caplog.records)
|
||||||
|
# Cached under the requested name so a misconfiguration costs one
|
||||||
|
# disk probe, not one per frame.
|
||||||
|
assert host._font_cache[("DoesNotExist.ttf", 8)] is font
|
||||||
|
|
||||||
|
def test_unknown_extension_falls_back(self):
|
||||||
|
host = probe()
|
||||||
|
font = load_font(host, {"font": "AUTHORS", "font_size": 8})
|
||||||
|
assert font.path.endswith(TTF_NAME)
|
||||||
|
|
||||||
|
def test_fallback_honors_the_supplied_default_font(self):
|
||||||
|
host = probe()
|
||||||
|
font = load_font(host, {"font": "DoesNotExist.ttf"}, 6, "4x6-font.ttf")
|
||||||
|
assert font.path.endswith("4x6-font.ttf")
|
||||||
|
|
||||||
|
|
||||||
|
class TestFontLoaderCwdIndependence:
|
||||||
|
"""The bug the _font_root() seam exists to prevent: every plugin copy
|
||||||
|
joins 'assets/fonts' onto the process cwd, so a process started anywhere
|
||||||
|
else silently degrades to PIL's default bitmap face (the same defect
|
||||||
|
already fixed in FontManager — see CHANGELOG Unreleased/Fixed)."""
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("font_name,expected_size",
|
||||||
|
[(TTF_NAME, 8), (BDF_NAME, BDF_NATIVE_SIZE)])
|
||||||
|
def test_fonts_load_from_an_unrelated_cwd(self, monkeypatch, font_name,
|
||||||
|
expected_size):
|
||||||
|
monkeypatch.chdir("/")
|
||||||
|
host = probe()
|
||||||
|
font = load_font(host, {"font": font_name, "font_size": 8})
|
||||||
|
assert isinstance(font, ImageFont.FreeTypeFont), (
|
||||||
|
f"{font_name} degraded to PIL's default face when the process "
|
||||||
|
"runs outside the install root")
|
||||||
|
assert font.size == expected_size
|
||||||
|
assert Path(font.path) == FONTS_DIR / font_name
|
||||||
|
|
||||||
|
def test_fallback_font_also_survives_an_unrelated_cwd(self, monkeypatch):
|
||||||
|
monkeypatch.chdir("/")
|
||||||
|
font = load_font(probe(), {"font": "DoesNotExist.ttf", "font_size": 8})
|
||||||
|
assert isinstance(font, ImageFont.FreeTypeFont)
|
||||||
|
assert Path(font.path) == FONTS_DIR / TTF_NAME
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("key", ["score", "time", "team", "status",
|
||||||
|
"detail", "rank"])
|
||||||
|
def test_load_fonts_survives_an_unrelated_cwd(self, monkeypatch, key):
|
||||||
|
"""`_load_fonts` had the same cwd-relative literals the seam exists to
|
||||||
|
remove; every scoreboard font silently became PIL's default bitmap face
|
||||||
|
when the process started outside the install root."""
|
||||||
|
monkeypatch.chdir("/")
|
||||||
|
fonts = SportsCore._load_fonts(probe())
|
||||||
|
assert isinstance(fonts[key], ImageFont.FreeTypeFont), (
|
||||||
|
f"fonts['{key}'] degraded to PIL's default face outside the "
|
||||||
|
"install root")
|
||||||
|
|
||||||
|
|
||||||
|
class TestShouldLogCooldown:
|
||||||
|
"""`_should_log` reads `self._last_warning_time` unguarded, so it must be
|
||||||
|
initialized in __init__ — otherwise the first warning of a run raises
|
||||||
|
AttributeError instead of logging."""
|
||||||
|
|
||||||
|
def test_cooldown_clock_is_initialized(self, build):
|
||||||
|
assert build()._last_warning_time == 0
|
||||||
|
|
||||||
|
def test_first_call_logs_then_cools_down(self, build):
|
||||||
|
manager = build()
|
||||||
|
assert manager._should_log("api", cooldown=60) is True
|
||||||
|
assert manager._should_log("api", cooldown=60) is False
|
||||||
|
|
||||||
|
def test_cooldown_expires(self, build):
|
||||||
|
manager = build()
|
||||||
|
assert manager._should_log("api", cooldown=60) is True
|
||||||
|
manager._warning_cooldowns["api"] -= 61
|
||||||
|
assert manager._should_log("api", cooldown=60) is True
|
||||||
|
|
||||||
|
def test_cooldowns_are_tracked_per_warning_type(self, build):
|
||||||
|
"""The parameter was accepted and ignored: one shared timestamp meant an
|
||||||
|
API warning silenced an unrelated cache warning for the next minute."""
|
||||||
|
manager = build()
|
||||||
|
assert manager._should_log("api", cooldown=60) is True
|
||||||
|
assert manager._should_log("cache", cooldown=60) is True
|
||||||
|
assert manager._should_log("api", cooldown=60) is False
|
||||||
|
assert manager._should_log("cache", cooldown=60) is False
|
||||||
|
|
||||||
|
def test_one_type_expiring_does_not_free_another(self, build):
|
||||||
|
manager = build()
|
||||||
|
manager._should_log("api")
|
||||||
|
manager._should_log("cache")
|
||||||
|
manager._warning_cooldowns["api"] -= 61
|
||||||
|
assert manager._should_log("api") is True
|
||||||
|
assert manager._should_log("cache") is False
|
||||||
|
|
||||||
|
def test_legacy_single_clock_field_is_kept_in_step(self, build):
|
||||||
|
"""Subclasses in the plugin copies read _last_warning_time directly."""
|
||||||
|
manager = build()
|
||||||
|
manager._should_log("api")
|
||||||
|
assert manager._last_warning_time == manager._warning_cooldowns["api"]
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 4. Seam guard rails
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestPromotedSeamsExist:
|
||||||
|
@pytest.mark.parametrize("name", [
|
||||||
|
"cleanup", "_get_layout_offset", "_load_custom_font_from_element_config",
|
||||||
|
"_config_schema_path", "_font_root", "_resolve_font_path",
|
||||||
|
])
|
||||||
|
def test_method_is_callable_on_the_base_class(self, name):
|
||||||
|
assert callable(getattr(SportsCore, name, None)), (
|
||||||
|
f"SportsCore.{name} is part of the promoted plugin-facing seam "
|
||||||
|
"(docs/SPORTS_UNIFICATION.md) — plugins probe for it with "
|
||||||
|
"hasattr before delegating.")
|
||||||
|
|
||||||
|
def test_no_sport_names_leaked_into_core(self):
|
||||||
|
"""core.py must never branch on which sport it is (prose and skin-id
|
||||||
|
examples in docstrings are fine — executable code is not)."""
|
||||||
|
tree = ast.parse((CORE_ROOT / "src" / "base_classes" / "sports"
|
||||||
|
/ "core.py").read_text())
|
||||||
|
docstrings = set()
|
||||||
|
for node in ast.walk(tree):
|
||||||
|
if isinstance(node, (ast.Module, ast.ClassDef, ast.FunctionDef,
|
||||||
|
ast.AsyncFunctionDef)):
|
||||||
|
first = node.body[0] if node.body else None
|
||||||
|
if (isinstance(first, ast.Expr)
|
||||||
|
and isinstance(first.value, ast.Constant)
|
||||||
|
and isinstance(first.value.value, str)):
|
||||||
|
docstrings.add(id(first.value))
|
||||||
|
|
||||||
|
tokens = []
|
||||||
|
for node in ast.walk(tree):
|
||||||
|
if isinstance(node, ast.Constant) and isinstance(node.value, str):
|
||||||
|
if id(node) not in docstrings:
|
||||||
|
tokens.append(node.value)
|
||||||
|
elif isinstance(node, ast.Name):
|
||||||
|
tokens.append(node.id)
|
||||||
|
elif isinstance(node, ast.Attribute):
|
||||||
|
tokens.append(node.attr)
|
||||||
|
elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef,
|
||||||
|
ast.ClassDef)):
|
||||||
|
tokens.append(node.name)
|
||||||
|
|
||||||
|
haystack = " ".join(tokens).lower()
|
||||||
|
for sport in ("afl", "nrl", "hockey", "baseball", "basketball",
|
||||||
|
"football", "lacrosse", "soccer", "ufc"):
|
||||||
|
assert sport not in haystack, (
|
||||||
|
f"core.py code mentions '{sport}' — core must never learn "
|
||||||
|
"sport names; add an override point instead.")
|
||||||
|
|
||||||
|
|
||||||
|
class TestInstallRootResolution:
|
||||||
|
"""Guards the depth bug the sports.py -> package move introduced.
|
||||||
|
|
||||||
|
The move was byte-identical in every class body, but `__file__` gained a
|
||||||
|
directory, so `Path(__file__).resolve().parents[2]` silently changed from
|
||||||
|
the repo root to `<root>/src`. Textual identity is not semantic identity
|
||||||
|
when code measures its own location: these tests assert the resolved
|
||||||
|
values, not the index.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def test_install_root_is_the_repo_root(self):
|
||||||
|
from src.base_classes.sports.core import _INSTALL_ROOT
|
||||||
|
|
||||||
|
# The repo root is the directory that actually holds src/ and assets/.
|
||||||
|
assert (_INSTALL_ROOT / "src").is_dir()
|
||||||
|
assert (_INSTALL_ROOT / "src" / "base_classes" / "sports").is_dir()
|
||||||
|
assert _INSTALL_ROOT.name != "src", (
|
||||||
|
"_INSTALL_ROOT resolved to src/ — the parents[] depth is off by "
|
||||||
|
"one, which is exactly the regression the package move caused.")
|
||||||
|
|
||||||
|
def test_resolve_project_path_roots_at_repo_not_src(self):
|
||||||
|
from src.base_classes.sports.core import SportsCore, _INSTALL_ROOT
|
||||||
|
|
||||||
|
resolved = SportsCore._resolve_project_path(None, Path("assets/fonts"))
|
||||||
|
assert resolved == _INSTALL_ROOT / "assets" / "fonts"
|
||||||
|
assert "src" not in resolved.relative_to(_INSTALL_ROOT).parts
|
||||||
|
|
||||||
|
def test_absolute_paths_pass_through_unchanged(self):
|
||||||
|
from src.base_classes.sports.core import SportsCore
|
||||||
|
|
||||||
|
absolute = Path("/tmp/some/logo/dir")
|
||||||
|
assert SportsCore._resolve_project_path(None, absolute) == absolute
|
||||||
|
|
||||||
|
def test_font_root_and_project_path_share_one_anchor(self):
|
||||||
|
"""Both consumers must derive from the same constant, so a future
|
||||||
|
move needs exactly one line changed rather than two."""
|
||||||
|
from src.base_classes.sports.core import SportsCore, _INSTALL_ROOT
|
||||||
|
|
||||||
|
assert SportsCore._font_root(None) == str(_INSTALL_ROOT)
|
||||||
@@ -0,0 +1,585 @@
|
|||||||
|
"""Tests for the methods promoted onto SportsUpcoming / SportsRecent /
|
||||||
|
SportsLive from the nine plugin copies (phase B1 of the sports unification;
|
||||||
|
see docs/SPORTS_UNIFICATION.md).
|
||||||
|
|
||||||
|
Covered promotions:
|
||||||
|
- SportsRecent: `_get_zero_clock_duration` / `_clear_zero_clock_tracking`
|
||||||
|
(+ the `_zero_clock_timestamps` initializer).
|
||||||
|
- SportsLive: `_is_game_really_over` / `_detect_stale_games`
|
||||||
|
(+ `game_update_timestamps` / `stale_game_timeout`, and the
|
||||||
|
`FINAL_PERIOD` / `CLOCK_COUNTS_DOWN` class attributes).
|
||||||
|
- SportsUpcoming: `_select_games_for_display`.
|
||||||
|
- SportsRecent: `_select_recent_games_for_display`.
|
||||||
|
|
||||||
|
The live pair is the risk centre: `_detect_stale_games` is the only caller
|
||||||
|
that *removes* games, so `_is_game_really_over` returning a false positive
|
||||||
|
silently drops a live game from the display. The canonical form deliberately
|
||||||
|
declines to treat a missing clock as 0:00 — the plugin variant that did so
|
||||||
|
dropped clockless sports (baseball) from the FINAL_PERIOD-th period onward.
|
||||||
|
That regression is pinned by
|
||||||
|
`test_missing_clock_at_late_period_is_not_over`.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import logging
|
||||||
|
import sys
|
||||||
|
import time
|
||||||
|
from datetime import datetime, timezone
|
||||||
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
import requests
|
||||||
|
from freezegun import freeze_time
|
||||||
|
from PIL import Image
|
||||||
|
|
||||||
|
# src.base_classes.sports transitively imports the hardware matrix driver;
|
||||||
|
# stub it so these tests can import the sports base classes off-device.
|
||||||
|
sys.modules.setdefault("rgbmatrix", MagicMock())
|
||||||
|
|
||||||
|
from src.base_classes.hockey import Hockey, HockeyLive
|
||||||
|
from src.base_classes.sports import (
|
||||||
|
SportsCore,
|
||||||
|
SportsLive,
|
||||||
|
SportsRecent,
|
||||||
|
SportsUpcoming,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Harnesses
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class _UpcomingHarness(Hockey, SportsUpcoming):
|
||||||
|
"""Cheapest concrete SportsUpcoming: hockey extractor + cache-fed data.
|
||||||
|
|
||||||
|
`_favorite_key` is inherited from SportsCore — these harnesses
|
||||||
|
deliberately do NOT define it, so the selection tests exercise the real
|
||||||
|
seam rather than a local stand-in.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def _fetch_data(self):
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
class _RecentHarness(Hockey, SportsRecent):
|
||||||
|
def _fetch_data(self):
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
class _LiveHarness(HockeyLive):
|
||||||
|
def _fetch_data(self):
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
class _ThreePeriodLiveHarness(_LiveHarness):
|
||||||
|
"""Hockey-shaped: regulation ends after period 3."""
|
||||||
|
|
||||||
|
FINAL_PERIOD = 3
|
||||||
|
|
||||||
|
|
||||||
|
class _CountUpLiveHarness(_LiveHarness):
|
||||||
|
"""Soccer/AFL/NRL-shaped: the clock counts up, so 0:00 is kickoff."""
|
||||||
|
|
||||||
|
CLOCK_COUNTS_DOWN = False
|
||||||
|
|
||||||
|
|
||||||
|
class _IdFavoriteUpcomingHarness(_UpcomingHarness):
|
||||||
|
"""NRL-shaped: abbreviations are ambiguous, so favorites match on team id."""
|
||||||
|
|
||||||
|
def _favorite_key(self, game, side):
|
||||||
|
team_id = game.get(f"{side}_id")
|
||||||
|
return str(team_id) if team_id is not None else None
|
||||||
|
|
||||||
|
|
||||||
|
class _IdFavoriteRecentHarness(_RecentHarness):
|
||||||
|
def _favorite_key(self, game, side):
|
||||||
|
team_id = game.get(f"{side}_id")
|
||||||
|
return str(team_id) if team_id is not None else None
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def build_manager(monkeypatch, tmp_path):
|
||||||
|
"""Factory for concrete sports managers: mocked display/cache managers,
|
||||||
|
logo dir redirected to tmp, background service stubbed, and the requests
|
||||||
|
session rigged to prove nothing hits the network."""
|
||||||
|
monkeypatch.setattr(
|
||||||
|
SportsCore, "_initialize_logo_dir", lambda self, configured: tmp_path)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"src.base_classes.sports.core.get_background_service",
|
||||||
|
lambda *args, **kwargs: MagicMock())
|
||||||
|
|
||||||
|
def build(cls, **mode_cfg):
|
||||||
|
config = {
|
||||||
|
"timezone": "UTC",
|
||||||
|
"display": {},
|
||||||
|
"nhl_scoreboard": {"enabled": True, **mode_cfg},
|
||||||
|
}
|
||||||
|
display_manager = MagicMock()
|
||||||
|
display_manager.matrix.width = 128
|
||||||
|
display_manager.matrix.height = 32
|
||||||
|
display_manager.width = 128
|
||||||
|
display_manager.height = 32
|
||||||
|
display_manager.image = Image.new("RGB", (128, 32))
|
||||||
|
cache_manager = MagicMock()
|
||||||
|
cache_manager.get.return_value = None
|
||||||
|
cache_manager.cache_dir = str(tmp_path)
|
||||||
|
manager = cls(config, display_manager, cache_manager,
|
||||||
|
logging.getLogger("test_sports_modes_promotions"),
|
||||||
|
"nhl")
|
||||||
|
manager.session = MagicMock()
|
||||||
|
manager.session.get.side_effect = requests.exceptions.ConnectionError(
|
||||||
|
"promotion tests are offline")
|
||||||
|
return manager
|
||||||
|
|
||||||
|
return build
|
||||||
|
|
||||||
|
|
||||||
|
def game(game_id="1", home="BOS", away="TOR", start=None, home_id=None,
|
||||||
|
away_id=None, **extra):
|
||||||
|
g = {
|
||||||
|
"id": game_id,
|
||||||
|
"home_abbr": home,
|
||||||
|
"away_abbr": away,
|
||||||
|
"home_id": home_id,
|
||||||
|
"away_id": away_id,
|
||||||
|
"start_time_utc": start,
|
||||||
|
}
|
||||||
|
g.update(extra)
|
||||||
|
return g
|
||||||
|
|
||||||
|
|
||||||
|
def at(day, hour=12):
|
||||||
|
return datetime(2026, 1, day, hour, tzinfo=timezone.utc)
|
||||||
|
|
||||||
|
|
||||||
|
def _ids(games):
|
||||||
|
return [g["id"] for g in games]
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Tier 1 — zero-clock tracking (SportsRecent)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestZeroClockTracking:
|
||||||
|
def test_initializer_present_and_empty(self, build_manager):
|
||||||
|
manager = build_manager(_RecentHarness)
|
||||||
|
assert manager._zero_clock_timestamps == {}
|
||||||
|
|
||||||
|
def test_first_call_returns_zero_and_starts_tracking(self, build_manager):
|
||||||
|
manager = build_manager(_RecentHarness)
|
||||||
|
assert manager._get_zero_clock_duration("g1") == 0.0
|
||||||
|
assert "g1" in manager._zero_clock_timestamps
|
||||||
|
|
||||||
|
def test_subsequent_call_returns_elapsed_seconds(self, build_manager):
|
||||||
|
manager = build_manager(_RecentHarness)
|
||||||
|
with freeze_time("2026-01-20 12:00:00") as frozen:
|
||||||
|
assert manager._get_zero_clock_duration("g1") == 0.0
|
||||||
|
frozen.tick(45)
|
||||||
|
assert manager._get_zero_clock_duration("g1") == pytest.approx(45.0)
|
||||||
|
frozen.tick(15)
|
||||||
|
assert manager._get_zero_clock_duration("g1") == pytest.approx(60.0)
|
||||||
|
|
||||||
|
def test_tracking_is_per_game(self, build_manager):
|
||||||
|
manager = build_manager(_RecentHarness)
|
||||||
|
with freeze_time("2026-01-20 12:00:00") as frozen:
|
||||||
|
manager._get_zero_clock_duration("g1")
|
||||||
|
frozen.tick(30)
|
||||||
|
assert manager._get_zero_clock_duration("g2") == 0.0
|
||||||
|
assert manager._get_zero_clock_duration("g1") == pytest.approx(30.0)
|
||||||
|
|
||||||
|
def test_clear_resets_tracking(self, build_manager):
|
||||||
|
manager = build_manager(_RecentHarness)
|
||||||
|
with freeze_time("2026-01-20 12:00:00") as frozen:
|
||||||
|
manager._get_zero_clock_duration("g1")
|
||||||
|
frozen.tick(30)
|
||||||
|
manager._clear_zero_clock_tracking("g1")
|
||||||
|
assert "g1" not in manager._zero_clock_timestamps
|
||||||
|
# Restarts from zero after clearing.
|
||||||
|
assert manager._get_zero_clock_duration("g1") == 0.0
|
||||||
|
|
||||||
|
def test_clear_unknown_game_is_a_noop(self, build_manager):
|
||||||
|
manager = build_manager(_RecentHarness)
|
||||||
|
manager._clear_zero_clock_tracking("never-seen") # must not raise
|
||||||
|
assert manager._zero_clock_timestamps == {}
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Tier 2a — _is_game_really_over (SportsLive)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestIsGameReallyOver:
|
||||||
|
def test_missing_clock_at_late_period_is_not_over(self, build_manager):
|
||||||
|
"""THE baseball regression: no `clock` key at all, period 7.
|
||||||
|
|
||||||
|
The rejected variant coerced a missing clock to the literal "0:00" and
|
||||||
|
declared the game over — dropping every MLB game from the 5th inning
|
||||||
|
onward, since baseball has no game clock and `period` is the inning.
|
||||||
|
A missing clock must fail safe.
|
||||||
|
"""
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
g = game(period=7, period_text="Top 7th")
|
||||||
|
assert "clock" not in g
|
||||||
|
assert manager._is_game_really_over(g) is False
|
||||||
|
|
||||||
|
def test_none_clock_at_late_period_is_not_over(self, build_manager):
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
assert manager._is_game_really_over(
|
||||||
|
game(clock=None, period=7, period_text="Top 7th")) is False
|
||||||
|
|
||||||
|
def test_non_string_clock_at_late_period_is_not_over(self, build_manager):
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
assert manager._is_game_really_over(
|
||||||
|
game(clock=0, period=9, period_text="Bot 9th")) is False
|
||||||
|
|
||||||
|
def test_blank_clock_at_late_period_is_not_over(self, build_manager):
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
assert manager._is_game_really_over(
|
||||||
|
game(clock=" ", period=5, period_text="5th")) is False
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("period_text", ["Final", "final", "Final/OT",
|
||||||
|
"FINAL", "Final - SO"])
|
||||||
|
def test_final_period_text_is_over(self, build_manager, period_text):
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
assert manager._is_game_really_over(
|
||||||
|
game(clock="12:00", period=2, period_text=period_text)) is True
|
||||||
|
|
||||||
|
def test_none_period_text_does_not_raise(self, build_manager):
|
||||||
|
"""All nine plugin copies called `.lower()` on `game.get("period_text", "")`,
|
||||||
|
which is None when the key is present-but-None; `_detect_stale_games`
|
||||||
|
has no try/except around the call."""
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
assert manager._is_game_really_over(
|
||||||
|
game(clock="12:00", period=2, period_text=None)) is False
|
||||||
|
|
||||||
|
def test_missing_period_text_does_not_raise(self, build_manager):
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
assert manager._is_game_really_over(game(clock="12:00", period=2)) is False
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"clock", ["0:00", ":00", "00", "000", " 0:00 ", "00:00", "0", "0000"])
|
||||||
|
def test_expired_clock_at_final_period_is_over(self, build_manager, clock):
|
||||||
|
"""Every spelling of a zeroed clock counts, not a hand-listed few.
|
||||||
|
|
||||||
|
"00:00" is the one that motivated comparing numerically: it normalizes
|
||||||
|
to "0000", which matched none of the literals the plugin copies listed,
|
||||||
|
so a two-digit-minute expired clock kept the game on screen forever.
|
||||||
|
"""
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
assert manager._is_game_really_over(
|
||||||
|
game(clock=clock, period=4, period_text="Q4")) is True
|
||||||
|
|
||||||
|
def test_none_period_at_expired_clock_does_not_raise(self, build_manager):
|
||||||
|
"""`period` present-but-None: `None >= FINAL_PERIOD` is a TypeError, and
|
||||||
|
`_detect_stale_games` has no try/except — the same failure shape as the
|
||||||
|
`period_text` case above."""
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
assert manager._is_game_really_over(
|
||||||
|
game(clock="0:00", period=None, period_text="Q4")) is False
|
||||||
|
|
||||||
|
def test_none_period_with_running_clock_does_not_raise(self, build_manager):
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
assert manager._is_game_really_over(
|
||||||
|
game(clock="8:12", period=None, period_text="Q2")) is False
|
||||||
|
|
||||||
|
def test_expired_clock_after_final_period_is_over(self, build_manager):
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
assert manager._is_game_really_over(
|
||||||
|
game(clock="0:00", period=5, period_text="OT")) is True
|
||||||
|
|
||||||
|
def test_expired_clock_before_final_period_is_not_over(self, build_manager):
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
assert manager._is_game_really_over(
|
||||||
|
game(clock="0:00", period=3, period_text="Q3")) is False
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("clock", [":40", "0:40", "1:00"])
|
||||||
|
def test_running_clock_is_not_over(self, build_manager, clock):
|
||||||
|
"""Sub-minute clocks like ':40' are legitimate, not expired."""
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
assert manager._is_game_really_over(
|
||||||
|
game(clock=clock, period=4, period_text="Q4")) is False
|
||||||
|
|
||||||
|
def test_defaults_are_four_period_countdown(self):
|
||||||
|
assert SportsLive.FINAL_PERIOD == 4
|
||||||
|
assert SportsLive.CLOCK_COUNTS_DOWN is True
|
||||||
|
|
||||||
|
def test_final_period_override_three(self, build_manager):
|
||||||
|
"""Hockey-shaped subclass: regulation ends after period 3."""
|
||||||
|
manager = build_manager(_ThreePeriodLiveHarness)
|
||||||
|
assert manager.FINAL_PERIOD == 3
|
||||||
|
assert manager._is_game_really_over(
|
||||||
|
game(clock="0:00", period=3, period_text="P3")) is True
|
||||||
|
assert manager._is_game_really_over(
|
||||||
|
game(clock="0:00", period=2, period_text="P2")) is False
|
||||||
|
# And the unmodified default still requires period 4.
|
||||||
|
assert build_manager(_LiveHarness)._is_game_really_over(
|
||||||
|
game(clock="0:00", period=3, period_text="P3")) is False
|
||||||
|
|
||||||
|
def test_count_up_clock_never_expires(self, build_manager):
|
||||||
|
"""Soccer/AFL/NRL: 0:00 means kickoff, so the clock branch must not run."""
|
||||||
|
manager = build_manager(_CountUpLiveHarness)
|
||||||
|
assert manager.CLOCK_COUNTS_DOWN is False
|
||||||
|
for period in (1, 2, 4, 9):
|
||||||
|
assert manager._is_game_really_over(
|
||||||
|
game(clock="0:00", period=period, period_text="1st Half")) is False
|
||||||
|
|
||||||
|
def test_count_up_clock_still_honors_final_text(self, build_manager):
|
||||||
|
manager = build_manager(_CountUpLiveHarness)
|
||||||
|
assert manager._is_game_really_over(
|
||||||
|
game(clock="0:00", period=2, period_text="Final")) is True
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Tier 2b — _detect_stale_games (SportsLive)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestDetectStaleGames:
|
||||||
|
def test_initializer_defaults(self, build_manager):
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
assert manager.game_update_timestamps == {}
|
||||||
|
assert manager.stale_game_timeout == 300
|
||||||
|
|
||||||
|
def test_stale_timeout_is_configurable(self, build_manager):
|
||||||
|
manager = build_manager(_LiveHarness, stale_game_timeout=42)
|
||||||
|
assert manager.stale_game_timeout == 42
|
||||||
|
|
||||||
|
def test_mutates_caller_list_in_place_and_returns_none(self, build_manager):
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
fresh = game("1", period_text="P2", clock="10:00", period=2)
|
||||||
|
over = game("2", period_text="Final", clock="0:00", period=3)
|
||||||
|
games = [fresh, over]
|
||||||
|
original = games
|
||||||
|
|
||||||
|
result = manager._detect_stale_games(games)
|
||||||
|
|
||||||
|
assert result is None
|
||||||
|
assert games is original # same object, mutated in place
|
||||||
|
assert _ids(games) == ["1"]
|
||||||
|
|
||||||
|
def test_evicts_only_past_timeout_games(self, build_manager):
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
with freeze_time("2026-01-20 12:00:00"):
|
||||||
|
now = time.time()
|
||||||
|
manager.game_update_timestamps = {
|
||||||
|
"1": {"last_seen": now - 10}, # fresh
|
||||||
|
"2": {"last_seen": now - 299}, # just inside the timeout
|
||||||
|
"3": {"last_seen": now - 301}, # past the timeout
|
||||||
|
}
|
||||||
|
games = [game("1", period_text="P1", clock="10:00", period=1),
|
||||||
|
game("2", period_text="P1", clock="10:00", period=1),
|
||||||
|
game("3", period_text="P1", clock="10:00", period=1)]
|
||||||
|
manager._detect_stale_games(games)
|
||||||
|
|
||||||
|
assert _ids(games) == ["1", "2"]
|
||||||
|
assert "3" not in manager.game_update_timestamps
|
||||||
|
assert set(manager.game_update_timestamps) == {"1", "2"}
|
||||||
|
|
||||||
|
def test_unknown_last_seen_is_never_stale(self, build_manager):
|
||||||
|
"""last_seen == 0 (or no entry) means 'never recorded', not 'ancient'."""
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
games = [game("1", period_text="P1", clock="10:00", period=1),
|
||||||
|
game("2", period_text="P1", clock="10:00", period=1)]
|
||||||
|
manager.game_update_timestamps = {"1": {"last_seen": 0}}
|
||||||
|
manager._detect_stale_games(games)
|
||||||
|
assert _ids(games) == ["1", "2"]
|
||||||
|
|
||||||
|
def test_removes_games_that_are_really_over(self, build_manager):
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
manager.game_update_timestamps = {"2": {"last_seen": 0}}
|
||||||
|
games = [game("1", period_text="Q2", clock="5:00", period=2),
|
||||||
|
game("2", period_text="Final", clock="0:00", period=4)]
|
||||||
|
manager._detect_stale_games(games)
|
||||||
|
assert _ids(games) == ["1"]
|
||||||
|
assert "2" not in manager.game_update_timestamps
|
||||||
|
|
||||||
|
def test_keeps_clockless_late_game(self, build_manager):
|
||||||
|
"""The end-to-end form of the baseball regression: a clockless game in
|
||||||
|
the 7th must survive the removal path."""
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
games = [game("mlb-1", period=7, period_text="Top 7th")]
|
||||||
|
manager._detect_stale_games(games)
|
||||||
|
assert _ids(games) == ["mlb-1"]
|
||||||
|
|
||||||
|
def test_games_without_id_are_skipped(self, build_manager):
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
no_id = {"home_abbr": "BOS", "away_abbr": "TOR",
|
||||||
|
"period_text": "Final", "clock": "0:00", "period": 4}
|
||||||
|
games = [no_id]
|
||||||
|
manager._detect_stale_games(games)
|
||||||
|
# `continue` fires before the "really over" check, so it stays.
|
||||||
|
assert games == [no_id]
|
||||||
|
|
||||||
|
def test_empty_list_is_tolerated(self, build_manager):
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
games = []
|
||||||
|
assert manager._detect_stale_games(games) is None
|
||||||
|
assert games == []
|
||||||
|
|
||||||
|
def test_removal_is_by_value_not_identity(self, build_manager):
|
||||||
|
"""Sharp edge worth pinning: `list.remove` compares with `dict.__eq__`,
|
||||||
|
so two structurally-equal dicts drop the FIRST occurrence."""
|
||||||
|
manager = build_manager(_LiveHarness)
|
||||||
|
first = game("1", period_text="Final", clock="0:00", period=4)
|
||||||
|
twin = dict(first)
|
||||||
|
games = [first, twin]
|
||||||
|
manager._detect_stale_games(games)
|
||||||
|
# Both entries are removed here (two iterations, two removals), but the
|
||||||
|
# first removal deletes `first`, not the dict being iterated.
|
||||||
|
assert games == []
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Tier 3a — _select_games_for_display (SportsUpcoming)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestSelectGamesForDisplay:
|
||||||
|
def test_no_favorites_returns_all_sorted_ascending(self, build_manager):
|
||||||
|
manager = build_manager(_UpcomingHarness)
|
||||||
|
games = [game("late", start=at(20)), game("early", start=at(10)),
|
||||||
|
game("mid", start=at(15))]
|
||||||
|
assert _ids(manager._select_games_for_display(games, [])) == [
|
||||||
|
"early", "mid", "late"]
|
||||||
|
|
||||||
|
def test_missing_start_time_sorts_last(self, build_manager):
|
||||||
|
manager = build_manager(_UpcomingHarness)
|
||||||
|
games = [game("none", start=None), game("early", start=at(10))]
|
||||||
|
assert _ids(manager._select_games_for_display(games, [])) == [
|
||||||
|
"early", "none"]
|
||||||
|
|
||||||
|
def test_filters_to_favorite_teams(self, build_manager):
|
||||||
|
manager = build_manager(_UpcomingHarness)
|
||||||
|
games = [game("1", home="BOS", away="TOR", start=at(10)),
|
||||||
|
game("2", home="NYR", away="PIT", start=at(11)),
|
||||||
|
game("3", home="MTL", away="BOS", start=at(12))]
|
||||||
|
assert _ids(manager._select_games_for_display(games, ["BOS"])) == ["1", "3"]
|
||||||
|
|
||||||
|
def test_respects_upcoming_games_to_show_per_team(self, build_manager):
|
||||||
|
manager = build_manager(_UpcomingHarness, upcoming_games_to_show=2)
|
||||||
|
games = [game(str(i), home="BOS", away="TOR", start=at(10 + i))
|
||||||
|
for i in range(5)]
|
||||||
|
assert _ids(manager._select_games_for_display(games, ["BOS"])) == ["0", "1"]
|
||||||
|
|
||||||
|
def test_game_between_two_favorites_counts_for_both(self, build_manager):
|
||||||
|
manager = build_manager(_UpcomingHarness, upcoming_games_to_show=1)
|
||||||
|
games = [game("shared", home="BOS", away="TOR", start=at(10)),
|
||||||
|
game("bos2", home="BOS", away="NYR", start=at(11)),
|
||||||
|
game("tor2", home="TOR", away="PIT", start=at(12))]
|
||||||
|
# "shared" fills both BOS's and TOR's single slot, so nothing else fits.
|
||||||
|
assert _ids(manager._select_games_for_display(
|
||||||
|
games, ["BOS", "TOR"])) == ["shared"]
|
||||||
|
|
||||||
|
def test_deduplicates_by_game_id(self, build_manager):
|
||||||
|
manager = build_manager(_UpcomingHarness, upcoming_games_to_show=5)
|
||||||
|
g = game("dupe", home="BOS", away="TOR", start=at(10))
|
||||||
|
assert _ids(manager._select_games_for_display(
|
||||||
|
[g, dict(g)], ["BOS"])) == ["dupe"]
|
||||||
|
|
||||||
|
def test_non_favorite_games_excluded(self, build_manager):
|
||||||
|
manager = build_manager(_UpcomingHarness)
|
||||||
|
games = [game("1", home="NYR", away="PIT", start=at(10))]
|
||||||
|
assert manager._select_games_for_display(games, ["BOS"]) == []
|
||||||
|
|
||||||
|
def test_favorite_key_seam_supports_id_matching(self, build_manager):
|
||||||
|
"""The NRL case: two clubs share the abbreviation 'NEW', so favorites
|
||||||
|
must be matched on team id. Only the seam changes — the promoted
|
||||||
|
method is identical."""
|
||||||
|
games = [
|
||||||
|
game("knights", home="NEW", away="SYD", home_id=1, away_id=2,
|
||||||
|
start=at(10)),
|
||||||
|
game("warriors", home="NEW", away="SYD", home_id=99, away_id=2,
|
||||||
|
start=at(11)),
|
||||||
|
]
|
||||||
|
abbr_manager = build_manager(_UpcomingHarness)
|
||||||
|
# Abbreviation matching cannot tell the two "NEW" clubs apart.
|
||||||
|
assert _ids(abbr_manager._select_games_for_display(games, ["NEW"])) == [
|
||||||
|
"knights", "warriors"]
|
||||||
|
|
||||||
|
id_manager = build_manager(_IdFavoriteUpcomingHarness)
|
||||||
|
assert _ids(id_manager._select_games_for_display(games, ["99"])) == [
|
||||||
|
"warriors"]
|
||||||
|
|
||||||
|
def test_favorite_key_none_never_matches(self, build_manager):
|
||||||
|
"""`_favorite_key` returning None (missing id) must not match, even
|
||||||
|
against a favorites list holding the string 'None'."""
|
||||||
|
manager = build_manager(_IdFavoriteUpcomingHarness)
|
||||||
|
games = [game("1", home="BOS", away="TOR", start=at(10))] # no ids
|
||||||
|
assert manager._select_games_for_display(games, ["None"]) == []
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Tier 3b — _select_recent_games_for_display (SportsRecent)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestSelectRecentGamesForDisplay:
|
||||||
|
def test_no_favorites_returns_all_sorted_descending(self, build_manager):
|
||||||
|
manager = build_manager(_RecentHarness)
|
||||||
|
games = [game("early", start=at(10)), game("late", start=at(20)),
|
||||||
|
game("mid", start=at(15))]
|
||||||
|
assert _ids(manager._select_recent_games_for_display(games, [])) == [
|
||||||
|
"late", "mid", "early"]
|
||||||
|
|
||||||
|
def test_missing_start_time_sorts_last(self, build_manager):
|
||||||
|
manager = build_manager(_RecentHarness)
|
||||||
|
games = [game("none", start=None), game("late", start=at(20))]
|
||||||
|
assert _ids(manager._select_recent_games_for_display(games, [])) == [
|
||||||
|
"late", "none"]
|
||||||
|
|
||||||
|
def test_respects_recent_games_to_show_per_team(self, build_manager):
|
||||||
|
manager = build_manager(_RecentHarness, recent_games_to_show=2)
|
||||||
|
games = [game(str(i), home="BOS", away="TOR", start=at(10 + i))
|
||||||
|
for i in range(5)]
|
||||||
|
# Most recent first.
|
||||||
|
assert _ids(manager._select_recent_games_for_display(
|
||||||
|
games, ["BOS"])) == ["4", "3"]
|
||||||
|
|
||||||
|
def test_game_between_two_favorites_counts_for_both(self, build_manager):
|
||||||
|
manager = build_manager(_RecentHarness, recent_games_to_show=1)
|
||||||
|
games = [game("shared", home="BOS", away="TOR", start=at(20)),
|
||||||
|
game("bos2", home="BOS", away="NYR", start=at(19)),
|
||||||
|
game("tor2", home="TOR", away="PIT", start=at(18))]
|
||||||
|
assert _ids(manager._select_recent_games_for_display(
|
||||||
|
games, ["BOS", "TOR"])) == ["shared"]
|
||||||
|
|
||||||
|
def test_deduplicates_by_game_id(self, build_manager):
|
||||||
|
manager = build_manager(_RecentHarness, recent_games_to_show=5)
|
||||||
|
g = game("dupe", home="BOS", away="TOR", start=at(10))
|
||||||
|
assert _ids(manager._select_recent_games_for_display(
|
||||||
|
[g, dict(g)], ["BOS"])) == ["dupe"]
|
||||||
|
|
||||||
|
def test_non_favorite_games_excluded(self, build_manager):
|
||||||
|
manager = build_manager(_RecentHarness)
|
||||||
|
games = [game("1", home="NYR", away="PIT", start=at(10))]
|
||||||
|
assert manager._select_recent_games_for_display(games, ["BOS"]) == []
|
||||||
|
|
||||||
|
def test_favorite_key_seam_supports_id_matching(self, build_manager):
|
||||||
|
games = [
|
||||||
|
game("knights", home="NEW", away="SYD", home_id=1, away_id=2,
|
||||||
|
start=at(10)),
|
||||||
|
game("warriors", home="NEW", away="SYD", home_id=99, away_id=2,
|
||||||
|
start=at(11)),
|
||||||
|
]
|
||||||
|
id_manager = build_manager(_IdFavoriteRecentHarness)
|
||||||
|
assert _ids(id_manager._select_recent_games_for_display(
|
||||||
|
games, ["99"])) == ["warriors"]
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Seam / inertness guards
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestPromotionShape:
|
||||||
|
def test_promoted_methods_live_on_the_right_classes(self):
|
||||||
|
assert hasattr(SportsRecent, "_get_zero_clock_duration")
|
||||||
|
assert hasattr(SportsRecent, "_clear_zero_clock_tracking")
|
||||||
|
assert hasattr(SportsRecent, "_select_recent_games_for_display")
|
||||||
|
assert hasattr(SportsUpcoming, "_select_games_for_display")
|
||||||
|
assert hasattr(SportsLive, "_is_game_really_over")
|
||||||
|
assert hasattr(SportsLive, "_detect_stale_games")
|
||||||
|
|
||||||
|
def test_modes_does_not_define_the_favorite_key_seam(self):
|
||||||
|
"""`_favorite_key` is a SportsCore override point. modes.py must call
|
||||||
|
it, never define it — this test fails if the seam is added in the
|
||||||
|
wrong file."""
|
||||||
|
import src.base_classes.sports.modes as modes
|
||||||
|
|
||||||
|
for cls in (SportsUpcoming, SportsRecent, SportsLive):
|
||||||
|
assert "_favorite_key" not in vars(cls)
|
||||||
|
assert "_favorite_key" not in modes.__dict__
|
||||||
@@ -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
|
||||||
@@ -0,0 +1,113 @@
|
|||||||
|
"""Version reporting must have exactly one answer.
|
||||||
|
|
||||||
|
`src.__version__` is the canonical core version. The plugin loader compares
|
||||||
|
plugin `ledmatrix_min_version` floors against it, and the plugin ecosystem
|
||||||
|
floors on the number recorded in `CHANGELOG.md` — so if those two disagree, a
|
||||||
|
plugin can declare a floor that is satisfied by a core which does not actually
|
||||||
|
ship the module it needs.
|
||||||
|
|
||||||
|
This has already gone wrong once. The `v3.1.0` tag was cut 2026-05-31, but
|
||||||
|
`src/__init__.py` was not bumped from `"1.0.0"` to `"3.1.0"` until 2026-07-12,
|
||||||
|
six weeks later. Every device installed from that release reports `1.0.0`,
|
||||||
|
which is below the `(2, 0, 0)` floor in `PluginLoader._warn_if_incompatible` —
|
||||||
|
so those users get no compatibility warning at all. See
|
||||||
|
`docs/SPORTS_UNIFICATION.md` (phase B4).
|
||||||
|
|
||||||
|
A tag is not available here, so the tag half of the check lives in
|
||||||
|
`scripts/check_release_version.py`. Wiring that script into CI (on pushed `v*`
|
||||||
|
tags and published releases) is a follow-up PR; until it lands, run it by hand
|
||||||
|
before tagging:
|
||||||
|
|
||||||
|
python scripts/check_release_version.py v3.2.0
|
||||||
|
|
||||||
|
Note: `src.plugin_system.__version__` is deliberately NOT checked. That module
|
||||||
|
versions the *plugin API* (it sits beside `__api_version__` and is documented as
|
||||||
|
such), which moves independently of the core version.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import re
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
import src
|
||||||
|
|
||||||
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||||
|
CHANGELOG = REPO_ROOT / "CHANGELOG.md"
|
||||||
|
|
||||||
|
# [0-9] rather than \d: \d also matches non-ASCII decimal digits, which int()
|
||||||
|
# happily parses, so a heading in Arabic-Indic numerals would pass the pattern
|
||||||
|
# and then mismatch confusingly. [ \t] rather than \s for the same class of
|
||||||
|
# reason -- \s matches newlines, so "##\n3.2.0" would read as a heading.
|
||||||
|
SEMVER = re.compile(r"^([0-9]+)\.([0-9]+)\.([0-9]+)$")
|
||||||
|
# Version headings look like "## 3.2.0". A leading "## Unreleased" section is
|
||||||
|
# allowed and skipped -- it is where module additions are staged before a bump.
|
||||||
|
HEADING = re.compile(
|
||||||
|
r"^##[ \t]+(?P<version>[0-9]+\.[0-9]+\.[0-9]+)[ \t]*$", re.MULTILINE)
|
||||||
|
|
||||||
|
|
||||||
|
def test_core_version_is_semver():
|
||||||
|
"""A floor comparison parses this string; it has to be parseable."""
|
||||||
|
assert SEMVER.match(src.__version__), (
|
||||||
|
f"src.__version__ is {src.__version__!r}, which is not X.Y.Z. "
|
||||||
|
"The loader's floor comparison cannot parse it."
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_changelog_documents_the_current_version():
|
||||||
|
"""The newest versioned CHANGELOG heading is the version we claim to be.
|
||||||
|
|
||||||
|
Plugins floor on the version recorded in the CHANGELOG as first shipping a
|
||||||
|
module. If the code says 3.2.0 and the CHANGELOG's newest entry is 3.1.0,
|
||||||
|
that record points at the wrong release.
|
||||||
|
"""
|
||||||
|
text = CHANGELOG.read_text(encoding="utf-8")
|
||||||
|
headings = HEADING.findall(text)
|
||||||
|
assert headings, "CHANGELOG.md has no '## X.Y.Z' version headings"
|
||||||
|
|
||||||
|
newest = headings[0]
|
||||||
|
assert newest == src.__version__, (
|
||||||
|
f"src.__version__ is {src.__version__!r} but the newest CHANGELOG "
|
||||||
|
f"heading is {newest!r}. Bump one to match the other: the CHANGELOG is "
|
||||||
|
"what plugin authors read to pick a ledmatrix_min_version floor."
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_changelog_versions_are_ordered_and_unique():
|
||||||
|
"""A duplicated or out-of-order heading makes 'first release shipping X'
|
||||||
|
ambiguous, which is exactly the question the sunset rule asks."""
|
||||||
|
text = CHANGELOG.read_text(encoding="utf-8")
|
||||||
|
versions = [tuple(int(p) for p in v.split(".")) for v in HEADING.findall(text)]
|
||||||
|
|
||||||
|
duplicates = {v for v in versions if versions.count(v) > 1}
|
||||||
|
assert not duplicates, f"CHANGELOG.md has duplicate version headings: {duplicates}"
|
||||||
|
|
||||||
|
assert versions == sorted(versions, reverse=True), (
|
||||||
|
"CHANGELOG.md version headings are not in descending order; "
|
||||||
|
f"got {['.'.join(map(str, v)) for v in versions]}"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_web_interface_version_tracks_the_core():
|
||||||
|
"""web_interface used to carry its own hardcoded "3.0.0", a third answer to
|
||||||
|
'what version is this'. It now re-exports the canonical one."""
|
||||||
|
web_interface = pytest.importorskip(
|
||||||
|
"web_interface", reason="web_interface needs Flask, which is optional here"
|
||||||
|
)
|
||||||
|
assert getattr(web_interface, "__version__", None) == src.__version__, (
|
||||||
|
"web_interface.__version__ has drifted from src.__version__; it should "
|
||||||
|
"re-export the canonical value rather than hardcode its own."
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_heading_pattern_is_strict_about_digits_and_whitespace():
|
||||||
|
"""`\\d` also matches non-ASCII decimal digits and `\\s` matches newlines,
|
||||||
|
either of which would let a malformed heading through and then fail the
|
||||||
|
comparison with a confusing message. Pin the tightened patterns."""
|
||||||
|
assert HEADING.findall("## 3.2.0\n") == ["3.2.0"]
|
||||||
|
assert HEADING.findall("##\t3.2.0 \n") == ["3.2.0"]
|
||||||
|
# A bare "##" whose version sits on the next line is not a heading.
|
||||||
|
assert HEADING.findall("##\n3.2.0\n") == []
|
||||||
|
# Arabic-Indic digits parse via int() but are not our version format.
|
||||||
|
assert HEADING.findall("## ٣.٢.٠\n") == []
|
||||||
|
assert SEMVER.match("٣.٢.٠") is None
|
||||||
@@ -2,5 +2,10 @@
|
|||||||
LED Matrix Web Interface V3
|
LED Matrix Web Interface V3
|
||||||
Modern web interface for controlling the LED Matrix display
|
Modern web interface for controlling the LED Matrix display
|
||||||
"""
|
"""
|
||||||
__version__ = "3.0.0"
|
|
||||||
|
# Re-exported, never hardcoded. This used to carry its own "3.0.0", a third
|
||||||
|
# answer to "what version is this" alongside the tag and src.__version__ —
|
||||||
|
# and disagreeing version numbers are what made plugin compatibility floors
|
||||||
|
# untrustworthy (see docs/SPORTS_UNIFICATION.md, phase B4).
|
||||||
|
from src import __version__ # noqa: F401
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user