diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 7bc4ffeb..80112162 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -74,4 +74,5 @@ jobs: test/test_sports_base_characterization.py \ test/test_element_style.py \ test/test_sports_core_promotions.py \ - test/test_sports_modes_promotions.py + test/test_sports_modes_promotions.py \ + test/test_sports_capabilities.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 69560434..7414632d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,6 +35,20 @@ release that ships it. 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`. + ### Changed - **Live games are no longer dropped when the feed omits a game clock.** `SportsLive._is_game_really_over` previously (in the baseball and UFC diff --git a/docs/SPORTS_UNIFICATION.md b/docs/SPORTS_UNIFICATION.md index 31858ff9..4ac7f34a 100644 --- a/docs/SPORTS_UNIFICATION.md +++ b/docs/SPORTS_UNIFICATION.md @@ -94,7 +94,8 @@ deprecation cycle. | `_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(game)` | Celebration wording (`"GOAL"` vs `"TOUCHDOWN"`) | `"SCORE"` — only consulted when `CelebrationMixin` is present | +| `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 | `" SCORES!"` — only consulted when `CelebrationMixin` is present | +| `win_phrase(team_abbr)` | Win-celebration wording | `" WINS!"` — mixin only | | `_favorite_key(game, side)` | Which view-model field identifies a team for favorites matching | `game["_abbr"]` | | `_config_schema_path()` | Plugin's `config_schema.json` — returning it routes `_get_layout_offset` through the `src.element_style` resolver (and gives it the defaults to compare against) | `None`, i.e. the classic inline `customization.layout` read | | `_font_root()` | Directory to resolve `assets/fonts` against | core install root | @@ -106,6 +107,7 @@ constants rather than behavior: |---|---|---| | `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 @@ -119,13 +121,63 @@ NRL fills it, and core never learns the string `"nrl"`. 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. A plugin needing an +ordering core does not ship calls `register_rotation_strategy` instead of core +growing a branch. + +`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. + ## 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 | +| **B1** ✅ | Promote the nine universal methods; convert `sports.py` → package | Characterization suite must stay green; no behavior change intended | +| **B2** ✅ | `CelebrationMixin` + rotation strategies as opt-in capabilities | Plugins that don't opt in have zero new code in their MRO; strategies checked against verbatim plugin transcriptions | | **B3** | Upstream `ScrollDisplay` as `src/common/sports_scroll.py`, reading `global_config['target_fps']` natively | Plugin copies remain until sunset | | **B4** | Bump to 3.2.0, record modules in CHANGELOG, migrate `ledmatrix_min` → `ledmatrix_min_version` | Gives plugins a version to floor on | | **B5** | Pilot one plugin per lineage (hockey, soccer, football) on guarded core imports; then the remaining six; then delete bundled copies | Pilot soaks before rollout; harness + golden suites gate each | diff --git a/src/base_classes/sports/capabilities/__init__.py b/src/base_classes/sports/capabilities/__init__.py new file mode 100644 index 00000000..e084ad7c --- /dev/null +++ b/src/base_classes/sports/capabilities/__init__.py @@ -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._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", +] diff --git a/src/base_classes/sports/capabilities/celebrations.py b/src/base_classes/sports/capabilities/celebrations.py new file mode 100644 index 00000000..539f45e1 --- /dev/null +++ b/src/base_classes/sports/capabilities/celebrations.py @@ -0,0 +1,379 @@ +"""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) + self.celebration_duration = mode_config.get("celebration_duration", 8) + # 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 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"Celebration ({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"Celebration 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 time.time() - celebration["started_at"] < self.celebration_duration: + try: + self._draw_celebration_layout(celebration, force_clear) + return True + except Exception as e: + self.logger.error(f"Error drawing celebration: {e}", exc_info=True) + 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) diff --git a/src/base_classes/sports/capabilities/rotation.py b/src/base_classes/sports/capabilities/rotation.py new file mode 100644 index 00000000..7a8560ba --- /dev/null +++ b/src/base_classes/sports/capabilities/rotation.py @@ -0,0 +1,228 @@ +"""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 = "" + + 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. + """ + 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] = 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 [] + preview = SmoothWeightedRotation(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``. + + A plugin needing an ordering core does not ship registers it here rather + than core growing a sport-specific branch. Re-registering a name replaces + it, so a plugin may also override a built-in for itself. + """ + if not name: + raise ValueError("rotation strategy name must be a non-empty string") + 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) diff --git a/test/test_sports_capabilities.py b/test/test_sports_capabilities.py new file mode 100644 index 00000000..31a991c5 --- /dev/null +++ b/test/test_sports_capabilities.py @@ -0,0 +1,764 @@ +"""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 +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) + + 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} + + +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_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 + + +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): + manager = celebrating(mode_config={"celebration_duration": 0}) + manager._check_for_score(game("g1", home_score=0)) + manager._check_for_score(game("g1", home_score=1)) + assert manager.display() is True + 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": 0}) + manager._check_for_score(game("g1", home_score=0)) + manager._check_for_score(game("g1", home_score=1)) + manager.display() + 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_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")])