feat(common): sports_view_model -- the reconciled game view model (sports family 10) (#796)

* feat(common): sports_view_model -- the reconciled game view model (sports family 10)

New hardware-free module src/common/sports_view_model.py, copied from
ledmatrix-plugins claude/family10-reconcile once the nine scoreboards made
_extract_game_details_common (nine bodies) one. It lives on the plugins'
SportsCore, so one mixin, SportsViewModelMixin, carries exactly that method,
the helpers it calls (_status_is_final, _score_text, _team_logo_url,
_extract_team_record), the _logo_filename seam and the two status-name
sets; no manager gains a method it lacks today. The module docstring lists
every key of the game dict with its type and default; keys are additive
only.

Owner decisions (2026-10-13):
- One final rule: state "post", not completed: false (a payload without
  the flag is judged on state and name), and not a not-played status (the
  union of the nine lists). A forfeit is final. A called-off game is never
  live, even while ESPN still reports it "in".
- Scores are integer text everywhere; a missing, empty or non-numeric
  score is "0".
- A time without an offset is UTC.
- Halftime reads "HALF" with no clock (each sport's own extractor; not in
  this module).
- Superset with safe defaults: period (0) and clock (""). period_text has
  none: the renderers read its absence (baseball) as "draw ESPN's status
  text", and an empty default blanked baseball's Recent "Final" in the
  harness.

- test/test_sports_view_model.py: the plugins' pinned cases (states, the
  final rule, scores, records, logos, ids, broadcast, dates, offsets, DST,
  malformed events), the seam, and the host contract (every self read
  documented, every key in the docstring's table). Mutation-checked (11
  mutants, all caught).
- test/test_sports_view_model_parity.py: with LEDMATRIX_PLUGINS, compares
  each method and both status sets with every plugin copy. Passes against
  claude/family10-reconcile; fails against the pre-reconcile tree.
- test_sports_twins.py, sports_card and sports_game_renderer tests pass
  with the reconciled plugins as the corpus.
- mypy ratchet, src/common/README.md, CHANGELOG (Unreleased: New modules),
  docs/SPORTS_UNIFICATION.md family 10 status and decisions, and the
  _logo_filename seam in the override-points table.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* docs: cite ledmatrix-plugins #654 for the family 10 reconcile

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test(sports): view-model parity requires SportsCore to inherit the mixin

