rick/LEDMatrix
1
0
Fork 0
mirror of https://github.com/ChuckBuilds/LEDMatrix.git synced 2026-10-05 14:55:08 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
11cfa639eef4d307d1b780e2a794d61523c59376
LEDMatrix/src/common/sports_shared.py
T
ChuckandClaude Opus 5.5 6fb2dc3595 fix(sports): honour every pending kickoff in the idle back-off, not just the first (#772) 
_note_scheduled_start_candidate kept one kickoff. While it was inside its
15-minute grace every later kickoff was refused, and by the time the grace
ended the later one had passed and was refused again as already past. So of
two favourites kicking off within 15 minutes of each other, the second lost
its own grace: if the first game was not live by then (a rain delay, a
postponement, ESPN slow to flip it) and ESPN had not flipped the second
either, the back-off went straight back to its ceiling and the second game
was noticed up to that late.

Later kickoffs now wait in a short queue (_later_scheduled_starts, the
earliest 8). When the current kickoff's grace ends, the earliest queued one
still inside its own grace takes over -- including one that has already
passed. A kickoff still holds the live cadence for at most its own grace, so
a postponed game costs the same quarter of an hour as before, and
_next_scheduled_start_ts keeps its meaning for anything that reads or sets
it. The promotion is a module function, so the mixin's method set is
unchanged.

Table tests replay the idle loop on a fake clock over kickoff schedules;
mutation-checked (the old code fails 10 of the new tests).

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-05 09:52:22 -04:00

1491 lines
72 KiB
Python
Raw Blame History

"""The sports.py surface that is byte-identical in every scoreboard.
Nine plugins ship their own ``sports.py`` -- 41,326 lines in total. Comparing
executable ASTs across the eight that share a lineage, 48 method bodies are
byte-identical in all eight: 1,007 lines carried in eight copies, so 8,056
duplicated lines that must be edited eight times to fix once.
They are the parts with no sport in them. The selection and rotation engine
(``_round_robin_favorites``, ``_favorites_first``, ``_compose_selection``,
``_check_ranking_coverage``, ``_game_divisions``, ``_normalise_quality``), the
font/colour/date subsystem (``_scale_headline_fonts``, ``_scorebug_font``,
``_resolve_font_size``, ``_format_game_date``, ``_font_color``), and the
switch-mode upcoming card (``_draw_upcoming_center_switch``). Nothing here knows
what an inning or a possession is.
Mixins rather than free functions, because every one of these reads host state
-- ``self.config``, ``self.fonts``, ``self.logger``, ``self.display_width``.
Rewriting 48 bodies into free functions would be a rewrite, not a move; as
mixins the bodies move verbatim, which is what keeps the renders identical.
THREE OF THE 48 ARE DELIBERATELY LEFT BEHIND
--------------------------------------------
Byte-identical bodies are not automatically safe to move: a body can bind a
module-level name that differs per plugin, and then it only *looks* the same.
- ``_get_timezone`` calls ``resolve_timezone``, imported from a per-plugin
module (``hockey_timezone``, ``soccer_timezone``, ...). All eight of those
differ -- each carries its own ``_WRITEBACK_FIXED_IN`` version -- so moving
the caller here would silently bind every scoreboard to one plugin's copy.
- ``_extract_game_details`` and ``_fetch_data`` are ``@abstractmethod`` stubs.
They are the sport-specific contract; satisfying them from a mixin would let a
plugin instantiate without implementing its own sport.
``_resolve_font_path`` went the other way: it is a module-level function in
sports.py rather than a method, identical in all eight, and ``_scale_headline_fonts``
needs it -- so it is inlined below rather than left behind.
WHAT A HOST MUST PROVIDE
------------------------
Enumerated by walking every ``self.<attr>`` the mixins read and subtracting what
they define, so this list is derived rather than remembered. Everything below is
supplied by all eight scoreboards today.
State: ``config``, ``fonts``, ``logger``, ``display_width``, ``display_height``,
``display_manager``, ``league``, ``sport``, ``mode_config``, ``session``,
``headers``, ``favorite_teams``, ``games_list``, ``current_game_index``,
``last_game_switch``, ``last_update``, ``update_interval``,
``no_data_interval``, ``game_display_duration``, ``stale_game_timeout``,
``other_games_min_quality``, ``schedule_lookback_days``,
``schedule_lookahead_days``, ``game_update_timestamps``,
``_zero_clock_timestamps``, ``_logo_cache``, ``_selection_pools``,
``_ranking_coverage_logged_at``, ``_empty_live_streak``, ``_last_warning_time``,
``_score_grew``.
Methods that stay per-plugin, because they are not identical across the eight
(or, for ``_get_timezone``, because they bind per-plugin modules):
``_get_layout_offset``, ``_by_importance``, ``_other_games_window``,
``_upcoming_date_and_time_text``, ``_extract_game_details_common``,
``_load_division_team_ids``, ``_get_timezone``, ``_is_favorite_game``,
``_is_ranked_game``, ``_passes_other_filters``. (``_is_game_really_over``,
which ``_detect_stale_games`` below calls, was here too until the plugins
reconciled it; it is now ``src.common.sports_game_over``.)
Of the fourteen shared class constants, thirteen are identical everywhere and
live here. Only ``_SCORE_PROBE_TEXT`` varies -- afl and basketball reach three digits
a side and override it, the same two that override ``_SCORE_PROBE`` on
``SportsGameRendererMixin``.
TWINS IN sports_card
--------------------
Many of these have same-named twins in ``src/common/sports_card.py``, which the
scoreboards' ``game_renderer.py`` uses. ``test/test_sports_twins.py`` calls
each pair with the same inputs (the plugins' fixture games in every payload
shape, plus edge cases) and splits them in two:
- Identical: ``_card_option``, ``_vs_text``, ``_format_game_time``,
``_coerce_rgb``, ``_crisp_size``, ``_unshare_element_fonts`` (given the same
element map) and the constant tables. These are now thin wrappers over the
``sports_card`` function; ``_format_game_date`` and ``_schema_font_size``
share its body/parser while keeping their own setting, zone and cache.
``_resolve_font_size`` agrees too but keeps its body, because it dispatches
through the overridable ``_schema_font_size``/``_crisp_size``.
- Different, and pinned as they are: ``_side_is_favorite`` /
``_favorite_result`` / ``_recent_score_color`` (flat keys and the host's
favourites only), ``_weekday_for`` (the plugin's resolved zone, not
``config["timezone"]``), ``_font_color`` / ``_ELEMENT_FOR_FONT`` (another
element vocabulary), ``_element_color`` (passes ``SKIN_MODE``). Each shows
up in one display mode only, so which side is right is a product decision;
the test that pins it names the difference.
"""
from __future__ import annotations
import logging
import os
import sys
import time
from datetime import datetime, timedelta, timezone
from typing import Any, ClassVar, Dict, List, Optional, Tuple
import pytz
from src.common.espn_dates import ESPN_MAX_LIMIT, fetch_espn_scoreboard
import requests
from PIL import Image, ImageDraw
from src.common import sports_card as _card
from src.common.font_layout import load_truetype, resolve_asset_path
from src.common.text_helper import OUTLINE_SQUARE, draw_text_outlined
logger = logging.getLogger(__name__)
# How long a live mode may stretch its poll interval when nothing is happening,
# and the streak lengths that earn each stretch. Identical in all eight plugins.
_IDLE_SHORT_STREAK = 6
_IDLE_SHORT_FACTOR = 2
_IDLE_LONG_STREAK = 24
_IDLE_LONG_FACTOR = 6
_DEFAULT_LIVE_IDLE_MAX_SECONDS = 900
#: How long after a scheduled start to keep looking on the live cadence. ESPN
#: does not flip a game to in-progress exactly at kickoff, and an escalated
#: back-off treats each of those early looks as another empty one.
_KICKOFF_GRACE_SECONDS = 900
#: Fallback cadence around a kickoff when the manager has no update_interval.
_KICKOFF_POLL_FLOOR = 30
#: How many kickoffs after the current one a live manager remembers. Only the
#: earliest few can matter before the next look refreshes the list, so this
#: bounds the memory without dropping a kickoff the board would wait for.
_KICKOFF_QUEUE_MAX = 8
def _current_scheduled_start(host: Any, now: float) -> Optional[float]:
"""The kickoff a live manager is honouring now, promoting the next queued one.
``_next_scheduled_start_ts`` is the kickoff being honoured: the earliest
one ahead of us, or one that has just passed and is inside its grace.
Kickoffs behind it wait in ``_later_scheduled_starts``. When the current
one's grace runs out, the earliest queued kickoff that is not itself past
its grace takes over -- including one that has already passed, so a
second kickoff inside the first one's grace still gets a grace of its own.
"""
current: Optional[float] = getattr(host, "_next_scheduled_start_ts", None)
if current and current > now - _KICKOFF_GRACE_SECONDS:
return current
queued: Optional[List[float]] = getattr(host, "_later_scheduled_starts", None)
if queued:
alive = sorted(s for s in queued if s > now - _KICKOFF_GRACE_SECONDS)
current = alive.pop(0) if alive else None
host._later_scheduled_starts = alive
host._next_scheduled_start_ts = current
return current if current and current > now - _KICKOFF_GRACE_SECONDS else None
def _resolve_font_path(path: str) -> str:
"""Resolve a bundled font path without depending on the process cwd.
These fonts ship with the LEDMatrix core, and every call site here named
them relative to the working directory. That holds under the packaged
systemd unit, whose WorkingDirectory is the install root, and breaks
everywhere else -- the plugin safety harness, a manual run from $HOME, a
unit file written without WorkingDirectory. The failure is quiet: the
load raises, the caller falls back, and the scoreboard renders in PIL's
default face instead of the pixel font it was laid out for.
Resolution order: the path as given, relative to the cwd, when it
exists -- the order the scoreboards' own sports.py copies used, so a
process running from another checkout keeps that checkout's fonts --
then :func:`src.common.font_layout.resolve_asset_path` (the install
root), which returns the original string when neither exists so callers
still raise and fall back.
"""
if os.path.exists(path):
return path
return resolve_asset_path(path)
class SportsCoreSharedMixin:
"""The ``SportsCore`` bodies identical in all eight scoreboards."""
#: Design height the font scale is expressed against.
_FONT_DESIGN_HEIGHT: ClassVar[int] = 32
#: Fraction of the centre strip a score may grow into.
_SCORE_GROWTH_BUDGET: ClassVar[float] = 0.65
#: Widest score the scorebug sizes itself to hold. Leagues that reach three
#: digits a side override this with "000-000".
_SCORE_PROBE_TEXT: ClassVar[str] = "00-00"
#: Whether this sport's scorebug draws a score at all.
_DRAWS_SCORE: ClassVar[bool] = False
#: Fallback (font, size) rungs for a score that will not fit.
_NARROW_SCORE_RUNGS: ClassVar[Tuple[Tuple[str, int], ...]] = (
("4x6-font.ttf", 14), ("4x6-font.ttf", 7))
#: Hard ceiling on score growth, in multiples of the configured size.
_SCORE_MAX_GROWTH: ClassVar[int] = 2
#: Which colour setting owns each font slot.
_ELEMENT_FOR_FONT: ClassVar[Dict[str, str]] = {
"score": "score_text", "time": "period_text", "team": "team_text",
"detail": "detail_text", "status": "status_text"}
# The tables below are sports_card's (and font_layout's) values. The dicts
# are copies, so a caller that mutates one module's table -- or a subclass
# that replaces it -- does not reach into the other.
#: Default tint for a favourite team's finished game.
FAVORITE_RESULT_COLOR_DEFAULTS: ClassVar[Dict[str, Tuple[int, int, int]]] = dict(
_card.FAVORITE_RESULT_COLOR_DEFAULTS)
_MONTH_ABBR: ClassVar[Tuple[str, ...]] = _card.MONTH_ABBR
_WEEKDAY_ABBR: ClassVar[Tuple[str, ...]] = _card.WEEKDAY_ABBR
#: Bitmap fonts snap to their native pixel grid.
_FONT_PIXEL_GRID: ClassVar[Dict[str, int]] = dict(_card.FONT_PIXEL_GRID)
_FONT_NAME_ALIASES: ClassVar[Dict[str, str]] = dict(_card.FONT_NAME_ALIASES)
#: Accepted values for the other-games quality filter.
_QUALITY_CHOICES: ClassVar[frozenset] = frozenset({"any", "ranked"})
#: How long to stay quiet between ranking-coverage warnings.
_RANKING_COVERAGE_SECONDS: ClassVar[int] = 60 * 60
def _draw_scorebug_layout(self, game: Dict, force_clear: bool = False) -> None:
"""Placeholder draw method - subclasses should override."""
# This base method will be simple, subclasses provide specifics
try:
img = Image.new("RGB", (self.display_width, self.display_height), (0, 0, 0))
draw = ImageDraw.Draw(img)
status = game.get("status_text", "N/A")
self._draw_text_with_outline(draw, status, (2, 2), self.fonts["status"],
element="status_text")
self.display_manager.image.paste(img, (0, 0))
# Don't call update_display here, let subclasses handle it after drawing
except Exception as e:
self.logger.error(
f"Error in base _draw_scorebug_layout: {e}", exc_info=True
)
@classmethod
def _crisp_size(cls, font_file, desired):
"""Snap *desired* to the nearest size *font_file* renders crisply at.
A face with no known grid is returned unchanged, so a user-supplied
font is never second-guessed. The class's own tables are passed, so a
host that declares extra faces keeps them.
"""
return _card.crisp_size(font_file, desired,
cls._FONT_NAME_ALIASES, cls._FONT_PIXEL_GRID)
#: Absolute path of this plugin's directory, declared by the plugin
#: itself. The mixin cannot work it out -- see _plugin_dir.
_PLUGIN_DIR: ClassVar[Optional[str]] = None
def _plugin_dir(self) -> Optional[str]:
"""Directory of the plugin that owns this instance.
In sports.py these methods could just use ``__file__``. Here that is
src/common/, so the directory has to come from the plugin.
It is TOLD, not deduced. The first version walked the MRO for a class
whose module sits beside a config_schema.json. That works when a test
imports the plugin itself, and returns None under the real loader, for
a specific reason worth recording:
PluginLoader._namespace_plugin_modules renames every bare module a
plugin brought in (sports, game_renderer, ...) to
"_plg_<plugin_id>_<module>" and REMOVES the bare entry, so two
plugins owning a module of the same name cannot collide.
The class still reports ``__module__ == "sports"``, but
``sys.modules["sports"]`` no longer exists, so the walk finds no
__file__ and falls off the end. The failure was silent and expensive:
_plugin_dir() -> None
_schema_font_size() -> None for every element
-> a configured size equal to the schema default stops looking
like a default and is treated as a deliberate user choice
-> the snap to the font's pixel grid is skipped
-> 4x6-font.ttf renders at 6 instead of 7: 3px-wide glyphs
instead of 4px
On a 256x64 panel that made the odds, the team records and the date row
hard to read. It was found by a user counting pixels on the panel. No
gate here caught it: the tests imported plugins directly and the
safety harness loads them its own way, so neither reproduced the
loader's renaming.
The MRO walk stays as a fallback for hosts that declare no
_PLUGIN_DIR -- the plugins' own probe harnesses build classes with
``type()`` -- but it is no longer the primary answer.
"""
declared = getattr(self, "_PLUGIN_DIR", None)
if declared and os.path.isfile(os.path.join(declared, "config_schema.json")):
return declared
for cls in type(self).__mro__:
module = sys.modules.get(getattr(cls, "__module__", ""), None)
path = getattr(module, "__file__", None)
if not path:
continue
directory = os.path.dirname(os.path.abspath(path))
if os.path.isfile(os.path.join(directory, "config_schema.json")):
return directory
return None
def _schema_font_size(self, element_key):
"""The font_size this plugin's config_schema.json declares, or None."""
if not element_key:
return None
# Cached per class, not in sports_card's per-path cache: the display
# service rebuilds the class when it reloads a plugin, and that is
# what makes an edited schema take effect. Both caches parse through
# sports_card._read_schema_font_sizes.
cache = getattr(self.__class__, '_SCHEMA_FONT_SIZES', None)
if cache is None:
cache = {}
try:
directory = self._plugin_dir()
if directory is None:
raise FileNotFoundError("no config_schema.json on the MRO")
cache = _card._read_schema_font_sizes(
os.path.join(directory, 'config_schema.json'))
except Exception as exc:
# Say so. An unreadable schema is not cosmetic: every element's
# configured size then stops matching "the schema default", is
# treated as a deliberate user choice, and skips the snap to the
# font's pixel grid -- which renders 4x6-font.ttf at 6 instead
# of 7, a 3px-wide glyph instead of 4px. That shipped once,
# silently, and was found by a user counting pixels on a photo
# of the panel.
#
# Logged, not raised: a missing schema must not stop a plugin
# rendering. The cache is built once per class, so this cannot
# repeat per frame.
logger.warning(
"%s: could not read config_schema.json (%s: %s); every font "
"size will be treated as user-chosen and will skip its pixel "
"grid snap. Font sizes may render a pixel narrow.",
type(self).__name__, type(exc).__name__, exc)
cache = {}
self.__class__._SCHEMA_FONT_SIZES = cache
return cache.get(element_key)
def _resolve_font_size(self, element_config, element_key, default_size, font_name):
"""Size to render at: the user's choice, or a grid-snapped default.
A configured size counts as a real choice only when it differs from
the schema default. The web UI writes the whole schema default block
on every save, so "font_size == schema default" carries no intent and
would otherwise pin every install to an anti-aliased size forever.
"""
configured = (element_config or {}).get('font_size')
if configured is not None:
try:
configured = int(configured)
if configured != self._schema_font_size(element_key):
return configured
except (TypeError, ValueError):
pass
return self._crisp_size(font_name, default_size)
def _card_option(self, key: str, default: Any = None) -> Any:
"""Read one key from the scroll_card config block."""
return _card.scroll_card_option(self.config, key, default)
def _switch_upcoming_center(self) -> str:
"""Middle of the full-screen upcoming scorebug: 'vs', 'date_time' or 'none'."""
mode = str(self._card_option("switch_upcoming_center", "date_time")
or "date_time").lower()
if mode == "inherit":
mode = str(self._card_option("upcoming_center", "vs") or "vs").lower()
return mode if mode in ("vs", "date_time", "none") else "date_time"
def _vs_text(self) -> str:
"""Separator drawn between the teams -- "VS", "@", "at", anything."""
return _card.vs_text(self.config)
def _switch_date_format(self) -> str:
"""Date style for the full-screen scorebug.
Its own key rather than the shared ``date_format`` because the two
displays disagree about the default: the scroll card renders "Sep 19"
while _extract_game_details_common emits "9/19", the "numeric" style,
and this scorebug has always drawn it. Reading the shared key here
would restyle every existing panel on update -- and "leave it alone
when unset" is not available, because the core merges schema defaults
into the config on every load, so the key is never actually unset.
"inherit" opts into the scroll and Vegas setting.
"""
fmt = str(self._card_option("switch_date_format", "numeric") or "numeric").lower()
if fmt == "inherit":
fmt = str(self._card_option("date_format", "abbrev") or "abbrev").lower()
return fmt
def _format_game_date(self, date_text: str, game: Optional[Dict] = None) -> str:
"""Format an upcoming date per scroll_card.switch_date_format.
The formatting is sports_card's. What differs from the card's
``format_game_date`` is passed in: the setting (``switch_date_format``,
see :meth:`_switch_date_format`) and the weekday, which comes from
:meth:`_weekday_for` and so from this plugin's resolved timezone
when the game's start cannot place the printed date. The game goes
in too, so both formatters name the printed date's own weekday.
"""
raw = str(date_text or "").strip()
if not raw:
return raw
return _card._format_date_as(self._switch_date_format(), raw,
lambda: self._weekday_for(game),
self._MONTH_ABBR, game=game)
def _weekday_for(self, game: Optional[Dict]) -> str:
"""Weekday abbreviation from the game's start time, or ''."""
if not game:
return ""
raw = game.get("start_time_utc") or game.get("start_time")
if not raw:
return ""
try:
start = raw if isinstance(raw, datetime) else datetime.fromisoformat(
str(raw).replace("Z", "+00:00"))
return self._WEEKDAY_ABBR[start.astimezone(self._get_timezone()).weekday()]
except (ValueError, TypeError, OverflowError):
return ""
def _format_game_time(self, time_text: str) -> str:
"""Return the time as-is (12h) or converted to 24h."""
return _card.format_game_time(self.config, time_text)
def _scorebug_font(self, draw, text: str, width: int):
"""The face this scorebug draws its date and time in.
Always the "time" face, which is what this display has used for both
rows for as long as it has existed: changing switch_upcoming_center
moves the two lines around, it is not meant to restyle them, so the
type stays put while the placement changes.
The single exception is text that cannot fit the panel at all. Only
the "weekday" date can do that -- "Fri Sep 19" measures 80px in an
8px face, on a board 64px wide -- and the smaller "detail" face is a
better answer there than running off both edges. Every other date and
time this display can produce fits, so in practice the face never
changes; it is a floor, not a style rule.
"""
font = self.fonts["time"]
if not text:
return font
try:
if draw.textlength(text, font=font) + 2 <= width:
return font
except (TypeError, ValueError):
return font
return self.fonts.get("detail") or font
def _draw_upcoming_center_switch(self, draw, game: Dict, center_y: int,
game_date: str, game_time: str,
display_width: Optional[int] = None,
display_height: Optional[int] = None,
date_element: str = 'date',
time_element: str = 'time',
second_row_y_offset: bool = True) -> bool:
"""Draw the middle of the full-screen upcoming scorebug.
Returns True when the header above it ("Next Game", or the league
name) should still be drawn. In "vs" and "none" the date and time move
out of the middle and into the top and bottom slots, mirroring the
scroll card -- and the top slot is where the header used to be, so the
caller drops it.
``date_element``/``time_element``/``second_row_y_offset`` exist only so
the layout-offset keys stay exactly what each plugin's schema
advertises; this sport's defaults are the common case.
"""
width = self.display_width if display_width is None else display_width
height = self.display_height if display_height is None else display_height
mode = self._switch_upcoming_center()
date_text, time_text = self._upcoming_date_and_time_text(
game_date, game_time, game)
swapped = bool(self._card_option("swap_date_time", False))
if mode == "date_time":
# Historically the date sat at center_y - 7 with the time 9px
# under it, and the time's row was derived from the date's, so a
# date y_offset moved the pair. Both still hold; the slots only
# trade places when swap_date_time is set, and hiding one line
# leaves the other where it was rather than re-centering the stack.
slots = [(time_element, time_text), (date_element, date_text)] if swapped \
else [(date_element, date_text), (time_element, time_text)]
row_y = center_y - 7
for index, (element, text) in enumerate(slots):
if index:
row_y += 9
if second_row_y_offset:
row_y += self._get_layout_offset(element, 'y_offset')
else:
row_y += self._get_layout_offset(element, 'y_offset')
if not text:
continue
font = self._scorebug_font(draw, text, width)
text_width = draw.textlength(text, font=font)
text_x = ((width - text_width) // 2
+ self._get_layout_offset(element, 'x_offset'))
self._draw_text_with_outline(
draw, text, (text_x, row_y), font
)
return True
if mode == "vs":
vs_text = self._vs_text()
if vs_text:
vs_width = draw.textlength(vs_text, font=self.fonts["score"])
vs_x = ((width - vs_width) // 2
+ self._get_layout_offset('score', 'x_offset'))
vs_y = (center_y - 3
+ self._get_layout_offset('score', 'y_offset'))
vs_x = self._aligned_x('score_text', vs_width, width, vs_x)
self._draw_text_with_outline(
draw, vs_text, (vs_x, vs_y), self.fonts["score"],
element="score_text"
)
# "vs" and "none" both push the date and time out to the edges, time
# on top unless swap_date_time says otherwise -- the same order the
# scroll card uses.
if swapped:
top_element, top_text = date_element, date_text
bottom_element, bottom_text = time_element, time_text
else:
top_element, top_text = time_element, time_text
bottom_element, bottom_text = date_element, date_text
if top_text:
top_font = self._scorebug_font(draw, top_text, width)
top_width = draw.textlength(top_text, font=top_font)
top_x = ((width - top_width) // 2
+ self._get_layout_offset(top_element, 'x_offset'))
top_y = 1 + self._get_layout_offset(top_element, 'y_offset')
self._draw_text_with_outline(
draw, top_text, (top_x, top_y), top_font
)
if bottom_text:
bottom_font = self._scorebug_font(draw, bottom_text, width)
bottom_width = draw.textlength(bottom_text, font=bottom_font)
bottom_x = ((width - bottom_width) // 2
+ self._get_layout_offset(bottom_element, 'x_offset'))
# Measured, not a fixed offset: the detail font is 6px in most
# plugins and 10px in soccer and nrl, where a fixed -7 ran the
# date off the panel.
ink_bottom = draw.textbbox((0, 0), bottom_text, font=bottom_font)[3]
bottom_y = (max(0, height - ink_bottom - 1)
+ self._get_layout_offset(bottom_element, 'y_offset'))
self._draw_text_with_outline(
draw, bottom_text, (bottom_x, bottom_y), bottom_font
)
return False
@staticmethod
def _coerce_rgb(value, fallback):
"""Turn a configured [R, G, B] list into a clamped (r, g, b) tuple."""
return _card.coerce_rgb(value, fallback)
@staticmethod
def _side_is_favorite(game: Dict, side: str, favorites: set) -> bool:
"""Is the home/away side of this game a favorite team?
Both the abbreviation and the ESPN id are checked, because a couple of
leagues (NRL) match favorites by id where abbreviations collide.
"""
for key in (f"{side}_abbr", f"{side}_id"):
value = game.get(key)
if value is not None and str(value).strip().upper() in favorites:
return True
return False
def _favorite_result(self, game: Dict) -> Optional[str]:
"""Say how the favorite team did in a finished game.
Returns 'win', 'loss' or 'tie', or None when there is no single team
to root for: no favorites configured, neither side is a favorite, or
*both* are -- a favorite-vs-favorite game has no losing side worth
flagging in red. Also None when the scores are not usable numbers.
"""
favorites = getattr(self, "favorite_teams", None) or []
favorites = {str(team).strip().upper() for team in favorites if str(team).strip()}
if not favorites:
return None
home_fav = self._side_is_favorite(game, "home", favorites)
away_fav = self._side_is_favorite(game, "away", favorites)
if home_fav == away_fav:
return None
try:
# int(float(...)) to match GameRenderer._side_score exactly -- the
# two paths must agree on what counts as a usable score.
home_score = int(float(str(game.get("home_score", "")).strip()))
away_score = int(float(str(game.get("away_score", "")).strip()))
except (TypeError, ValueError):
return None
if home_score == away_score:
return "tie"
favorite_score, other_score = (
(home_score, away_score) if home_fav else (away_score, home_score)
)
return "win" if favorite_score > other_score else "loss"
def _recent_score_color(self, game: Dict, default):
"""Fill color for a finished game's score, per favorite_result_colors."""
try:
settings = (self.config.get("customization") or {}).get(
"favorite_result_colors"
) or {}
if not settings.get("enabled", False):
return default
result = self._favorite_result(game)
if result is None:
return default
return self._coerce_rgb(
settings.get(f"{result}_color"),
self.FAVORITE_RESULT_COLOR_DEFAULTS[result],
)
except Exception:
self.logger.debug(
"Could not resolve favorite result color", exc_info=True
)
return default
def _score_font_size(self) -> int:
"""Pixel size the score is currently drawn at."""
return getattr(self.fonts.get("score"), "size", 8) or 8
def _time_font_size(self) -> int:
"""Pixel size the clock/date face is currently drawn at."""
return getattr(self.fonts.get("time"), "size", 8) or 8
def _user_chose_size(self, element_key: str) -> bool:
"""True when customization.<element>.font_size is a real choice.
The web UI's save flow writes the whole schema default block into
config.json on every save, whether or not the user touched that
section, so a size merely being PRESENT carries no intent. Only one
that differs from the schema default does.
"""
element = (self.config.get('customization', {}) or {}).get(element_key) or {}
configured = element.get('font_size')
if configured is None:
return False
try:
return int(configured) != self._schema_font_size(element_key)
except (TypeError, ValueError):
return False
def _grid_scaled_size(self, font):
"""(path, grid, size) for *font* regrown to this panel's height.
None when the panel is at or below the design height (nothing to do),
or when the face has no known pixel grid -- a user-supplied font is
never second-guessed, because we do not know what it renders crisply
at.
"""
path = getattr(font, 'path', None)
base = getattr(font, 'size', None)
if not base or not isinstance(path, str):
return None
face = os.path.basename(path)
grid = self._FONT_PIXEL_GRID.get(self._FONT_NAME_ALIASES.get(face, face))
if not grid:
return None
scale = float(self.display_height) / (self._FONT_DESIGN_HEIGHT or 32)
if scale <= 1.0:
return None
return path, grid, max(int(base), int(self._crisp_size(face, base * scale)))
def _scale_headline_fonts(self, fonts):
"""Grow the score with the panel, and hold the clock/date below it.
The score is the one number the card exists to show, and it was the
only element not sized from the panel. Worse, it was not even bigger
than its neighbours: PressStart2P renders crisply on an 8px grid, so
the 10px default snapped to 8 -- the same 8 the period/clock above it
and the game date below it are drawn at. Three lines of identical
type, none of them the headline, which is what makes the score read as
lower priority than the time and the date rather than the point of the
card.
So the score is sized from display_height and snapped to its face's
pixel grid (off the grid FreeType anti-aliases the strokes, and on an
LED matrix a part-lit pixel is a dim lamp rather than a soft edge),
then stepped back down that grid until it fits its share of the width.
The clock/date face is regrown the same way but held at least one grid
step below the score, so the ranking between them is visible rather
than implied.
A 32-tall panel scales by exactly 1.0 and is left byte-identical; a
size the user set explicitly is never overridden.
"""
self._score_grew = False
if not self._DRAWS_SCORE:
# No score on this screen, so none of the sizing below is for it.
return fonts
try:
scaled = None if self._user_chose_size('score_text') else \
self._grid_scaled_size(fonts.get('score'))
if scaled is not None:
path, grid, size = scaled
base = getattr(fonts['score'], 'size', size) or size
size = min(size, base * self._SCORE_MAX_GROWTH)
probe = ImageDraw.Draw(Image.new('RGB', (4, 4)))
budget = self.display_width * self._SCORE_GROWTH_BUDGET
# Measured from a fixed five-character score rather than the
# live one, so the card does not resize when a side passes 9.
while size > grid:
if probe.textlength(
self._SCORE_PROBE_TEXT,
font=load_truetype(path, size)) <= budget:
break
size -= grid
if size != getattr(fonts['score'], 'size', size):
fonts['score'] = load_truetype(path, size)
self._score_grew = True
if not self._score_grew and not self._user_chose_size('score_text') \
and self.display_height > self._FONT_DESIGN_HEIGHT:
# PressStart2P could not grow inside the budget -- its next crisp
# size is simply too wide for this panel. A narrower face still
# can: 4x6-font at 14px is nearly as tall as PressStart2P at 16
# and about half as wide. This matters beyond the score itself,
# because a card whose score never grows never reserves the
# centre either, so its logos stay at the uncapped 1.5x and are
# drawn straight over the score -- which is what a three-digit
# basketball score does on a 128x64 board.
probe = ImageDraw.Draw(Image.new('RGB', (4, 4)))
budget = self.display_width * self._SCORE_GROWTH_BUDGET
current = getattr(fonts.get('score'), 'size', 0) or 0
for _name, _size in self._NARROW_SCORE_RUNGS:
if _size <= current:
continue
_path = _resolve_font_path(f"assets/fonts/{_name}")
_candidate = load_truetype(_path, _size)
if probe.textlength(self._SCORE_PROBE_TEXT,
font=_candidate) <= budget:
fonts['score'] = _candidate
self._score_grew = True
break
scaled = None if self._user_chose_size('period_text') else \
self._grid_scaled_size(fonts.get('time'))
if scaled is not None:
path, grid, size = scaled
ceiling = getattr(fonts.get('score'), 'size', 0) or 0
if ceiling and size >= ceiling:
size = max(grid, ceiling - grid)
if size != getattr(fonts['time'], 'size', size):
fonts['time'] = load_truetype(path, size)
except Exception:
self.logger.debug("Headline font scaling skipped", exc_info=True)
return fonts
def _get_layout_offset(self, element: str, axis: str,
default: int = 0) -> int:
"""X/Y nudge for one element, from ``customization.layout``.
Promoted here so every scoreboard reads offsets the same way the
scroll card does. Each plugin still carries its own copy in its
bundled sports.py, which wins by MRO until that copy is deleted --
deleting it is what buys the alias handling (a plugin asking for
``score_text`` finds the ``score`` its users configured) and the
per-mode overrides, since this resolves through SKIN_MODE.
"""
from src.element_style import layout_offset
return layout_offset(self.config, element, axis, default,
getattr(self, "SKIN_MODE", None))
def _element_color(self, element: str, default: Tuple[int, int, int] = (255, 255, 255)):
"""Per-element text colour from customization.<element>.text_color.
Mode-aware through SKIN_MODE, so Live and Recent instances of the
same scoreboard resolve their own colours without any call site
passing a mode.
"""
from src.element_style import element_color as _shared
return _shared(self.config, element, default,
getattr(self, "SKIN_MODE", None))
def _element_visible(self, element: str, default: bool = True) -> bool:
"""Whether ``customization.<element>.visible`` allows this draw.
Mode-aware like the colour read, so a user can hide the records on the
recent card and keep them on the upcoming one.
"""
from src.element_style import element_visible
return element_visible(self.config, element, default,
getattr(self, "SKIN_MODE", None))
def _element_align(self, element: str, default: Optional[str] = None):
"""``customization.<element>.align``: 'left', 'center' or 'right'."""
from src.element_style import element_align
return element_align(self.config, element, default,
getattr(self, "SKIN_MODE", None))
def _element_scale(self, element: str, default: float = 1.0) -> float:
"""``customization.layout.<element>.scale`` -- logos, mostly."""
from src.element_style import element_scale
return element_scale(self.config, element, default,
getattr(self, "SKIN_MODE", None))
def _aligned_x(self, element: str, text_width: float, container_width: int,
centered_x: float) -> float:
"""Where a run of text starts, honouring ``align``.
Unset means "leave it exactly where it was", so this returns the
caller's own x rather than re-deriving a centre: these draws have
accumulated per-sport nudges and a centre computed here would not be
the same pixel.
"""
align = self._element_align(element)
if not align:
return centered_x
if align == 'left':
return 0
if align == 'right':
return max(0, container_width - text_width)
return centered_x
def _unshare_element_fonts(self, fonts):
"""Give each colourable element its own face object.
The colour a draw gets is resolved from the face it was handed, and
several of these loaders legitimately hand one object to more than one
element -- a size resolver that lands two elements on the same face, a
fallback that fills every key from one default, football's narrowing
step that deliberately shrinks the clock along with the score. Sharing
the object makes the element ambiguous and the colour unresolvable.
Re-instantiating from the same path and size gives a distinct object
with identical metrics, so nothing about the rendering changes; only
the ability to tell two elements apart does. Faces that cannot be
rebuilt (a BDF loaded through freetype.Face, anything without a usable
path) are left shared, and their draws stay white as before.
The body is sports_card's; this class's own element map is passed, so
the keys considered are the ones this class colours by.
"""
return _card.unshare_element_fonts(self.logger, fonts,
self._ELEMENT_FOR_FONT)
def _font_color(self, font, default: Tuple[int, int, int] = (255, 255, 255)):
"""Colour for whichever element owns this face.
The fallback for draw sites that were only ever handed a font. Prefer
``element=`` on :meth:`_draw_text_with_outline`, which needs none of
this. Shared with the scroll card's copy so the narrowing rule that
rescues bitmap-font colours lives in one place; the element vocabulary
stays this class's own, because its map says ``team_text`` where
sports_card's says ``team_name``.
"""
from src.common.sports_card import resolve_font_color
return resolve_font_color(
getattr(self, "config", None), getattr(self, "fonts", None), font,
default, self._ELEMENT_FOR_FONT, getattr(self, "SKIN_MODE", None))
def _draw_text_with_outline(
self, draw, text, position, font, fill=None, outline_color=(0, 0, 0),
element=None
):
"""Draw text with a black outline for better readability.
Pass ``element`` (``"score_text"``, ``"status_text"``, ...) wherever the
caller knows what it is drawing: the colour is then read by name, which
is exact. Without it the colour has to be inferred from the identity of
the font object, which cannot tell two elements apart when they share a
face -- the case every bitmap font is in, because a ``freetype.Face``
cannot be re-instantiated.
"""
# Disable anti-aliasing: pixel/bitmap fonts (e.g. PressStart2P) get
# anti-aliased into dim partial-lit pixels on a 1:1 LED matrix, muddying
# glyphs. 1-bit mode keeps strokes crisp.
# Defaults to the configured colour for whichever element owns
# this face rather than to white, so customization.<element>.text_color
# reaches every draw. The schema has offered those pickers all along
# and they only ever changed the font. An explicit fill still wins:
# the odds colours and the favourite-result score tint mean something
# the palette does not.
if element is not None:
# Named, so both questions can be answered exactly: whether this
# element is meant to be on screen at all, and what colour it is.
if not self._element_visible(element):
return
if fill is None:
fill = self._element_color(element)
elif fill is None:
fill = self._font_color(font)
draw.fontmode = "1"
# The eight-neighbour outline, then the text on top. Rasterized once
# and stamped nine times rather than drawn nine times; the pixels are
# the same (draw_text_outlined falls back to the nine draws wherever
# that is not proven).
draw_text_outlined(draw, position, text, font, fill, outline_color,
OUTLINE_SQUARE)
def _should_log(self, warning_type: str, cooldown: int = 60) -> bool:
"""True at most once per ``cooldown`` seconds, for rate-limiting a
warning. The cooldown is shared by every warning on this manager:
``warning_type`` is part of the signature scoreboards inherit, but
does not give each type its own cooldown."""
current_time = time.time()
if current_time - self._last_warning_time > cooldown:
self._last_warning_time = current_time
return True
return False
def _get_weeks_data(self) -> Optional[Dict]:
"""
Get partial data for immediate display while background fetch is in progress.
This fetches current/recent games only for quick response.
"""
try:
# Fetch current week and next few days for immediate display
now = datetime.now(pytz.utc)
start_date = now - timedelta(days=self.schedule_lookback_days)
end_date = now + timedelta(days=self.schedule_lookahead_days)
date_str = f"{start_date.strftime('%Y%m%d')}-{end_date.strftime('%Y%m%d')}"
url = f"https://site.api.espn.com/apis/site/v2/sports/{self.sport}/{self.league}/scoreboard"
data = fetch_espn_scoreboard(
self.session,
url,
params={"dates": date_str, "limit": ESPN_MAX_LIMIT},
headers=self.headers,
timeout=10,
logger=self.logger,
)
immediate_events = data.get("events", [])
if immediate_events:
self.logger.info(f"Fetched {len(immediate_events)} events {date_str}")
return {"events": immediate_events}
except requests.exceptions.RequestException as e:
self.logger.warning(
f"Error fetching this weeks games for {self.sport} - {self.league} - {date_str}: {e}"
)
return None
def _custom_scorebug_layout(self, game: dict, draw_overlay: ImageDraw.ImageDraw):
pass
def cleanup(self):
"""Clean up resources when plugin is unloaded."""
# Close HTTP session
if hasattr(self, 'session') and self.session:
try:
self.session.close()
except Exception as e:
self.logger.warning(f"Error closing session: {e}")
# Clear caches
if hasattr(self, '_logo_cache'):
self._logo_cache.clear()
self.logger.info(f"{self.__class__.__name__} cleanup completed")
def _game_divisions(self, game: Dict) -> Optional[set]:
"""Divisions of BOTH sides, or None when they cannot be told.
Both sides are collected, but the caller only needs ONE of them to sit
in a checked division. Requiring every participant read as "FBS games
only" and removed a ranked side hosting an FCS school -- which is still
a game involving a team the viewer checked the box for, and on a real
Week 2 slate it silently dropped five of the twenty ranked matchups.
What the checkbox is for is keeping FCS-versus-FCS out of a board
configured for FBS, and that still holds: a game with no checked
division on either side is dropped.
"""
divisions = self._load_division_team_ids()
if not any(divisions.values()):
return None
try:
ids = [int(game.get("home_id")), int(game.get("away_id"))]
except (TypeError, ValueError):
return None
present = set()
for team_id in ids:
for name in ("fbs", "fcs"):
if team_id in divisions.get(name, set()):
present.add(name)
break
else:
present.add("other")
return present
def _league_has_rankings(self) -> bool:
"""Only college leagues publish a poll; everyone else 404s.
This gate matters more than it looks. _fetch_team_rankings only
short-circuits when the cache is non-empty, so a failed fetch leaves it
empty and the next update tries again -- at a 30s interval that is
~2,900 pointless requests a day, per league, all of them 404s.
"""
league = (self.league or "").lower()
return "college" in league or "ncaa" in league
@staticmethod
def _normalise_divisions(raw) -> List[str]:
"""Division names from config, in the shape the filter expects.
A hand-edited config can hold "fbs" where the schema says ["fbs"], and
list("fbs") is ['f', 'b', 's'] -- three names that match no division, so
every non-favourite game is rejected by a setting the user believes says
the opposite. An empty list is left empty: that means "no division
filter" and is a legitimate choice, not a mistake to correct.
"""
if isinstance(raw, str):
raw = [raw]
try:
items = list(raw or [])
except TypeError:
return []
return [str(d).strip().lower() for d in items if str(d).strip()]
def _round_robin_favorites(self, games: List[Dict], limit: int) -> List[Dict]:
"""Each favourite team's next game before any team's second one.
Taking the soonest N favourite games spends the slots on whoever plays
most often. Walked across a real season with two favourites and a limit
of 2, nine days of it showed Auburn twice and Georgia not at all --
Auburn played either side of a Georgia bye, so both slots went to
Auburn. The other-games pool already refuses to do this; favourites
were still doing it.
Depth is kept where there is room: one favourite with three slots still
gets its next three games, because the round-robin only comes back for
a team's second game once every team has had a first.
A game between two favourites is picked once and counts for both.
"""
if limit <= 0 or not games:
return []
wanted = [t for t in (self.favorite_teams or []) if t]
if len(wanted) < 2:
return games[:limit] # nothing to share the slots between
# Which side of a game belongs to which favourite is a per-lineage
# question: NRL matches on ESPN team IDs because its abbreviations are
# not unique ("NEW" is both Newcastle and New Zealand), while the rest
# match on abbreviation. Ask for the lineage's own matcher rather than
# assuming, or this silently groups nothing and every slot goes empty.
matcher = getattr(self, "_team_in", None)
if not callable(matcher):
matcher = None # an is-None test narrows for static analysis
if matcher is None:
def belongs(game, team):
return team in (game.get("home_abbr"), game.get("away_abbr"))
else:
def belongs(game, team):
return bool(matcher(game.get("home_id"), [team])
or matcher(game.get("away_id"), [team]))
queues = {team: [] for team in wanted}
for game in games: # already in kickoff order
for team in wanted:
if belongs(game, team):
queues[team].append(game)
picked, taken = [], set()
while len(picked) < limit:
progressed = False
for team in wanted:
queue = queues[team]
while queue and queue[0].get("id") in taken:
queue.pop(0)
if queue and len(picked) < limit:
game = queue.pop(0)
taken.add(game.get("id"))
picked.append(game)
progressed = True
if not progressed:
break # every queue is empty
return picked
def _normalise_quality(self, raw) -> str:
"""other_games_min_quality, as one of the values the code implements.
An unusable value used to fall through every branch of
_passes_other_filters and silently mean "any" -- a quality bar the
board believes it has and does not.
"""
value = str(raw or "").strip().lower()
if value in self._QUALITY_CHOICES:
return value
if value == "broadcast":
# Retired in football-scoreboard 3.0.0 and now here. Measured
# against a real Week 1 and Week 2 college slate it passed 174 of
# 175 games: ESPN publishes a broadcaster for nearly everything
# now, ESPN+ included, so the tier read as a quality bar and
# behaved as "any". Boards holding it get the bar they thought
# they were getting.
self.logger.warning(
"%s: other_games_min_quality 'broadcast' has been retired -- "
"it let through nearly every game -- using 'ranked'. Change "
"the setting to clear this.", getattr(self, "sport_key", "?"),
)
return "ranked"
self.logger.warning(
"%s: ignoring unusable other_games_min_quality=%r, using 'ranked'",
getattr(self, "sport_key", "?"), raw,
)
return "ranked"
def _check_ranking_coverage(self, games: List[Dict]) -> None:
"""Say so when a loaded poll matches nothing on the schedule.
The table is keyed by the abbreviation the RANKINGS endpoint returns and
matched against the one the SCOREBOARD endpoint returns. Nothing
guarantees the two agree, and if they ever stop agreeing the filter
quietly removes every non-favourite game -- no exception, no log line,
just a shorter board. That is the same shape as the bug where rankings
were never loading at all, which survived until someone went looking.
Throttled to once an hour: selection runs on every update.
"""
if self.other_games_min_quality != "ranked":
return
rankings = getattr(self, "_team_rankings_cache", None) or {}
if not rankings or not games:
return
if any(self._is_ranked_game(g) for g in games):
return
now = time.monotonic()
# Zero means never logged, not "logged at the epoch". monotonic() counts
# from an arbitrary origin -- on a freshly booted board it is a few
# hundred seconds -- so comparing against 0 swallowed the first warning
# for the first hour of uptime, which is exactly when a misconfigured
# board is being watched. CI caught this; a machine with days of uptime
# cannot.
if (self._ranking_coverage_logged_at
and now - self._ranking_coverage_logged_at < self._RANKING_COVERAGE_SECONDS):
return
self._ranking_coverage_logged_at = now
self.logger.warning(
"%s: %d ranked teams loaded, but none of the %d other games match "
"one -- the quality filter is removing every non-favourite game. "
"Ranked abbreviations look like: %s",
self.league, len(rankings), len(games),
", ".join(sorted(rankings)[:8]),
)
def _favorites_first(
self,
processed_games: List[Dict],
favorite_limit: int,
other_limit: int,
newest_first: bool = False,
) -> List[Dict]:
"""Favourite games first, then a bounded number of everything else.
This is the middle setting the plugin was missing. `show_favorite_teams_only`
used to be the whole story: on, and you saw nothing but your teams; off,
and your teams were ignored entirely -- the selection just took the next
N games league-wide, so a UGA fan with 946 upcoming college games in the
window saw UGA about as often as chance allowed.
Both counts are TOTALS here, not per-team. In favourites-only mode
`upcoming_games_to_show` is a per-team budget, which is reasonable when
the list is your own teams; applied to a dynamic group it is not. With
AP_TOP_10 resolving to a dozen teams, three games each is 28 distinct
cards before a single non-favourite is added. A total keeps the rotation
the length the user asked for.
"""
if newest_first:
def key(g):
return g.get("start_time_utc") or datetime.min.replace(tzinfo=timezone.utc)
ordered = sorted(processed_games, key=key, reverse=True)
else:
def key(g):
return g.get("start_time_utc") or datetime.max.replace(tzinfo=timezone.utc)
ordered = sorted(processed_games, key=key)
favorites, others, unfiltered = [], [], []
for game in ordered:
if self._is_favorite_game(game):
favorites.append(game) # never filtered: your team is your team
continue
unfiltered.append(game)
if self._passes_other_filters(game):
others.append(game)
self._check_ranking_coverage(unfiltered)
self._selection_pools = {
"favorites": favorites,
"others": self._by_importance(others, newest_first),
"unfiltered": self._by_importance(unfiltered, newest_first),
"favorite_limit": favorite_limit,
"other_limit": other_limit,
"newest_first": newest_first,
}
return self._compose_selection()
def _compose_selection(self) -> List[Dict]:
"""Favourites plus the current slice of others, in schedule order.
Split out of _favorites_first so the slice can be re-cut between
fetches. The pools are settled -- which games exist, and which of them
are worth a slot -- while WHICH of the others is on screen is a display
decision, and gating it on the fetch made the rotation interval a lie:
update() returns early until upcoming_update_interval has passed, so a
four-minute rotation actually stepped fifteen windows once an hour.
Same lesson as _advance_live_game_if_due further down this file.
"""
pools = self._selection_pools
favorites, others = pools["favorites"], pools["others"]
favorite_limit, other_limit = pools["favorite_limit"], pools["other_limit"]
newest_first = pools["newest_first"]
if newest_first:
def key(g):
return g.get("start_time_utc") or datetime.min.replace(tzinfo=timezone.utc)
else:
def key(g):
return g.get("start_time_utc") or datetime.max.replace(tzinfo=timezone.utc)
selected = self._round_robin_favorites(favorites, max(0, favorite_limit))
selected.extend(self._other_games_window(others, max(0, other_limit)))
if not selected and other_limit > 0:
# Nothing survived at all: your teams are not playing inside the
# schedule window AND the filters removed every other game. Each
# check fails open on missing data, but a filter working exactly as
# asked can still match nothing on a given day, and with no
# favourite game left there is nothing to carry the mode -- an empty
# list is a blank panel, not a short one. Same whole-list fallback
# `_filtered_or_all` makes for a board with no favourites at all.
# `other_limit` of 0 is an explicit "favourites only", so that one
# is left to go quiet as asked.
selected = self._other_games_window(pools["unfiltered"], max(0, other_limit))
# Re-sort so the card order still reads as a schedule. Selection decides
# WHICH games; it should not reorder them into favourites-then-others,
# which would show next week's UGA game before tonight's.
selected.sort(key=key, reverse=newest_first)
return selected
class SportsLiveSharedMixin:
"""The ``SportsLive`` bodies identical in all eight scoreboards."""
def _detect_stale_games(self, games: List[Dict]) -> None:
"""Remove games that appear stale or haven't updated."""
current_time = time.time()
for game in games[:]: # Copy list to iterate safely
game_id = game.get("id")
if not game_id:
continue
# Check if game data is stale
timestamps = self.game_update_timestamps.get(game_id, {})
last_seen = timestamps.get("last_seen", 0)
if last_seen > 0 and current_time - last_seen > self.stale_game_timeout:
self.logger.warning(
f"Removing stale game {game.get('away_abbr')}@{game.get('home_abbr')} "
f"(last seen {int(current_time - last_seen)}s ago)"
)
games.remove(game)
if game_id in self.game_update_timestamps:
del self.game_update_timestamps[game_id]
continue
# Also check if game appears to be over
if self._is_game_really_over(game):
self.logger.debug(
f"Removing game that appears over: {game.get('away_abbr')}@{game.get('home_abbr')} "
f"(clock={game.get('clock')}, period={game.get('period')}, period_text={game.get('period_text')})"
)
games.remove(game)
if game_id in self.game_update_timestamps:
del self.game_update_timestamps[game_id]
def _idle_live_interval(self) -> int:
"""How long to wait before looking for live games again, when there are none.
Escalates the longer nothing turns up, and any live game resets it, so
an in-season gap between games costs at most one escalated wait while
an out-of-season league stops polling on a live cadence entirely.
Capped rather than unbounded: the cost of backing off is how late the
first game after a quiet spell is noticed, and past the cap the saving
stops being worth that.
The escalation is then clamped by the next kickoff the league already
knows about -- see _clamp_to_scheduled_start. Without that clamp the cap
*is* the miss: a league idle overnight reaches the ceiling, and the
first game of the next day is not noticed for up to that long.
"""
streak = getattr(self, "_empty_live_streak", 0)
base = self.no_data_interval
ceiling = getattr(self, "live_idle_max_interval",
_DEFAULT_LIVE_IDLE_MAX_SECONDS)
# The ceiling bounds the un-escalated interval too. The two settings are
# independent integers with no cross-validation, so base > ceiling is a
# reachable config -- and returning base unclamped there made the wait
# *shrink* as the streak grew (3600s at streak 0, 900s at streak 24),
# the opposite of what the setting named "maximum" promises.
if streak >= _IDLE_LONG_STREAK:
interval = min(int(base * _IDLE_LONG_FACTOR), ceiling)
elif streak >= _IDLE_SHORT_STREAK:
interval = min(int(base * _IDLE_SHORT_FACTOR), ceiling)
else:
interval = min(base, ceiling)
return self._clamp_to_scheduled_start(interval)
def _clamp_to_scheduled_start(self, interval: int) -> int:
"""Shorten an idle wait that would sleep through a known kickoff.
The back-off counts consecutive empty looks and nothing else, so it
cannot tell an out-of-season league from an in-season one a few hours
before kickoff. Both reach the ceiling, and the ceiling then becomes the
blind spot: measured on two rigs on 2026-09-19, gaps of up to 928s
between looks, 10 of them at or above 900s. A game starting inside such
a gap is not noticed until it ends -- which is the "it doesn't pick up
new live games until I restart it" report, restarting being the one
thing that forces an immediate look.
The fix costs no extra request: the live fetch already downloads the
whole day's scoreboard, upcoming games included, and
_note_scheduled_start_candidate keeps the earliest start still ahead of
us out of exactly that payload.
Two cases, either side of the kickoff:
* before it -- wait at most until it starts, never past it;
* just after it -- hold the live cadence for _KICKOFF_GRACE_SECONDS,
because a provider that has not yet flipped the status would
otherwise look like another empty check and escalate the back-off
again, right when the game is actually starting.
"""
now = time.time()
start = _current_scheduled_start(self, now)
if not start:
return interval
live = getattr(self, "update_interval", None) or _KICKOFF_POLL_FLOOR
if now < start:
return max(live, min(interval, int(start - now)))
if now - start <= _KICKOFF_GRACE_SECONDS:
return live
return interval
def _note_scheduled_start_candidate(self, details) -> None:
"""Offer a game from the current look as the next kickoff to wake for.
Called for every event the live fetch returns, live or not, so the
earliest start still ahead of us falls out of the payload the manager
already has. Self-correcting: a stored start that has passed is
replaced by the next one offered, so a postponed game cannot pin the
cadence to a kickoff that never happens.
Every pending kickoff is honoured, not just the first. A kickoff that
arrives while an earlier one is inside its grace is queued in
``_later_scheduled_starts`` (the earliest _KICKOFF_QUEUE_MAX of them)
and takes over when that grace ends, with a grace of its own. Keeping
only the one kickoff dropped the second of two favourites starting
within the grace of each other: it was refused while the first held
the slot, and refused again once it had passed, so if ESPN had not
flipped it live by the end of the first grace the back-off went
straight back to its ceiling and the game was noticed up to that late.
"""
if not isinstance(details, dict):
return
if details.get("is_live") or details.get("is_halftime"):
return
start = details.get("start_time_utc")
timestamp = getattr(start, "timestamp", None)
if timestamp is None:
return
try:
candidate = float(timestamp())
except (TypeError, ValueError, OSError, OverflowError):
return
now = time.time()
if candidate <= now:
return
current = _current_scheduled_start(self, now)
# A kickoff that has only just passed is *kept*, not replaced by the
# next one on the card. Replacing it immediately is what made the grace
# window in _clamp_to_scheduled_start dead code: the moment 13:00 came
# round, the stored start jumped to the 16:05 games, `now < start` went
# true again, and the back-off returned to its ceiling -- at exactly the
# moment the games were starting. Observed live on 2026-09-20: the rig
# polled at 13:00:45, found nothing live because ESPN had not flipped
# the status yet, and then went quiet for the next quarter of an hour,
# which is the behaviour this whole clamp exists to prevent.
#
# Nor is it forgotten: whichever kickoff loses is queued behind the
# one honoured now, so it gets its own grace when that one's ends.
if current is None:
self._next_scheduled_start_ts = candidate
return
if candidate == current:
return
if candidate < current:
self._next_scheduled_start_ts, candidate = candidate, current
queued = getattr(self, "_later_scheduled_starts", None) or []
if candidate not in queued:
self._later_scheduled_starts = sorted(
[*queued, candidate])[:_KICKOFF_QUEUE_MAX]
#: How long a game that finished live is still reported by
#: finished_games_snapshot(): long enough for the recent-games list, which
#: refreshes about hourly, to take it over well before most slates would.
FINISHED_GAME_TTL = 900.0
def _record_finished_game(self, details: Dict) -> None:
"""Remember a game that was live and has just gone final (or looks over).
A finished game leaves ``live_games`` at the next poll, and the recent
list that will show it refreshes about hourly, so in between nothing
holds the game's final score -- and a live Vegas card for it would keep
its last live score. Call this wherever a poll drops a game as final
or over. Only a game this manager had as live is taken; one already
held takes the newer details (a game dropped by an "is it over"
heuristic, then marked final by the feed) but keeps its expiry, so a
feed that lists finals all day cannot keep one here all day.
"""
game_id = details.get("id") if isinstance(details, dict) else None
if not game_id:
return
finished = self.__dict__.setdefault("_finished_games", {})
held = finished.get(game_id)
if held is not None:
finished[game_id] = (held[0], dict(details))
return
if not any(g.get("id") == game_id for g in getattr(self, "live_games", ()) or ()):
return
finished[game_id] = (time.monotonic(), dict(details))
def finished_games_snapshot(self) -> List[Dict]:
"""Games that went final here within FINISHED_GAME_TTL, newest data first.
Copies, safe to decorate. The caller dedupes them against its other
lists (src/common/sports_vegas.dedupe_games keeps the liveliest copy,
and a final beats nothing but a live one).
"""
finished = self.__dict__.get("_finished_games")
if not finished:
return []
now = time.monotonic()
# A copy first: a manager finishing its update in the background (off
# the plugin's lock) may record a game while the ticker reads these.
held = list(finished.items())
for game_id, (seen, _game) in held:
if now - seen > self.FINISHED_GAME_TTL:
finished.pop(game_id, None)
return [dict(game) for _id, (seen, game) in held
if now - seen <= self.FINISHED_GAME_TTL]
def _note_live_fetch(self, found_live: bool) -> None:
"""Record whether a look for live games found any."""
if found_live:
if getattr(self, "_empty_live_streak", 0):
self.logger.info(
"Live games found after %d empty check(s); back to the "
"live update interval", self._empty_live_streak)
self._empty_live_streak = 0
else:
self._empty_live_streak = getattr(self, "_empty_live_streak", 0) + 1
class SportsRecentSharedMixin:
"""The ``SportsRecent`` bodies identical in all eight scoreboards."""
def __init__(
self,
config: Dict[str, Any],
display_manager,
cache_manager,
logger: logging.Logger,
sport_key: str,
):
super().__init__(config, display_manager, cache_manager, logger, sport_key)
self.games_list = [] # Filtered list for display (favorite teams)
self.current_game_index = 0
self.last_update = 0
self.update_interval = self.mode_config.get(
"recent_update_interval", 3600
) # Check for recent games every hour
self.last_game_switch = 0
self.game_display_duration = self.mode_config.get("recent_game_duration", 15)
self._zero_clock_timestamps: Dict[str, float] = {} # Track games at 0:00
def _get_zero_clock_duration(self, game_id: str) -> float:
"""Track how long a game has been at 0:00 clock."""
current_time = time.time()
if game_id not in self._zero_clock_timestamps:
self._zero_clock_timestamps[game_id] = current_time
return 0.0
return current_time - self._zero_clock_timestamps[game_id]
def _clear_zero_clock_tracking(self, game_id: str) -> None:
"""Clear tracking when game clock moves away from 0:00 or game ends."""
if game_id in self._zero_clock_timestamps:
del self._zero_clock_timestamps[game_id]
Reference in New Issue View Git Blame Copy Permalink