A plugin without a copy counted as adopted if its source merely mentioned
the module. It now also has to list SportsViewModelMixin among SportsCore's
bases (CodeRabbit on #796).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Chuck
2026-10-15 13:07:38 -04:00
committed by GitHub
co-authored by Claude Opus 5.5
parent c5ce497c37
commit f4435d79e0
7 changed files with 905 additions and 3 deletions
+346
View File
@@ -0,0 +1,346 @@
"""src.common.sports_view_model: the game dict built from one ESPN event.
The cases follow ledmatrix-plugins' ``scripts/test_view_model.py`` (the table
the family 10 reconcile was checked against); the expected values are its
``EXPECTED_COMMON`` cells, which agree across all nine scoreboards once the
reconcile landed (nrl differs only through the ``_logo_filename`` seam and
its id favourites, which are host data). Each row is one change to one base
event, a scheduled AAA (home, id 1) v BBB (away, id 2) at 7:00 PM EDT, and
lists only the keys whose value differs from the base game. The host shows
times in America/New_York. Nothing reaches the network.
"""
import ast
import copy
import logging
from datetime import datetime
from pathlib import Path
import pytest
import pytz
from src.common import sports_view_model
from src.common.sports_view_model import SportsViewModelMixin
LOGO_DIR = Path("logos")
class Host(SportsViewModelMixin):
def __init__(self, config=None):
self.logger = logging.getLogger("test_sports_view_model")
self.config = config or {}
self.logo_dir = LOGO_DIR
self.favorite_teams = ["AAA"]
def _get_timezone(self):
return pytz.timezone("America/New_York")
def competitor(home_away, tid, abbr, name, record):
return {"id": tid, "homeAway": home_away, "score": "0",
"team": {"id": tid, "abbreviation": abbr, "name": name,
"logo": f"https://a.espncdn.com/i/teamlogos/{abbr.lower()}.png"},
"records": [{"type": "total", "summary": record}]}
def base_event():
return {"id": "401", "date": "2026-10-12T23:00Z", "competitions": [{
"id": "4011", "broadcast": "ESPN",
"status": {"displayClock": "0:00", "period": 0, "type": {
"name": "STATUS_SCHEDULED", "state": "pre", "completed": False,
"shortDetail": "10/12 - 7:00 PM EDT"}},
"competitors": [competitor("home", "1", "AAA", "Alpha", "3-1"),
competitor("away", "2", "BBB", "Bravo", "2-2")]}]}
BASE = {
"id": "401", "game_time": "7:00PM", "game_date": "10/12",
"start_time_utc": datetime(2026, 10, 12, 23, 0, tzinfo=pytz.UTC),
"status_text": "10/12 - 7:00 PM EDT", "is_live": False, "is_final": False,
"is_upcoming": True, "is_halftime": False, "is_period_break": False,
"broadcast": "ESPN", "home_abbr": "AAA", "home_id": "1", "home_score": "0",
"home_logo_path": LOGO_DIR / "AAA.png",
"home_logo_url": "https://a.espncdn.com/i/teamlogos/aaa.png",
"home_record": "3-1", "away_record": "2-2", "away_abbr": "BBB", "away_id": "2",
"away_score": "0", "away_logo_path": LOGO_DIR / "BBB.png",
"away_logo_url": "https://a.espncdn.com/i/teamlogos/bbb.png",
"is_within_window": True, "favorite_teams": ["AAA"],
"period": 0, "clock": "",
}
def comp(e):
return e["competitions"][0]
def side(e, home_away):
return next(c for c in comp(e)["competitors"] if c["homeAway"] == home_away)
def status(state, name, short, completed=False, drop_completed=False):
def edit(e):
st = {"name": name, "state": state, "completed": completed, "shortDetail": short}
if drop_completed:
del st["completed"]
comp(e)["status"] = {"displayClock": "5:32", "period": 2, "type": st}
return edit
def scores(home, away):
def edit(e):
side(e, "home")["score"], side(e, "away")["score"] = home, away
return edit
def drop_scores(e):
for c in comp(e)["competitors"]:
del c["score"]
def both(fn):
def edit(e):
for c in comp(e)["competitors"]:
fn(c)
return edit
def dated(date):
def edit(e):
e["date"] = date
return edit
def chain(*edits):
def edit(e):
for fn in edits:
fn(e)
return edit
def utc(*args):
return datetime(*args, tzinfo=pytz.UTC)
FINAL = status("post", "STATUS_FINAL", "Final", completed=True)
POST = {"is_upcoming": False, "is_final": True, "status_text": "Final"}
#: label -> (edit, extra host config, keys that differ from BASE; None: no game)
ROWS = {
"pre": (None, None, {}),
"short date format": (dated("2026-03-05T23:00Z"), {"display": {"use_short_date_format": True}},
{"game_date": "3/5", "game_time": "6:00PM",
"start_time_utc": utc(2026, 3, 5, 23, 0)}),
"in progress": (status("in", "STATUS_IN_PROGRESS", "5:32 - 2nd"), None,
{"is_live": True, "is_upcoming": False, "status_text": "5:32 - 2nd"}),
"halftime": (status("in", "STATUS_HALFTIME", "Halftime"), None,
{"is_live": True, "is_halftime": True, "is_upcoming": False,
"status_text": "Halftime"}),
"end of period": (status("in", "STATUS_END_PERIOD", "End of 1st"), None,
{"is_live": True, "is_period_break": True, "is_upcoming": False,
"status_text": "End of 1st"}),
"suspended while in": (status("in", "STATUS_SUSPENDED", "Suspended"), None,
{"is_upcoming": False, "status_text": "Suspended"}),
"final": (FINAL, None, POST),
"final, no completed flag": (status("post", "STATUS_FINAL", "Final", drop_completed=True),
None, POST),
"final, completed false": (status("post", "STATUS_FINAL", "Final"), None,
{"is_upcoming": False, "status_text": "Final"}),
"postponed": (status("post", "STATUS_POSTPONED", "Postponed"), None,
{"is_upcoming": False, "status_text": "Postponed"}),
"canceled": (status("post", "STATUS_CANCELED", "Canceled"), None,
{"is_upcoming": False, "status_text": "Canceled"}),
"delayed, completed": (status("post", "STATUS_DELAYED", "Delayed", completed=True), None,
{"is_upcoming": False, "status_text": "Delayed"}),
"forfeit, completed": (status("post", "STATUS_FORFEIT", "Forfeit", completed=True), None,
{**POST, "status_text": "Forfeit"}),
"forfeit pending": (status("post", "STATUS_FORFEIT_PENDING", "Forfeit", completed=True),
None, {"is_upcoming": False, "status_text": "Forfeit"}),
"scores int": (chain(FINAL, scores(21, 3)), None, {**POST, "home_score": "21", "away_score": "3"}),
"scores '21.0'": (chain(FINAL, scores("21.0", " 3 ")), None,
{**POST, "home_score": "21", "away_score": "3"}),
"scores dict": (chain(FINAL, scores({"value": 21.0, "displayValue": "21"},
{"value": 0, "displayValue": "4"})), None,
{**POST, "home_score": "21", "away_score": "4"}),
"scores None and ''": (chain(FINAL, scores(None, "")), None, POST),
"scores missing": (chain(FINAL, drop_scores), None, POST),
"scores not numbers": (chain(FINAL, scores("SO", "12 (3)")), None, {**POST, "away_score": "12"}),
"records 0-0 and 0-0-0": (lambda e: (side(e, "home")["records"][0].update(summary="0-0"),
side(e, "away")["records"][0].update(summary="0-0-0")),
None, {"home_record": "", "away_record": ""}),
"records missing": (both(lambda c: c.pop("records")), None,
{"home_record": "", "away_record": ""}),
"record, team-schedule shape": (both(lambda c: (c.pop("records"),
c.update(record=[{"displayValue": "7-0"}]))),
None, {"home_record": "7-0", "away_record": "7-0"}),
"no abbreviation": (both(lambda c: c["team"].pop("abbreviation")), None,
{"home_abbr": "Alp", "away_abbr": "Bra",
"home_logo_path": LOGO_DIR / "ALP.png",
"away_logo_path": LOGO_DIR / "BRA.png"}),
"logos array only": (both(lambda c: c["team"].update(
logo=None, logos=[{"href": "https://a.espncdn.com/x.png"}])), None,
{"home_logo_url": "https://a.espncdn.com/x.png", "away_logo_url": "https://a.espncdn.com/x.png"}),
"logo and logos": (both(lambda c: c["team"].update(logos=[{"href": "https://a.espncdn.com/x.png"}])),
None, {}),
"no logos": (both(lambda c: c["team"].pop("logo")), None,
{"home_logo_url": None, "away_logo_url": None}),
"competitor without id": (lambda e: side(e, "home").pop("id"), None, {}),
"no ids at all": (lambda e: (side(e, "home").pop("id"), side(e, "home")["team"].pop("id")),
None, {"home_id": None}),
"broadcast object": (lambda e: comp(e).update(broadcast={"names": ["ESPN"]}), None,
{"broadcast": "{'names': ['ESPN']}"}),
"broadcast missing": (lambda e: comp(e).pop("broadcast"), None, {"broadcast": ""}),
"date 'TBD'": (dated("TBD"), None, {"game_date": "", "game_time": "", "start_time_utc": None}),
"date +10:00": (dated("2026-10-13T19:40+10:00"), None,
{"game_date": "10/13", "game_time": "5:40AM",
"start_time_utc": utc(2026, 10, 13, 9, 40)}),
"date naive, read as UTC": (dated("2026-10-12T23:00:00"), None, {}),
"late UTC, previous local day": (dated("2026-10-13T03:30Z"), None,
{"game_time": "11:30PM", "start_time_utc": utc(2026, 10, 13, 3, 30)}),
"DST ends, 05:30Z (EDT)": (dated("2026-11-01T05:30Z"), None,
{"game_date": "11/01", "game_time": "1:30AM",
"start_time_utc": utc(2026, 11, 1, 5, 30)}),
"DST ends, 06:30Z (EST)": (dated("2026-11-01T06:30Z"), None,
{"game_date": "11/01", "game_time": "1:30AM",
"start_time_utc": utc(2026, 11, 1, 6, 30)}),
"DST starts, 07:30Z": (dated("2026-03-08T07:30Z"), None,
{"game_date": "03/08", "game_time": "3:30AM",
"start_time_utc": utc(2026, 3, 8, 7, 30)}),
"naive, repeated DST hour": (dated("2026-11-01T01:30:00"), None,
{"game_date": "10/31", "game_time": "9:30PM",
"start_time_utc": utc(2026, 11, 1, 1, 30)}),
"one competitor": (lambda e: comp(e)["competitors"].pop(), None, None),
"no competitors": (lambda e: comp(e).pop("competitors"), None, None),
"no competitions": (lambda e: e.update(competitions=[]), None, None),
"no status": (lambda e: comp(e).pop("status"), None, None),
"no date": (lambda e: e.pop("date"), None, None),
}
@pytest.mark.parametrize("label", ROWS)
def test_the_game_dict(label):
edit, config, changed = ROWS[label]
event = base_event()
if edit:
edit(event)
details, home, away, st, situation = Host(config)._extract_game_details_common(event)
if changed is None:
assert (details, home, away, st, situation) == (None,) * 5
return
assert details == {**BASE, **changed}
assert home is side(event, "home") and away is side(event, "away")
assert st is comp(event)["status"] and situation is None
def test_an_empty_event_is_no_game():
assert Host()._extract_game_details_common({}) == (None,) * 5
def test_the_situation_is_handed_back():
event = base_event()
comp(event)["situation"] = {"possession": "1"}
assert Host()._extract_game_details_common(event)[4] == {"possession": "1"}
def test_the_logo_filename_seam():
class ById(Host):
def _logo_filename(self, abbr, team_id):
return f"{abbr}_{team_id}.png"
details = ById()._extract_game_details_common(base_event())[0]
assert (details["home_logo_path"], details["away_logo_path"]) == (
LOGO_DIR / "AAA_1.png", LOGO_DIR / "BBB_2.png")
def test_favourites_are_a_copy():
host = Host()
host.favorite_teams = None
assert host._extract_game_details_common(base_event())[0]["favorite_teams"] == []
host.favorite_teams = ["AAA"]
details = host._extract_game_details_common(base_event())[0]
assert details["favorite_teams"] == ["AAA"] and details["favorite_teams"] is not host.favorite_teams
@pytest.mark.parametrize("status_type, final", [
({"state": "post", "name": "STATUS_FINAL", "completed": True}, True),
({"state": "post", "name": "STATUS_FINAL"}, True),
({"state": "post", "name": "status_final", "completed": True}, True),
({"state": "post", "name": "STATUS_FINAL", "completed": False}, False),
({"state": "post", "name": "STATUS_FORFEIT", "completed": True}, True),
({"state": "in", "name": "STATUS_IN_PROGRESS"}, False),
({"state": "post"}, True),
(None, False),
("post", False),
] + [({"state": "post", "name": n, "completed": True}, False)
for n in sorted(SportsViewModelMixin._NOT_PLAYED_STATUS_NAMES)])
def test_status_is_final(status_type, final):
assert SportsViewModelMixin._status_is_final(status_type) is final
def test_the_status_lists():
assert SportsViewModelMixin._CALLED_OFF_STATUS_NAMES == {
"STATUS_POSTPONED", "STATUS_CANCELED", "STATUS_CANCELLED",
"STATUS_SUSPENDED", "STATUS_ABANDONED"}
assert SportsViewModelMixin._NOT_PLAYED_STATUS_NAMES == (
SportsViewModelMixin._CALLED_OFF_STATUS_NAMES
| {"STATUS_DELAYED", "STATUS_FORFEIT_PENDING", "STATUS_NO_CONTEST_PENDING"})
@pytest.mark.parametrize("score, text", [
(21, "21"), (21.9, "21"), ("21", "21"), ("21.0", "21"), (" 7 ", "7"), ("", "0"),
(None, "0"), ("SO", "0"), ("12 (3)", "12"), ({"value": 21.0}, "21"),
({"value": 0, "displayValue": "4"}, "4"), ({"value": None}, "0"), ([], "0"),
(float("nan"), "0"), (float("inf"), "0"),
])
def test_score_text(score, text):
assert Host()._score_text(score) == text
def test_records_do_not_mutate_the_event():
event = base_event()
before = copy.deepcopy(event)
Host()._extract_game_details_common(event)
assert event == before
# ---------------------------------------------------------------------------
# Carrier and host contract
# ---------------------------------------------------------------------------
MEMBERS = ["_CALLED_OFF_STATUS_NAMES", "_NOT_PLAYED_STATUS_NAMES",
"_extract_game_details_common", "_extract_team_record", "_logo_filename",
"_score_text", "_status_is_final", "_team_logo_url"]
def _self_reads():
tree = ast.parse(Path(sports_view_model.__file__).read_text(encoding="utf-8"))
return {node.attr for node in ast.walk(tree)
if isinstance(node, ast.Attribute) and isinstance(node.ctx, ast.Load)
and isinstance(node.value, ast.Name) and node.value.id in ("self", "cls")}
class TestHostContract:
def test_the_mixin_carries_only_the_family(self):
assert sorted(n for n in vars(SportsViewModelMixin) if not n.startswith("__")) == MEMBERS
def test_every_host_read_is_documented(self):
undocumented = sorted(n for n in _self_reads() - set(MEMBERS)
if f"``{n}``" not in sports_view_model.__doc__)
assert undocumented == [], f"read but not in the host contract: {undocumented}"
def test_every_key_is_in_the_key_table(self):
doc = sports_view_model.__doc__
table = doc[doc.index("THE KEYS"):doc.index("THE RULES")]
missing = [k for k in BASE if k not in table]
assert missing == []
def test_no_other_shared_mixin_defines_these(self):
from src.common import (sports_display_rules, sports_favorites, sports_fetch,
sports_helpers, sports_live_fetch, sports_rankings,
sports_rotation, sports_shared)
others = [sports_shared.SportsCoreSharedMixin, sports_shared.SportsRecentSharedMixin,
sports_shared.SportsLiveSharedMixin, sports_helpers.SportsHelpersMixin,
sports_favorites.SportsFavoritesMixin, sports_rotation.SportsRotationMixin,
sports_rankings.SportsRankingsMixin, sports_fetch.SportsFetchMixin,
sports_live_fetch.SportsLiveFetchMixin,
sports_display_rules.SportsGameRulesMixin]
assert [m for m in MEMBERS if any(m in vars(o) for o in others)] == []
+134
View File
@@ -0,0 +1,134 @@
"""sports_view_model still matches every plugin copy.
``src.common.sports_view_model`` was copied from the scoreboards once family
10 had made ``_extract_game_details_common`` one body in all nine
``SportsCore`` classes, with the helpers it calls (``_status_is_final``,
``_score_text``, ``_team_logo_url``, ``_extract_team_record``, the
``_logo_filename`` seam) and the two status-name sets. The plugins delete
their copies once they floor on the release that ships this module. nrl
overrides the seam on ``BaseNrlManager``, not on ``SportsCore``. Until each has, a copy that changes on its own is a fix one
side has and the other lacks.
Point LEDMATRIX_PLUGINS at a ledmatrix-plugins checkout and each method is
compared with every plugin copy using ``scripts/sports_drift_report.py``'s own
normalisation (the AST with docstrings and annotations dropped), plus the
decorators. A copy that is gone counts as adopted when the plugin's ``sports.py`` names the module.
Without the variable this skips: core CI has no plugins checkout.
"""
import ast
import importlib.util
import os
from pathlib import Path
import pytest
from src.common import sports_view_model
REPO = Path(__file__).resolve().parents[1]
SPORTS = ("afl", "baseball", "basketball", "football", "hockey", "lacrosse",
"nrl", "soccer", "ufc")
CARRIER, MIXIN = "SportsCore", "SportsViewModelMixin"
METHODS = ("_extract_game_details_common", "_status_is_final", "_score_text",
"_team_logo_url", "_extract_team_record", "_logo_filename")
CONSTANTS = ("_CALLED_OFF_STATUS_NAMES", "_NOT_PLAYED_STATUS_NAMES")
def _drift_report():
"""scripts/sports_drift_report.py, loaded by path (scripts/ is no package)."""
spec = importlib.util.spec_from_file_location(
"sports_drift_report", REPO / "scripts" / "sports_drift_report.py")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
DRIFT = _drift_report()
def _plugins_root():
root = DRIFT.resolve_plugins_dir(os.environ.get("LEDMATRIX_PLUGINS"))
if root is None:
pytest.skip("set LEDMATRIX_PLUGINS to a ledmatrix-plugins checkout to "
"compare this module against the plugin copies")
return root
def _class(tree, name):
return next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == name)
def _method(cls, name):
return next((n for n in cls.body
if isinstance(n, ast.FunctionDef) and n.name == name), None)
def _constant(cls, name):
"""The value assigned to ``name`` in the class body, as an AST dump."""
return next((ast.dump(n.value) for n in cls.body
if isinstance(n, ast.AnnAssign) and getattr(n.target, "id", None) == name),
None)
def _adopted(source, tree):
"""The plugin imports core's module and its carrier inherits the mixin."""
bases = _class(tree, CARRIER).bases
return (sports_view_model.__name__ in source
and any(getattr(b, "id", getattr(b, "attr", None)) == MIXIN for b in bases))
def _fingerprint(node):
return (DRIFT._digest(node, DRIFT._Canonical()),
tuple(ast.unparse(d) for d in node.decorator_list))
def _plugin_tree(root, sport):
source = (root / f"{sport}-scoreboard" / "sports.py").read_text(encoding="utf-8")
return source, ast.parse(source)
OURS = _class(ast.parse(Path(sports_view_model.__file__).read_text(encoding="utf-8")), MIXIN)
@pytest.mark.parametrize("sport, name", [(s, n) for s in SPORTS for n in METHODS])
def test_every_remaining_plugin_copy_matches(sport, name):
source, tree = _plugin_tree(_plugins_root(), sport)
copy = _method(_class(tree, CARRIER), name)
if copy is None:
assert _adopted(source, tree), (
f"{sport}: no {name} on {CARRIER}, and {CARRIER} does not inherit {MIXIN}")
else:
assert _fingerprint(copy) == _fingerprint(_method(OURS, name)), (
f"{CARRIER}.{name} in {sport} differs from sports_view_model. "
f"Port the change to both, or stop treating it as shared.")
@pytest.mark.parametrize("sport, name", [(s, n) for s in SPORTS for n in CONSTANTS])
def test_every_remaining_status_set_matches(sport, name):
source, tree = _plugin_tree(_plugins_root(), sport)
copy = _constant(_class(tree, CARRIER), name)
if copy is None:
assert _adopted(source, tree), (
f"{sport}: no {name} on {CARRIER}, and {CARRIER} does not inherit {MIXIN}")
else:
assert copy == _constant(OURS, name), f"{CARRIER}.{name} in {sport} differs"
@pytest.mark.parametrize("sport", SPORTS)
def test_no_other_plugin_class_carries_a_copy(sport):
"""A copy on another class would shadow the shared one for that class."""
_, tree = _plugin_tree(_plugins_root(), sport)
strays = [f"{node.name}.{name}" for node in tree.body if isinstance(node, ast.ClassDef)
for name in METHODS if node.name != CARRIER and _method(node, name) is not None]
assert strays == []
def test_the_drift_report_calls_them_identical():
families = DRIFT.build(_plugins_root(), ("sports.py",))
rows = {(r["file"], r["family"]): r
for r in (DRIFT.summarise(k, v) for k, v in families.items())}
for name in METHODS:
row = rows.get(("sports.py", name))
assert row is None or row["worst_class_variants"] == 1, name