mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-18 12:59:12 +00:00
Compare commits
6
Commits
v3.3.0
..
9e1c339273
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9e1c339273 | ||
|
|
c2ace90353 | ||
|
|
cc6a75712c | ||
|
|
ed2b81a4fb | ||
|
|
ac44b5a6f5 | ||
|
|
df8f69af30 |
@@ -17,81 +17,6 @@ release that ships it.
|
|||||||
accepts both, but the store flags the old spelling as deprecated
|
accepts both, but the store flags the old spelling as deprecated
|
||||||
(`store_manager.py`) and only the new one is in `schema/manifest_schema.json`.
|
(`store_manager.py`) and only the new one is in `schema/manifest_schema.json`.
|
||||||
|
|
||||||
## 3.3.0
|
|
||||||
|
|
||||||
**The release the sports scoreboards floor on to delete their bundled copies.**
|
|
||||||
3.2.0 shipped the unified sports library and made `ledmatrix_min_version`
|
|
||||||
enforceable; this ships the last three shared modules and completes the store
|
|
||||||
gate, so a scoreboard can now floor here and carry no fallback at all.
|
|
||||||
|
|
||||||
New modules a plugin may import via `src.*` and floor on 3.3.0 for:
|
|
||||||
|
|
||||||
- `src/common/sports_card.py` — settings, colour, font and date helpers for a
|
|
||||||
scoreboard's `game_renderer.py`. Free functions taking `config`/`fonts`
|
|
||||||
explicitly, so nothing about the caller's class is assumed.
|
|
||||||
- `src/common/sports_game_renderer.py` — `SportsGameRendererMixin`: scroll/Vegas
|
|
||||||
card geometry (centre gap, logo slot and cache key, layout offsets, the
|
|
||||||
upcoming-card date and time layout). No `__init__` and no state, so adoption
|
|
||||||
is one line on the class statement.
|
|
||||||
- `src/common/sports_shared.py` — `SportsCoreSharedMixin`,
|
|
||||||
`SportsLiveSharedMixin`, `SportsRecentSharedMixin`: the `sports.py` bodies
|
|
||||||
byte-identical in all eight lineage-sharing scoreboards.
|
|
||||||
|
|
||||||
All three sit under `src/common/` rather than `src/base_classes/sports/`,
|
|
||||||
deliberately: importing that package pulls `core.py` → `DisplayManager` →
|
|
||||||
`rgbmatrix`, and these are pure logic. Plugins importing them must not acquire a
|
|
||||||
hardware dependency.
|
|
||||||
|
|
||||||
Three notes for anyone adopting `sports_shared`:
|
|
||||||
|
|
||||||
- `SportsRecentSharedMixin` defines `__init__`. Its bare `super()` binds to the
|
|
||||||
mixin, so it reaches the host only when the mixin is listed **first** in the
|
|
||||||
bases. Reversing that order silently skips the host constructor.
|
|
||||||
- Methods that resolve the plugin's `config_schema.json` use `_plugin_dir()`,
|
|
||||||
which walks the MRO rather than reading `__file__` — `__file__` is now
|
|
||||||
`src/common/`. It walks because `SportsCore` is an ABC: a subclass built with
|
|
||||||
`type(name, bases, ns)` reports `__module__` as `"abc"`.
|
|
||||||
- `_get_timezone`, `_extract_game_details` and `_fetch_data` are byte-identical
|
|
||||||
across the eight but stay in the plugins. The first binds a per-plugin
|
|
||||||
timezone module whose contents differ; the other two are the abstract stubs
|
|
||||||
that define the sport.
|
|
||||||
|
|
||||||
**The store's compatibility gate is now on every registry-managed route.** 3.2.0
|
|
||||||
gated `install_plugin`. This release gates the git-pull update path and
|
|
||||||
`install_from_url`, so a plugin whose floor the core cannot meet is refused
|
|
||||||
after download with no partial directory left behind.
|
|
||||||
|
|
||||||
### Fixed
|
|
||||||
|
|
||||||
- **ESPN 403s.** `site.api` began rejecting the User-Agent strings this repo
|
|
||||||
sent on 2026-08-04; every shared-data-source scoreboard returned
|
|
||||||
`403 Forbidden`. Requests now send an identifying token with a project URL —
|
|
||||||
browser-style strings and bare custom tokens are both refused.
|
|
||||||
- **Low-memory boards becoming unreachable under load** while still answering
|
|
||||||
pings and serving the web UI. Fetched payloads are released after delivery
|
|
||||||
rather than pinned on the completed request for up to an hour, malloc arenas
|
|
||||||
are capped, and log volume and SD writes are reduced. Available memory is now
|
|
||||||
reported in Tools diagnostics.
|
|
||||||
- **A failed logo download pinning a team to a grey box**: the placeholder was
|
|
||||||
written under the real logo's filename, so later attempts found a file and
|
|
||||||
reported success without retrying.
|
|
||||||
- **A plugin enabled but never loaded is retried** rather than staying absent
|
|
||||||
with `error = null`.
|
|
||||||
- `ttl` now controls cache expiry; abandoned cache writes no longer leave temp
|
|
||||||
files; one cache-cleanup thread per directory rather than per manager.
|
|
||||||
- Odds are fetched for the games displayed, not the whole schedule window, and a
|
|
||||||
stalled ESPN no longer stalls the whole plugin update.
|
|
||||||
- Array-item secrets are no longer wiped or logged.
|
|
||||||
- A restored backup matches the device it was taken from.
|
|
||||||
- The web UI reports the real error instead of "unknown", rejects non-finite
|
|
||||||
JSON numbers, and stops checkbox groups posting back hidden options.
|
|
||||||
|
|
||||||
### Added
|
|
||||||
|
|
||||||
- `display.hardware.orientation` for panels mounted upside down.
|
|
||||||
- Vegas keeps live content in the ticker rather than being preempted by it.
|
|
||||||
- Schemas can label enum dropdown options.
|
|
||||||
|
|
||||||
## 3.2.0
|
## 3.2.0
|
||||||
|
|
||||||
**The first release shipping the unified sports library.** This is the version
|
**The first release shipping the unified sports library.** This is the version
|
||||||
|
|||||||
+1
-1
@@ -4,5 +4,5 @@ LEDMatrix Display System
|
|||||||
Core source package for the LED Matrix Display project.
|
Core source package for the LED Matrix Display project.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
__version__ = "3.3.0"
|
__version__ = "3.2.0"
|
||||||
|
|
||||||
|
|||||||
@@ -143,10 +143,8 @@ class LogoHelper:
|
|||||||
"""
|
"""
|
||||||
logo_path = Path(logo_path)
|
logo_path = Path(logo_path)
|
||||||
|
|
||||||
# Try to load existing logo first. A placeholder written by a previous
|
# Try to load existing logo first
|
||||||
# failed download does not count: it wears the real logo's filename, so
|
if logo_path.exists():
|
||||||
# trusting the file's existence is what left teams as grey boxes.
|
|
||||||
if logo_path.exists() and not self._is_stale_placeholder(logo_path):
|
|
||||||
return self.load_logo(team_abbr, logo_path, max_width, max_height)
|
return self.load_logo(team_abbr, logo_path, max_width, max_height)
|
||||||
|
|
||||||
# Download if URL provided and file doesn't exist
|
# Download if URL provided and file doesn't exist
|
||||||
@@ -154,61 +152,13 @@ class LogoHelper:
|
|||||||
try:
|
try:
|
||||||
self.logger.info(f"Downloading logo for {team_abbr} from {logo_url}")
|
self.logger.info(f"Downloading logo for {team_abbr} from {logo_url}")
|
||||||
self._download_logo(logo_url, logo_path)
|
self._download_logo(logo_url, logo_path)
|
||||||
# The file on disk just changed. Any cached image for it is the
|
|
||||||
# placeholder we came here to replace, and load_logo() answers
|
|
||||||
# from the cache before touching the disk -- so without this the
|
|
||||||
# real logo would not appear until the process restarted.
|
|
||||||
self._invalidate_cached_logo(team_abbr, logo_path)
|
|
||||||
return self.load_logo(team_abbr, logo_path, max_width, max_height)
|
return self.load_logo(team_abbr, logo_path, max_width, max_height)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
self.logger.error(f"Failed to download logo for {team_abbr}: {e}")
|
self.logger.error(f"Failed to download logo for {team_abbr}: {e}")
|
||||||
# The retry failed, so restart the back-off. The stale
|
|
||||||
# placeholder is still on disk with its old timestamp, and
|
|
||||||
# leaving it there means the next call retries immediately --
|
|
||||||
# a download attempt per call, which is what the back-off
|
|
||||||
# exists to prevent.
|
|
||||||
self._refresh_stale_placeholder(logo_path)
|
|
||||||
|
|
||||||
# Create placeholder if all else fails
|
# Create placeholder if all else fails
|
||||||
return self._create_placeholder_logo(team_abbr, max_width, max_height)
|
return self._create_placeholder_logo(team_abbr, max_width, max_height)
|
||||||
|
|
||||||
def _invalidate_cached_logo(self, team_abbr: str, logo_path: Path) -> None:
|
|
||||||
"""Drop every cached size of one logo after its file changed on disk."""
|
|
||||||
prefix = f"{team_abbr}_{logo_path}_"
|
|
||||||
for key in [k for k in self._logo_cache if k.startswith(prefix)]:
|
|
||||||
self._logo_cache.pop(key, None)
|
|
||||||
if key in self._cache_order:
|
|
||||||
self._cache_order.remove(key)
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def _refresh_stale_placeholder(logo_path: Path) -> None:
|
|
||||||
"""Restart the retry back-off after a failed download attempt."""
|
|
||||||
try:
|
|
||||||
from src.logo_downloader import refresh_placeholder_timestamp
|
|
||||||
except ImportError:
|
|
||||||
return
|
|
||||||
refresh_placeholder_timestamp(logo_path)
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def _is_stale_placeholder(logo_path: Path) -> bool:
|
|
||||||
"""True if the file is a placeholder old enough to be worth retrying.
|
|
||||||
|
|
||||||
Imported lazily so this module keeps working against a core build whose
|
|
||||||
logo_downloader predates placeholder marking.
|
|
||||||
"""
|
|
||||||
try:
|
|
||||||
from src.logo_downloader import (
|
|
||||||
PLACEHOLDER_RETRY_SECONDS,
|
|
||||||
is_placeholder_logo,
|
|
||||||
placeholder_age_seconds,
|
|
||||||
)
|
|
||||||
except ImportError:
|
|
||||||
return False
|
|
||||||
if not is_placeholder_logo(logo_path):
|
|
||||||
return False
|
|
||||||
age = placeholder_age_seconds(logo_path)
|
|
||||||
return age is None or age >= PLACEHOLDER_RETRY_SECONDS
|
|
||||||
|
|
||||||
def get_logo_variations(self, team_abbr: str) -> List[str]:
|
def get_logo_variations(self, team_abbr: str) -> List[str]:
|
||||||
"""
|
"""
|
||||||
Get possible filename variations for a team abbreviation.
|
Get possible filename variations for a team abbreviation.
|
||||||
|
|||||||
@@ -1,460 +0,0 @@
|
|||||||
"""Card-drawing helpers shared by every sports scoreboard plugin.
|
|
||||||
|
|
||||||
The eight scoreboards each carried byte-identical copies of the functions
|
|
||||||
below: the colour pickers, the settings lookup, the date and time formatting,
|
|
||||||
the favourite-team rules and the font-size grid snapping. One fix had to be
|
|
||||||
made eight times, and a new scoreboard started by copying them a ninth.
|
|
||||||
|
|
||||||
Everything here is a **free function taking explicit arguments**, not a base
|
|
||||||
class. Adoption is therefore per-function and reversible: a plugin keeps its
|
|
||||||
method and delegates the body, so the call sites and the override points are
|
|
||||||
untouched. That is also why `config`, `logger` and `fonts` are parameters
|
|
||||||
rather than attributes -- the helper never reaches back into the caller.
|
|
||||||
|
|
||||||
The bodies are the plugins' own code, moved rather than rewritten. The one
|
|
||||||
deliberate difference is `crisp_size`, which takes the seven-plugin guard
|
|
||||||
(`not desired`) instead of football's: they agree on every real input, and
|
|
||||||
the extra guard only stops a None size raising TypeError.
|
|
||||||
"""
|
|
||||||
|
|
||||||
from datetime import datetime, timezone
|
|
||||||
from typing import Any, Dict, Optional, Tuple
|
|
||||||
from zoneinfo import ZoneInfo
|
|
||||||
|
|
||||||
__all__ = [
|
|
||||||
"ELEMENT_FOR_FONT", "FAVORITE_RESULT_COLOR_DEFAULTS", "FONT_NAME_ALIASES",
|
|
||||||
"FONT_PIXEL_GRID", "MONTH_ABBR", "WEEKDAY_ABBR",
|
|
||||||
"scroll_card_option", "element_color", "font_color", "coerce_rgb",
|
|
||||||
"score_color_for", "recent_score_color", "favorite_teams_for",
|
|
||||||
"side_is_favorite", "side_score", "favorite_result",
|
|
||||||
"card_tzinfo", "weekday_for", "format_game_date", "format_game_time",
|
|
||||||
"vs_text", "upcoming_center_mode", "crisp_size", "schema_font_size",
|
|
||||||
"resolve_font_size", "unshare_element_fonts",
|
|
||||||
]
|
|
||||||
|
|
||||||
#: Which customization element owns each font key, for colour resolution.
|
|
||||||
ELEMENT_FOR_FONT: Dict[str, str] = {
|
|
||||||
"score": "score_text",
|
|
||||||
"time": "period_text",
|
|
||||||
"team": "team_name",
|
|
||||||
"status": "status_text",
|
|
||||||
"detail": "detail_text",
|
|
||||||
"rank": "rank_text",
|
|
||||||
}
|
|
||||||
|
|
||||||
#: Fallback colours when favourite_result_colors is on but a slot is unset.
|
|
||||||
FAVORITE_RESULT_COLOR_DEFAULTS: Dict[str, Tuple[int, int, int]] = {
|
|
||||||
"win": (0, 255, 0),
|
|
||||||
"loss": (255, 0, 0),
|
|
||||||
"tie": (255, 200, 0),
|
|
||||||
}
|
|
||||||
|
|
||||||
#: Family aliases the web UI may write, mapped to the shipped filename.
|
|
||||||
FONT_NAME_ALIASES: Dict[str, str] = {
|
|
||||||
"press_start": "PressStart2P-Regular.ttf",
|
|
||||||
"four_by_six": "4x6-font.ttf",
|
|
||||||
}
|
|
||||||
|
|
||||||
#: Pixel grid each face renders crisply on. Off-grid sizes anti-alias, which
|
|
||||||
#: on an LED matrix is a dim lamp rather than a soft edge.
|
|
||||||
FONT_PIXEL_GRID: Dict[str, int] = {
|
|
||||||
"PressStart2P-Regular.ttf": 8,
|
|
||||||
"4x6-font.ttf": 7,
|
|
||||||
}
|
|
||||||
|
|
||||||
MONTH_ABBR = ("Jan", "Feb", "Mar", "Apr", "May", "Jun",
|
|
||||||
"Jul", "Aug", "Sep", "Oct", "Nov", "Dec")
|
|
||||||
WEEKDAY_ABBR = ("Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun")
|
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
|
||||||
# Settings lookup
|
|
||||||
# ---------------------------------------------------------------------------
|
|
||||||
|
|
||||||
def scroll_card_option(config: Optional[Dict[str, Any]], key: str,
|
|
||||||
default: Any = None) -> Any:
|
|
||||||
"""Read one key from the scroll_card config block."""
|
|
||||||
block = (config or {}).get("scroll_card")
|
|
||||||
if isinstance(block, dict) and block.get(key) is not None:
|
|
||||||
return block.get(key)
|
|
||||||
return default
|
|
||||||
|
|
||||||
|
|
||||||
def vs_text(config: Optional[Dict[str, Any]]) -> str:
|
|
||||||
"""Separator drawn between the teams -- "VS", "@", "at", anything."""
|
|
||||||
return str(scroll_card_option(config, "vs_text", "VS"))
|
|
||||||
|
|
||||||
|
|
||||||
def upcoming_center_mode(config: Optional[Dict[str, Any]]) -> str:
|
|
||||||
"""Middle of an upcoming card: 'vs', 'date_time' or 'none'."""
|
|
||||||
mode = str(scroll_card_option(config, "upcoming_center", "vs") or "vs").lower()
|
|
||||||
return mode if mode in ("vs", "date_time", "none") else "vs"
|
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
|
||||||
# Colour
|
|
||||||
# ---------------------------------------------------------------------------
|
|
||||||
|
|
||||||
def element_color(config: Optional[Dict[str, Any]], element: str,
|
|
||||||
default: Tuple[int, int, int] = (255, 255, 255)):
|
|
||||||
"""Per-element text colour from customization.<element>.text_color."""
|
|
||||||
try:
|
|
||||||
cfg = (config or {}).get("customization", {}).get(element, {})
|
|
||||||
value = cfg.get("text_color")
|
|
||||||
if isinstance(value, (list, tuple)) and len(value) == 3:
|
|
||||||
return tuple(max(0, min(255, int(c))) for c in value)
|
|
||||||
if isinstance(value, str) and value.startswith("#") and len(value) == 7:
|
|
||||||
return tuple(int(value[i:i + 2], 16) for i in (1, 3, 5))
|
|
||||||
except (TypeError, ValueError):
|
|
||||||
pass
|
|
||||||
return default
|
|
||||||
|
|
||||||
|
|
||||||
def font_color(config: Optional[Dict[str, Any]], fonts: Optional[Dict[str, Any]],
|
|
||||||
font, default: Tuple[int, int, int] = (255, 255, 255)):
|
|
||||||
"""Colour for whichever element owns this face.
|
|
||||||
|
|
||||||
Matched on identity, and deliberately gives up when one object is
|
|
||||||
shared: the last-resort font path can hand the same face to several
|
|
||||||
keys, and there is no right answer for which element's colour that is.
|
|
||||||
White is what those draws used before, so ambiguity costs nothing.
|
|
||||||
"""
|
|
||||||
try:
|
|
||||||
fonts = fonts or {}
|
|
||||||
matches = [element for key, element in ELEMENT_FOR_FONT.items()
|
|
||||||
if fonts.get(key) is font]
|
|
||||||
if len(matches) == 1:
|
|
||||||
return element_color(config, matches[0], default)
|
|
||||||
except (AttributeError, TypeError):
|
|
||||||
pass
|
|
||||||
return default
|
|
||||||
|
|
||||||
|
|
||||||
def coerce_rgb(value, fallback):
|
|
||||||
"""Turn a configured [R, G, B] list into a clamped (r, g, b) tuple."""
|
|
||||||
# Checked before unpacking: a 3-character string ("123") would otherwise
|
|
||||||
# iterate into three digits and yield a colour rather than the fallback.
|
|
||||||
if not isinstance(value, (list, tuple)) or len(value) != 3:
|
|
||||||
return fallback
|
|
||||||
try:
|
|
||||||
r, g, b = (max(0, min(255, int(channel))) for channel in value)
|
|
||||||
except (TypeError, ValueError):
|
|
||||||
return fallback
|
|
||||||
return (r, g, b)
|
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
|
||||||
# Favourite teams
|
|
||||||
# ---------------------------------------------------------------------------
|
|
||||||
|
|
||||||
def favorite_teams_for(config: Dict[str, Any], game: Dict[str, Any]) -> list:
|
|
||||||
"""Favorite teams that apply to this game.
|
|
||||||
|
|
||||||
Both sources are used. Games carry the league manager's *resolved*
|
|
||||||
favorites, which is the only place dynamic groups such as AP_TOP_25
|
|
||||||
appear expanded; the config is read as well so an edit takes effect on
|
|
||||||
already-fetched games, and so hand-built game dicts (tests, other
|
|
||||||
callers) still work.
|
|
||||||
"""
|
|
||||||
favorites = list(game.get("favorite_teams") or [])
|
|
||||||
league_config = config.get(str(game.get("league", "") or ""))
|
|
||||||
if isinstance(league_config, dict):
|
|
||||||
favorites += list(league_config.get("favorite_teams") or [])
|
|
||||||
else:
|
|
||||||
favorites += list(config.get("favorite_teams") or [])
|
|
||||||
return favorites
|
|
||||||
|
|
||||||
|
|
||||||
def side_is_favorite(game: Dict[str, Any], side: str, favorites: set) -> bool:
|
|
||||||
"""Is the home/away side of this game a favorite team?
|
|
||||||
|
|
||||||
Reads both the flat (``home_abbr``) and nested (``home_team.abbrev``)
|
|
||||||
payload shapes, and matches on the ESPN id too, because a couple of
|
|
||||||
leagues (NRL) key favorites by id where abbreviations collide.
|
|
||||||
"""
|
|
||||||
candidates = [game.get(f"{side}_abbr"), game.get(f"{side}_id")]
|
|
||||||
team = game.get(f"{side}_team")
|
|
||||||
if isinstance(team, dict):
|
|
||||||
candidates += [team.get("abbrev"), team.get("abbreviation"), team.get("id")]
|
|
||||||
for value in candidates:
|
|
||||||
if value is not None and str(value).strip().upper() in favorites:
|
|
||||||
return True
|
|
||||||
return False
|
|
||||||
|
|
||||||
|
|
||||||
def side_score(game: Dict[str, Any], side: str) -> Optional[int]:
|
|
||||||
"""Numeric score for one side, from either payload shape."""
|
|
||||||
raw = None
|
|
||||||
team = game.get(f"{side}_team")
|
|
||||||
if isinstance(team, dict) and team.get("score") is not None:
|
|
||||||
raw = team.get("score")
|
|
||||||
if raw is None:
|
|
||||||
raw = game.get(f"{side}_score")
|
|
||||||
try:
|
|
||||||
return int(float(str(raw).strip()))
|
|
||||||
except (TypeError, ValueError):
|
|
||||||
return None
|
|
||||||
|
|
||||||
|
|
||||||
def favorite_result(config: Dict[str, Any], game: Dict[str, Any]) -> 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 = {
|
|
||||||
str(team).strip().upper()
|
|
||||||
for team in favorite_teams_for(config, game)
|
|
||||||
if str(team).strip()
|
|
||||||
}
|
|
||||||
if not favorites:
|
|
||||||
return None
|
|
||||||
|
|
||||||
home_fav = side_is_favorite(game, "home", favorites)
|
|
||||||
away_fav = side_is_favorite(game, "away", favorites)
|
|
||||||
if home_fav == away_fav:
|
|
||||||
return None
|
|
||||||
|
|
||||||
home_score = side_score(game, "home")
|
|
||||||
away_score = side_score(game, "away")
|
|
||||||
if home_score is None or away_score is None:
|
|
||||||
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(config: Dict[str, Any], logger, game: Dict[str, Any], default):
|
|
||||||
"""Fill color for a finished game's score, per favorite_result_colors."""
|
|
||||||
try:
|
|
||||||
settings = (config.get("customization") or {}).get(
|
|
||||||
"favorite_result_colors"
|
|
||||||
) or {}
|
|
||||||
if not settings.get("enabled", False):
|
|
||||||
return default
|
|
||||||
result = favorite_result(config, game)
|
|
||||||
if result is None:
|
|
||||||
return default
|
|
||||||
return coerce_rgb(
|
|
||||||
settings.get(f"{result}_color"),
|
|
||||||
FAVORITE_RESULT_COLOR_DEFAULTS[result],
|
|
||||||
)
|
|
||||||
except Exception:
|
|
||||||
logger.debug("Could not resolve favorite result color", exc_info=True)
|
|
||||||
return default
|
|
||||||
|
|
||||||
|
|
||||||
def score_color_for(config: Dict[str, Any], logger, game: Dict[str, Any],
|
|
||||||
game_type: str, default=None):
|
|
||||||
"""Fill color for a game card's score. Only finished games are tinted.
|
|
||||||
|
|
||||||
The default is the configured score colour rather than a flat white,
|
|
||||||
so customization.score_text.text_color shows on games the favourite
|
|
||||||
tint does not apply to. The tint still wins where it applies.
|
|
||||||
"""
|
|
||||||
if default is None:
|
|
||||||
default = element_color(config, 'score_text')
|
|
||||||
if game_type != "recent":
|
|
||||||
return default
|
|
||||||
return recent_score_color(config, logger, game, default)
|
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
|
||||||
# Date and time
|
|
||||||
# ---------------------------------------------------------------------------
|
|
||||||
|
|
||||||
def card_tzinfo(config: Optional[Dict[str, Any]], logger):
|
|
||||||
"""Timezone for weekday/24h conversions; falls back to UTC."""
|
|
||||||
configured = (config or {}).get("timezone")
|
|
||||||
if configured:
|
|
||||||
try:
|
|
||||||
return ZoneInfo(configured)
|
|
||||||
except (KeyError, ValueError, TypeError, OSError) as exc:
|
|
||||||
# KeyError covers ZoneInfoNotFoundError. A bad zone name in
|
|
||||||
# config should fall back to UTC, not blank the card.
|
|
||||||
logger.debug("Unusable timezone %r: %s", configured, exc)
|
|
||||||
return timezone.utc
|
|
||||||
|
|
||||||
|
|
||||||
def weekday_for(config: Optional[Dict[str, Any]], logger,
|
|
||||||
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 WEEKDAY_ABBR[start.astimezone(card_tzinfo(config, logger)).weekday()]
|
|
||||||
except (ValueError, TypeError):
|
|
||||||
return ""
|
|
||||||
|
|
||||||
|
|
||||||
def format_game_date(config: Optional[Dict[str, Any]], logger, date_text: str,
|
|
||||||
game: Optional[Dict] = None) -> str:
|
|
||||||
"""Format an upcoming card's date per scroll_card.date_format."""
|
|
||||||
raw = str(date_text or "").strip()
|
|
||||||
if not raw:
|
|
||||||
return ""
|
|
||||||
fmt = str(scroll_card_option(config, "date_format", "abbrev") or "abbrev")
|
|
||||||
if fmt == "numeric":
|
|
||||||
return raw
|
|
||||||
parts = raw.replace("-", "/").split("/")
|
|
||||||
if not (len(parts) >= 2 and parts[0].strip().isdigit() and parts[1].strip().isdigit()):
|
|
||||||
return raw
|
|
||||||
month, day = int(parts[0]), int(parts[1])
|
|
||||||
if not 1 <= month <= 12:
|
|
||||||
return raw
|
|
||||||
name = MONTH_ABBR[month - 1]
|
|
||||||
if fmt == "numeric_day_first":
|
|
||||||
return f"{day}/{month}"
|
|
||||||
if fmt == "day_first":
|
|
||||||
return f"{day} {name}"
|
|
||||||
if fmt == "weekday":
|
|
||||||
weekday = weekday_for(config, logger, game)
|
|
||||||
return f"{weekday} {name} {day}" if weekday else f"{name} {day}"
|
|
||||||
return f"{name} {day}"
|
|
||||||
|
|
||||||
|
|
||||||
def format_game_time(config: Optional[Dict[str, Any]], time_text: str) -> str:
|
|
||||||
"""Return the time as-is (12h) or converted to 24h."""
|
|
||||||
raw = str(time_text or "").strip()
|
|
||||||
if not raw or str(scroll_card_option(config, "time_format", "12h")) != "24h":
|
|
||||||
return raw
|
|
||||||
cleaned = raw.upper().replace(" ", "")
|
|
||||||
meridiem = "AM" if cleaned.endswith("AM") else "PM" if cleaned.endswith("PM") else ""
|
|
||||||
if not meridiem:
|
|
||||||
return raw
|
|
||||||
try:
|
|
||||||
hh, _, mm = cleaned[:-2].partition(":")
|
|
||||||
hour, minute = int(hh), int(mm or 0)
|
|
||||||
except ValueError:
|
|
||||||
return raw
|
|
||||||
if not (0 <= hour <= 12 and 0 <= minute <= 59):
|
|
||||||
return raw
|
|
||||||
hour = hour % 12 + (12 if meridiem == "PM" else 0)
|
|
||||||
return f"{hour:02d}:{minute:02d}"
|
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
|
||||||
# Font sizing
|
|
||||||
# ---------------------------------------------------------------------------
|
|
||||||
|
|
||||||
#: Per-schema caches, keyed by the schema's absolute path. Keyed rather than
|
|
||||||
#: global because each plugin declares its own defaults; keyed rather than
|
|
||||||
#: per-class because the helper has no class to hang it on.
|
|
||||||
_SCHEMA_FONT_SIZE_CACHE: Dict[str, Dict[str, int]] = {}
|
|
||||||
|
|
||||||
|
|
||||||
def crisp_size(font_file, desired, aliases=None, grid_table=None):
|
|
||||||
"""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.
|
|
||||||
|
|
||||||
``aliases`` and ``grid_table`` default to the shared tables; a plugin
|
|
||||||
that ships an extra face can pass its own without forking this.
|
|
||||||
"""
|
|
||||||
aliases = FONT_NAME_ALIASES if aliases is None else aliases
|
|
||||||
grid_table = FONT_PIXEL_GRID if grid_table is None else grid_table
|
|
||||||
font_file = aliases.get(font_file, font_file)
|
|
||||||
grid = grid_table.get(font_file)
|
|
||||||
if not grid or not desired or desired <= 0:
|
|
||||||
return desired
|
|
||||||
return max(grid, int(round(float(desired) / grid)) * grid)
|
|
||||||
|
|
||||||
|
|
||||||
def schema_font_size(schema_path: str, element_key) -> Optional[int]:
|
|
||||||
"""The font_size this plugin's config_schema.json declares, or None.
|
|
||||||
|
|
||||||
Cached per schema path. The plugins cached this on their own class; the
|
|
||||||
path is the same distinction expressed without one, so two plugins never
|
|
||||||
share an entry.
|
|
||||||
"""
|
|
||||||
if not element_key:
|
|
||||||
return None
|
|
||||||
cache = _SCHEMA_FONT_SIZE_CACHE.get(schema_path)
|
|
||||||
if cache is None:
|
|
||||||
cache = {}
|
|
||||||
try:
|
|
||||||
import json
|
|
||||||
with open(schema_path) as fh:
|
|
||||||
schema = json.load(fh)
|
|
||||||
props = (schema.get('properties', {})
|
|
||||||
.get('customization', {})
|
|
||||||
.get('properties', {}))
|
|
||||||
for key, spec in props.items():
|
|
||||||
size = spec.get('properties', {}).get('font_size', {}).get('default')
|
|
||||||
if size is not None:
|
|
||||||
cache[key] = int(size)
|
|
||||||
except Exception:
|
|
||||||
cache = {}
|
|
||||||
_SCHEMA_FONT_SIZE_CACHE[schema_path] = cache
|
|
||||||
return cache.get(element_key)
|
|
||||||
|
|
||||||
|
|
||||||
def resolve_font_size(schema_path: str, element_config, element_key,
|
|
||||||
default_size, font_name, aliases=None, grid_table=None):
|
|
||||||
"""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 != schema_font_size(schema_path, element_key):
|
|
||||||
return configured
|
|
||||||
except (TypeError, ValueError):
|
|
||||||
pass
|
|
||||||
return crisp_size(font_name, default_size, aliases, grid_table)
|
|
||||||
|
|
||||||
|
|
||||||
def unshare_element_fonts(logger, 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.
|
|
||||||
"""
|
|
||||||
try:
|
|
||||||
from PIL import ImageFont as _IF
|
|
||||||
except ImportError: # pragma: no cover
|
|
||||||
return fonts
|
|
||||||
seen = {}
|
|
||||||
for key in ELEMENT_FOR_FONT:
|
|
||||||
font = fonts.get(key)
|
|
||||||
if font is None:
|
|
||||||
continue
|
|
||||||
if id(font) not in seen:
|
|
||||||
seen[id(font)] = key
|
|
||||||
continue
|
|
||||||
path, size = getattr(font, "path", None), getattr(font, "size", None)
|
|
||||||
if not path or not size:
|
|
||||||
continue
|
|
||||||
try:
|
|
||||||
fonts[key] = _IF.truetype(path, size)
|
|
||||||
except (OSError, ValueError, TypeError):
|
|
||||||
logger.debug(
|
|
||||||
"Could not un-share the %s face; it keeps the default colour", key)
|
|
||||||
return fonts
|
|
||||||
@@ -1,261 +0,0 @@
|
|||||||
"""The scroll/Vegas card geometry the sports scoreboards all share.
|
|
||||||
|
|
||||||
Eight scoreboards -- afl, baseball, basketball, football, hockey, lacrosse,
|
|
||||||
nrl and soccer -- each carried their own ``game_renderer.py``. After the card
|
|
||||||
helpers moved to ``src/common/sports_card.py`` the settings lookups were
|
|
||||||
shared, but the *geometry* was not: nine methods that decide how wide the
|
|
||||||
centre gap is, how much room a logo gets, and where an upcoming card's date
|
|
||||||
and time land were still eight separate copies. Five were byte-identical
|
|
||||||
across all eight plugins; the other four were identical in seven, with a
|
|
||||||
different single outlier each time.
|
|
||||||
|
|
||||||
That last detail is why this is a mixin rather than free functions. There is
|
|
||||||
no per-sport branching to write -- baseball needs its own
|
|
||||||
``_draw_upcoming_game_status`` and ``_logo_slot_width``, hockey its own
|
|
||||||
``_upcoming_date_and_time``, football its own ``_score_reserve_width``, and
|
|
||||||
every one of those is an ordinary override. The seven that agree inherit and
|
|
||||||
say nothing.
|
|
||||||
|
|
||||||
It is deliberately *only* a mixin: no ``__init__``, no state of its own. The
|
|
||||||
plugins' constructors differ in six ways and none of that difference is worth
|
|
||||||
unifying, so adoption is one line on the class statement plus deleting the
|
|
||||||
methods that now come from here.
|
|
||||||
|
|
||||||
What a host class must provide
|
|
||||||
------------------------------
|
|
||||||
Attributes: ``display_width``, ``display_height``, ``config``, ``fonts``,
|
|
||||||
``logger``, ``_team_rankings_cache``.
|
|
||||||
|
|
||||||
Methods: ``_draw_text_with_outline(draw, text, position, font, fill=None,
|
|
||||||
outline_color=(0, 0, 0))`` -- the one hook whose body genuinely varies -- plus
|
|
||||||
the ``sports_card`` delegations ``_scroll_card_option``,
|
|
||||||
``_upcoming_center_mode``, ``_vs_text``, ``_element_color``,
|
|
||||||
``_format_game_date`` and ``_format_game_time``.
|
|
||||||
"""
|
|
||||||
|
|
||||||
import math
|
|
||||||
from typing import ClassVar, Dict, Tuple
|
|
||||||
|
|
||||||
from PIL import Image, ImageDraw
|
|
||||||
|
|
||||||
|
|
||||||
class SportsGameRendererMixin:
|
|
||||||
"""Shared card geometry for the sports scoreboards. See module docstring."""
|
|
||||||
|
|
||||||
#: Centre strip as a fraction of card width, before clamping.
|
|
||||||
CENTER_GAP_RATIO: ClassVar[float] = 0.28
|
|
||||||
#: Clamp floor for the derived centre gap.
|
|
||||||
CENTER_GAP_MIN_PX: ClassVar[int] = 22
|
|
||||||
#: Clamp ceiling for the derived centre gap.
|
|
||||||
CENTER_GAP_MAX_PX: ClassVar[int] = 40
|
|
||||||
#: Breathing room kept between the score and each logo.
|
|
||||||
_SCORE_LOGO_GUTTER_PX: ClassVar[int] = 4
|
|
||||||
#: Widest score the centre strip must fit. Leagues that can reach three
|
|
||||||
#: digits a side override this with "000-000".
|
|
||||||
_SCORE_PROBE: ClassVar[str] = "00-00"
|
|
||||||
|
|
||||||
# ---- geometry ------------------------------------------------------
|
|
||||||
|
|
||||||
# Non-finite settings are rejected before any int()/round(): "inf" reaches
|
|
||||||
# these from config as a float or a string, passes an `isinstance` plus
|
|
||||||
# `>= 0` check unharmed, and then raises OverflowError out of int() --
|
|
||||||
# which the old `except (TypeError, ValueError)` did not catch, so it
|
|
||||||
# aborted the whole card render. Present in all eight plugins before this
|
|
||||||
# moved to the core; fixing it here fixes it in all eight.
|
|
||||||
|
|
||||||
def _score_reserve_width(self) -> int:
|
|
||||||
"""Centre strip the score actually needs, measured rather than assumed.
|
|
||||||
|
|
||||||
The gap was derived from the card width alone (width x
|
|
||||||
CENTER_GAP_RATIO, clamped to CENTER_GAP_MAX_PX) while the score's size
|
|
||||||
comes from config and the element-style resolver. Nothing compared the
|
|
||||||
two, so any score wider than the clamp was drawn over the logos.
|
|
||||||
Measuring it keeps the strip wide enough for whatever font is in play.
|
|
||||||
"""
|
|
||||||
try:
|
|
||||||
probe = ImageDraw.Draw(Image.new("RGB", (4, 4)))
|
|
||||||
width = probe.textlength(self._SCORE_PROBE, font=self.fonts['score'])
|
|
||||||
return int(width) + 2 * self._SCORE_LOGO_GUTTER_PX
|
|
||||||
except Exception:
|
|
||||||
self.logger.debug("Score reserve measurement failed", exc_info=True)
|
|
||||||
return 0
|
|
||||||
|
|
||||||
def _center_gap_width(self) -> int:
|
|
||||||
"""Width of the middle strip kept clear of logos.
|
|
||||||
|
|
||||||
``scroll_card.center_gap`` pins it outright; otherwise it scales with
|
|
||||||
the card width between the configurable min and max. 0 restores
|
|
||||||
edge-to-edge logos.
|
|
||||||
"""
|
|
||||||
configured = self._scroll_card_option("center_gap")
|
|
||||||
if (isinstance(configured, (int, float))
|
|
||||||
and math.isfinite(configured) and configured >= 0):
|
|
||||||
return int(configured)
|
|
||||||
ratio = self._scroll_card_option("center_gap_ratio", self.CENTER_GAP_RATIO)
|
|
||||||
low = self._scroll_card_option("center_gap_min", self.CENTER_GAP_MIN_PX)
|
|
||||||
high = self._scroll_card_option("center_gap_max", self.CENTER_GAP_MAX_PX)
|
|
||||||
try:
|
|
||||||
ratio, low, high = float(ratio), float(low), float(high)
|
|
||||||
if not (math.isfinite(ratio) and math.isfinite(low)
|
|
||||||
and math.isfinite(high)):
|
|
||||||
return self.CENTER_GAP_MIN_PX
|
|
||||||
scaled = round(self.display_width * ratio)
|
|
||||||
derived = int(max(int(low), min(int(high), scaled)))
|
|
||||||
# A strip narrower than the score is the bug, not a style choice.
|
|
||||||
# An explicit ``center_gap`` is still honoured above, including 0.
|
|
||||||
return max(derived, self._score_reserve_width())
|
|
||||||
except (TypeError, ValueError, OverflowError):
|
|
||||||
return self.CENTER_GAP_MIN_PX
|
|
||||||
|
|
||||||
def _logo_slot_width(self) -> int:
|
|
||||||
"""Per-side logo slot, leaving the center gap clear.
|
|
||||||
|
|
||||||
No longer capped at display_height: the card is sized as two
|
|
||||||
full-height logos plus the measured gap, so what is left after the gap
|
|
||||||
is exactly the logo's share. The cap was what froze the logos at 46px
|
|
||||||
on the old flat 128px card.
|
|
||||||
"""
|
|
||||||
available = (self.display_width - self._center_gap_width()) // 2
|
|
||||||
return max(8, available)
|
|
||||||
|
|
||||||
def _logo_cache_key(self, name: str) -> str:
|
|
||||||
"""Cache key scoped to the logo slot.
|
|
||||||
|
|
||||||
One cache dict is shared by renderers built for different card widths,
|
|
||||||
so a logo sized for a wide slot must not be handed to a narrow one.
|
|
||||||
"""
|
|
||||||
return f"{name}@{self._logo_slot_width()}x{self.display_height}"
|
|
||||||
|
|
||||||
def _layout_offset(self, element: str, axis: str, default: int = 0) -> int:
|
|
||||||
"""X/Y nudge for one element, from customization.layout.
|
|
||||||
|
|
||||||
Same block the full-screen scorebug reads (sports.py
|
|
||||||
_get_layout_offset), so a nudge configured in the web UI now moves
|
|
||||||
the element on the scroll/Vegas card too -- previously the schema
|
|
||||||
advertised these offsets but this renderer ignored them.
|
|
||||||
"""
|
|
||||||
try:
|
|
||||||
layout = (self.config or {}).get("customization", {}).get("layout", {})
|
|
||||||
value = (layout.get(element) or {}).get(axis, default)
|
|
||||||
if isinstance(value, bool):
|
|
||||||
return default
|
|
||||||
if isinstance(value, (int, float)):
|
|
||||||
return int(value) if math.isfinite(value) else default
|
|
||||||
if isinstance(value, str):
|
|
||||||
parsed = float(value)
|
|
||||||
return int(parsed) if math.isfinite(parsed) else default
|
|
||||||
except (TypeError, ValueError, OverflowError):
|
|
||||||
pass
|
|
||||||
return default
|
|
||||||
|
|
||||||
# ---- upcoming cards ------------------------------------------------
|
|
||||||
|
|
||||||
def _upcoming_date_and_time(self, game: Dict) -> Tuple[str, str]:
|
|
||||||
"""(date, time) for an upcoming card, from the extractor's flat keys."""
|
|
||||||
return (
|
|
||||||
str(game.get("game_date", "") or ""),
|
|
||||||
str(game.get("game_time", "") or ""),
|
|
||||||
)
|
|
||||||
|
|
||||||
def _draw_upcoming_center(self, draw: "ImageDraw.ImageDraw", game: Dict) -> None:
|
|
||||||
"""Draw the middle of an upcoming card.
|
|
||||||
|
|
||||||
Never a score: an upcoming game has not started, so the extractor's
|
|
||||||
0-0 is noise. Either the VS text (default), the date and time stacked,
|
|
||||||
or nothing at all.
|
|
||||||
"""
|
|
||||||
mode = self._upcoming_center_mode()
|
|
||||||
if mode == "none":
|
|
||||||
return
|
|
||||||
|
|
||||||
if mode == "vs":
|
|
||||||
vs_text = self._vs_text()
|
|
||||||
if not vs_text:
|
|
||||||
return
|
|
||||||
vs_width = draw.textlength(vs_text, font=self.fonts['score'])
|
|
||||||
vs_x = (self.display_width - vs_width) // 2 + self._layout_offset('score', 'x_offset')
|
|
||||||
vs_y = (self.display_height // 2) - 3 + self._layout_offset('score', 'y_offset')
|
|
||||||
self._draw_text_with_outline(
|
|
||||||
draw, vs_text, (vs_x, vs_y), self.fonts['score'],
|
|
||||||
fill=self._element_color('score_text')
|
|
||||||
)
|
|
||||||
return
|
|
||||||
|
|
||||||
date_text, time_text = self._upcoming_date_and_time(game)
|
|
||||||
lines = []
|
|
||||||
if self._scroll_card_option("show_date", True):
|
|
||||||
lines.append(self._format_game_date(date_text, game))
|
|
||||||
if self._scroll_card_option("show_time", True):
|
|
||||||
lines.append(self._format_game_time(time_text))
|
|
||||||
lines = [t for t in lines if t]
|
|
||||||
if not lines:
|
|
||||||
return
|
|
||||||
font = self.fonts.get('detail') or self.fonts['time']
|
|
||||||
line_h = 7
|
|
||||||
top = (self.display_height // 2) - (len(lines) * line_h) // 2
|
|
||||||
top += self._layout_offset('score', 'y_offset')
|
|
||||||
for i, line in enumerate(lines):
|
|
||||||
width = draw.textlength(line, font=font)
|
|
||||||
x = (self.display_width - width) // 2 + self._layout_offset('score', 'x_offset')
|
|
||||||
self._draw_text_with_outline(
|
|
||||||
draw, line, (x, top + i * line_h), font,
|
|
||||||
fill=self._element_color('detail_text')
|
|
||||||
)
|
|
||||||
|
|
||||||
def _draw_upcoming_game_status(self, draw: "ImageDraw.ImageDraw", game: Dict) -> None:
|
|
||||||
"""Draw the date and time around an upcoming card.
|
|
||||||
|
|
||||||
Time top and date bottom by default; scroll_card.swap_date_time puts
|
|
||||||
the date on top instead. Skipped when the pair is stacked in the
|
|
||||||
middle, which would otherwise print them twice.
|
|
||||||
"""
|
|
||||||
if self._upcoming_center_mode() == "date_time":
|
|
||||||
return
|
|
||||||
|
|
||||||
date_raw, time_raw = self._upcoming_date_and_time(game)
|
|
||||||
date_text = (self._format_game_date(date_raw, game)
|
|
||||||
if self._scroll_card_option("show_date", True) else "")
|
|
||||||
time_text = (self._format_game_time(time_raw)
|
|
||||||
if self._scroll_card_option("show_time", True) else "")
|
|
||||||
|
|
||||||
if self._scroll_card_option("swap_date_time", False):
|
|
||||||
top_text, top_el, bottom_text, bottom_el = (
|
|
||||||
date_text, 'date', time_text, 'time')
|
|
||||||
top_font = self.fonts.get('detail') or self.fonts['time']
|
|
||||||
bottom_font = self.fonts['time']
|
|
||||||
top_color, bottom_color = 'detail_text', 'period_text'
|
|
||||||
else:
|
|
||||||
top_text, top_el, bottom_text, bottom_el = (
|
|
||||||
time_text, 'time', date_text, 'date')
|
|
||||||
top_font = self.fonts['time']
|
|
||||||
bottom_font = self.fonts.get('detail') or self.fonts['time']
|
|
||||||
top_color, bottom_color = 'period_text', 'detail_text'
|
|
||||||
|
|
||||||
if top_text:
|
|
||||||
top_width = draw.textlength(top_text, font=top_font)
|
|
||||||
top_x = (self.display_width - top_width) // 2 + self._layout_offset(top_el, 'x_offset')
|
|
||||||
top_y = 1 + self._layout_offset(top_el, 'y_offset')
|
|
||||||
self._draw_text_with_outline(
|
|
||||||
draw, top_text, (top_x, top_y), top_font,
|
|
||||||
fill=self._element_color(top_color)
|
|
||||||
)
|
|
||||||
|
|
||||||
if bottom_text:
|
|
||||||
bottom_width = draw.textlength(bottom_text, font=bottom_font)
|
|
||||||
bottom_x = ((self.display_width - bottom_width) // 2
|
|
||||||
+ self._layout_offset(bottom_el, 'x_offset'))
|
|
||||||
# Measured, not a fixed -7: the detail font is 6px in most plugins
|
|
||||||
# but 10px in soccer and nrl, where "Sep 19" ran past the card.
|
|
||||||
ink_bottom = draw.textbbox((0, 0), bottom_text, font=bottom_font)[3]
|
|
||||||
bottom_y = (max(0, self.display_height - ink_bottom - 1)
|
|
||||||
+ self._layout_offset(bottom_el, 'y_offset'))
|
|
||||||
self._draw_text_with_outline(
|
|
||||||
draw, bottom_text, (bottom_x, bottom_y), bottom_font,
|
|
||||||
fill=self._element_color(bottom_color)
|
|
||||||
)
|
|
||||||
|
|
||||||
# ---- rankings ------------------------------------------------------
|
|
||||||
|
|
||||||
def set_rankings_cache(self, rankings: Dict[str, int]) -> None:
|
|
||||||
"""Set the team rankings cache for display."""
|
|
||||||
self._team_rankings_cache = rankings
|
|
||||||
File diff suppressed because it is too large
Load Diff
+8
-116
@@ -14,7 +14,6 @@ import json
|
|||||||
from typing import Dict, List, Optional, Tuple
|
from typing import Dict, List, Optional, Tuple
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from PIL import Image, ImageDraw, ImageFont
|
from PIL import Image, ImageDraw, ImageFont
|
||||||
from PIL.PngImagePlugin import PngInfo
|
|
||||||
from requests.adapters import HTTPAdapter
|
from requests.adapters import HTTPAdapter
|
||||||
from urllib3.util.retry import Retry
|
from urllib3.util.retry import Retry
|
||||||
from src.common.permission_utils import (
|
from src.common.permission_utils import (
|
||||||
@@ -26,96 +25,6 @@ from src.common.permission_utils import (
|
|||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
#: PNG text key stamped into a generated placeholder so a later run can tell it
|
|
||||||
#: apart from a real logo that happens to be small.
|
|
||||||
PLACEHOLDER_MARKER = "ledmatrix_placeholder"
|
|
||||||
|
|
||||||
#: Geometry of a generated placeholder, used to recognise ones written before
|
|
||||||
#: the marker existed. Those are already on users' disks and would otherwise
|
|
||||||
#: never be retried.
|
|
||||||
PLACEHOLDER_SIZE = (64, 64)
|
|
||||||
PLACEHOLDER_BG = (100, 100, 100, 255)
|
|
||||||
|
|
||||||
#: How long a placeholder is trusted before the real logo is attempted again.
|
|
||||||
#: A placeholder means the download failed, and download failures are usually
|
|
||||||
#: transient (no network at boot, ESPN blipping). Retrying every frame would
|
|
||||||
#: hammer the API from a Pi that is also driving a panel; never retrying leaves
|
|
||||||
#: the team a grey box forever, which is the bug this exists to avoid.
|
|
||||||
PLACEHOLDER_RETRY_SECONDS = 6 * 60 * 60
|
|
||||||
|
|
||||||
|
|
||||||
def is_placeholder_logo(filepath: Path) -> bool:
|
|
||||||
"""True if the file at ``filepath`` is a generated placeholder, not a logo.
|
|
||||||
|
|
||||||
Checks the marker first, then falls back to matching the placeholder's
|
|
||||||
exact geometry and background colour so files written before the marker was
|
|
||||||
introduced are still recognised.
|
|
||||||
"""
|
|
||||||
try:
|
|
||||||
with Image.open(filepath) as img:
|
|
||||||
if img.info.get(PLACEHOLDER_MARKER):
|
|
||||||
return True
|
|
||||||
if img.size != PLACEHOLDER_SIZE:
|
|
||||||
return False
|
|
||||||
return img.convert("RGBA").getpixel((0, 0)) == PLACEHOLDER_BG
|
|
||||||
except Exception:
|
|
||||||
# Unreadable file: not provably a placeholder, and the caller's own
|
|
||||||
# error handling is better placed to deal with it.
|
|
||||||
return False
|
|
||||||
|
|
||||||
|
|
||||||
def should_attempt_download(filepath: Path, force_download: bool = False) -> bool:
|
|
||||||
"""Whether a real logo is worth (re)fetching for ``filepath``.
|
|
||||||
|
|
||||||
True when nothing is there, when the caller forced it, or when what is
|
|
||||||
there is a placeholder old enough to retry. A *fresh* placeholder says a
|
|
||||||
download just failed, so retrying it immediately would hammer the API for
|
|
||||||
a result that is very unlikely to have changed.
|
|
||||||
"""
|
|
||||||
if force_download or not filepath.exists():
|
|
||||||
return True
|
|
||||||
if not is_placeholder_logo(filepath):
|
|
||||||
return False
|
|
||||||
age = placeholder_age_seconds(filepath)
|
|
||||||
return age is None or age >= PLACEHOLDER_RETRY_SECONDS
|
|
||||||
|
|
||||||
|
|
||||||
def refresh_placeholder_timestamp(filepath: Path) -> bool:
|
|
||||||
"""Restamp a placeholder so a failed retry restarts the back-off clock.
|
|
||||||
|
|
||||||
Without this a stale placeholder stays stale: every later call sees an
|
|
||||||
expired timestamp, retries, fails, and leaves the timestamp untouched --
|
|
||||||
which is a download attempt per call, the opposite of what the back-off is
|
|
||||||
for.
|
|
||||||
"""
|
|
||||||
try:
|
|
||||||
if not is_placeholder_logo(filepath):
|
|
||||||
return False
|
|
||||||
metadata = PngInfo()
|
|
||||||
metadata.add_text(PLACEHOLDER_MARKER, str(time.time()))
|
|
||||||
with Image.open(filepath) as img:
|
|
||||||
img.copy().save(filepath, "PNG", pnginfo=metadata)
|
|
||||||
return True
|
|
||||||
except Exception:
|
|
||||||
logger.debug("Could not refresh placeholder timestamp for %s", filepath,
|
|
||||||
exc_info=True)
|
|
||||||
return False
|
|
||||||
|
|
||||||
|
|
||||||
def placeholder_age_seconds(filepath: Path) -> Optional[float]:
|
|
||||||
"""Seconds since a placeholder was written, or None if unknown."""
|
|
||||||
try:
|
|
||||||
with Image.open(filepath) as img:
|
|
||||||
stamped = img.info.get(PLACEHOLDER_MARKER)
|
|
||||||
if stamped and stamped != "1":
|
|
||||||
return max(0.0, time.time() - float(stamped))
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
try:
|
|
||||||
return max(0.0, time.time() - filepath.stat().st_mtime)
|
|
||||||
except OSError:
|
|
||||||
return None
|
|
||||||
|
|
||||||
class LogoDownloader:
|
class LogoDownloader:
|
||||||
"""Centralized logo downloader for team logos from ESPN API."""
|
"""Centralized logo downloader for team logos from ESPN API."""
|
||||||
|
|
||||||
@@ -590,10 +499,8 @@ class LogoDownloader:
|
|||||||
filename = f"{self.normalize_abbreviation(abbreviation)}.png"
|
filename = f"{self.normalize_abbreviation(abbreviation)}.png"
|
||||||
filepath = Path(logo_dir) / filename
|
filepath = Path(logo_dir) / filename
|
||||||
|
|
||||||
# A placeholder does not count as existing -- it is a previous
|
# Skip if already exists and not forcing download
|
||||||
# failure, and a bulk pass is exactly where it should get another
|
if filepath.exists() and not force_download:
|
||||||
# chance, subject to the same back-off as everywhere else.
|
|
||||||
if not should_attempt_download(filepath, force_download):
|
|
||||||
logger.debug(f"Skipping {display_name}: {filename} already exists")
|
logger.debug(f"Skipping {display_name}: {filename} already exists")
|
||||||
continue
|
continue
|
||||||
|
|
||||||
@@ -652,9 +559,8 @@ class LogoDownloader:
|
|||||||
filename = f"{self.normalize_abbreviation(abbreviation)}.png"
|
filename = f"{self.normalize_abbreviation(abbreviation)}.png"
|
||||||
filepath = Path(logo_dir) / filename
|
filepath = Path(logo_dir) / filename
|
||||||
|
|
||||||
# Same eligibility rule as every other download site: a stale
|
# Skip if already exists and not forcing download
|
||||||
# placeholder is a failed download, not a logo.
|
if filepath.exists() and not force_download:
|
||||||
if not should_attempt_download(filepath, force_download):
|
|
||||||
logger.debug(f"Skipping {display_name} ({category}, {conference}): {filename} already exists")
|
logger.debug(f"Skipping {display_name} ({category}, {conference}): {filename} already exists")
|
||||||
continue
|
continue
|
||||||
|
|
||||||
@@ -768,16 +674,11 @@ class LogoDownloader:
|
|||||||
# Fallback without font
|
# Fallback without font
|
||||||
draw.text((16, 24), text, fill=(255, 255, 255, 255))
|
draw.text((16, 24), text, fill=(255, 255, 255, 255))
|
||||||
|
|
||||||
# Stamp it so a later run can tell this apart from a real logo and
|
logo.save(filepath)
|
||||||
# retry the download, instead of treating the file's existence as
|
|
||||||
# proof the logo was fetched.
|
|
||||||
metadata = PngInfo()
|
|
||||||
metadata.add_text(PLACEHOLDER_MARKER, str(time.time()))
|
|
||||||
logo.save(filepath, "PNG", pnginfo=metadata)
|
|
||||||
|
|
||||||
# Set proper file permissions after saving
|
# Set proper file permissions after saving
|
||||||
ensure_file_permissions(filepath, get_assets_file_mode())
|
ensure_file_permissions(filepath, get_assets_file_mode())
|
||||||
|
|
||||||
logger.info(f"Created placeholder logo for {team_abbreviation} at {filepath}")
|
logger.info(f"Created placeholder logo for {team_abbreviation} at {filepath}")
|
||||||
return True
|
return True
|
||||||
|
|
||||||
@@ -869,18 +770,9 @@ def download_missing_logo(league: str, team_id: str, team_abbreviation: str, log
|
|||||||
# Use the exact filepath that was passed in (respects config settings)
|
# Use the exact filepath that was passed in (respects config settings)
|
||||||
filepath = logo_path
|
filepath = logo_path
|
||||||
|
|
||||||
if filepath.exists() and not should_attempt_download(filepath):
|
if filepath.exists():
|
||||||
# Either a real logo, or a placeholder too fresh to be worth retrying.
|
|
||||||
logger.debug(f"Logo already exists for {team_abbreviation} ({league})")
|
logger.debug(f"Logo already exists for {team_abbreviation} ({league})")
|
||||||
return True
|
return True
|
||||||
if filepath.exists():
|
|
||||||
# A placeholder is a *failed* download wearing the real logo's
|
|
||||||
# filename. Treating it as "already exists" is what pinned a team to a
|
|
||||||
# grey box permanently after one transient failure.
|
|
||||||
logger.info(
|
|
||||||
"Logo for %s (%s) is a placeholder from a failed download; "
|
|
||||||
"retrying the real logo", team_abbreviation, league,
|
|
||||||
)
|
|
||||||
|
|
||||||
# Try to download the real logo first
|
# Try to download the real logo first
|
||||||
logger.info(f"Attempting to download logo for {team_abbreviation} from {league}")
|
logger.info(f"Attempting to download logo for {team_abbreviation} from {league}")
|
||||||
|
|||||||
@@ -1602,26 +1602,6 @@ class PluginStoreManager:
|
|||||||
'error': f'Manifest missing required fields: {", ".join(missing_fields)}'
|
'error': f'Manifest missing required fields: {", ".join(missing_fields)}'
|
||||||
}
|
}
|
||||||
|
|
||||||
# Refuse a plugin that needs a newer core than this one, exactly as
|
|
||||||
# _install_plugin_impl does after its download. Sideloading is an
|
|
||||||
# explicit act rather than an automatic store update, but the floor
|
|
||||||
# is not advice about intent -- it is a statement that the plugin
|
|
||||||
# cannot run here, and letting it through produces the same silent
|
|
||||||
# PluginState.ERROR at load. This was the last of the three routes
|
|
||||||
# in that skipped the check.
|
|
||||||
#
|
|
||||||
# Before the move, so the `finally` below removes the temp tree and
|
|
||||||
# nothing half-installed is left behind.
|
|
||||||
from src import __version__ as core_version
|
|
||||||
from src.plugin_system import compatibility
|
|
||||||
|
|
||||||
compatible, reason = compatibility.check(manifest, core_version)
|
|
||||||
if not compatible:
|
|
||||||
self.logger.error(
|
|
||||||
"Refusing to install %s from %s: %s",
|
|
||||||
plugin_id, repo_url, reason)
|
|
||||||
return {'success': False, 'error': reason}
|
|
||||||
|
|
||||||
# Validate version fields consistency (warnings only, not required)
|
# Validate version fields consistency (warnings only, not required)
|
||||||
validation_errors = self._validate_manifest_version_fields(manifest)
|
validation_errors = self._validate_manifest_version_fields(manifest)
|
||||||
if validation_errors:
|
if validation_errors:
|
||||||
|
|||||||
@@ -8,27 +8,11 @@ ensure_logo_directory, and the download_missing_logo function path
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
import os
|
import os
|
||||||
import time
|
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from unittest.mock import patch, Mock, MagicMock
|
from unittest.mock import patch, Mock, MagicMock
|
||||||
|
|
||||||
from PIL import Image
|
from src.logo_downloader import LogoDownloader
|
||||||
from PIL.PngImagePlugin import PngInfo
|
|
||||||
|
|
||||||
from src.logo_downloader import (
|
|
||||||
PLACEHOLDER_BG,
|
|
||||||
PLACEHOLDER_MARKER,
|
|
||||||
PLACEHOLDER_RETRY_SECONDS,
|
|
||||||
PLACEHOLDER_SIZE,
|
|
||||||
LogoDownloader,
|
|
||||||
download_missing_logo,
|
|
||||||
is_placeholder_logo,
|
|
||||||
placeholder_age_seconds,
|
|
||||||
refresh_placeholder_timestamp,
|
|
||||||
should_attempt_download,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
@@ -143,211 +127,3 @@ class TestEnsureLogoDirectory:
|
|||||||
with patch("builtins.open", side_effect=mock_open):
|
with patch("builtins.open", side_effect=mock_open):
|
||||||
result = downloader.ensure_logo_directory(test_dir)
|
result = downloader.ensure_logo_directory(test_dir)
|
||||||
assert result is False
|
assert result is False
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
|
||||||
# Placeholder detection and retry
|
|
||||||
#
|
|
||||||
# A failed download used to be cached as a placeholder wearing the real logo's
|
|
||||||
# filename, and download_missing_logo returned early on "the file exists". One
|
|
||||||
# transient failure therefore pinned a team to a grey box permanently.
|
|
||||||
# ---------------------------------------------------------------------------
|
|
||||||
|
|
||||||
class TestPlaceholderLogos:
|
|
||||||
def _placeholder(self, tmp_path, abbrev="COLL"):
|
|
||||||
downloader = LogoDownloader()
|
|
||||||
assert downloader.create_placeholder_logo(abbrev, str(tmp_path)) is True
|
|
||||||
return tmp_path / f"{abbrev}.png"
|
|
||||||
|
|
||||||
def test_generated_placeholder_is_recognised(self, tmp_path):
|
|
||||||
assert is_placeholder_logo(self._placeholder(tmp_path)) is True
|
|
||||||
|
|
||||||
def test_real_logo_is_not_a_placeholder(self, tmp_path):
|
|
||||||
real = tmp_path / "REAL.png"
|
|
||||||
Image.new("RGBA", (500, 500), (12, 34, 56, 255)).save(real)
|
|
||||||
assert is_placeholder_logo(real) is False
|
|
||||||
|
|
||||||
def test_legacy_unmarked_placeholder_is_recognised(self, tmp_path):
|
|
||||||
"""Placeholders written before the marker existed must still be caught.
|
|
||||||
|
|
||||||
They are already sitting on users' disks; if they were not recognised
|
|
||||||
those teams would stay grey boxes forever even after this fix.
|
|
||||||
"""
|
|
||||||
legacy = tmp_path / "LEGACY.png"
|
|
||||||
Image.new("RGBA", PLACEHOLDER_SIZE, PLACEHOLDER_BG).save(legacy)
|
|
||||||
assert is_placeholder_logo(legacy) is True
|
|
||||||
|
|
||||||
def test_same_size_but_different_colour_is_not_a_placeholder(self, tmp_path):
|
|
||||||
real = tmp_path / "SMALL.png"
|
|
||||||
Image.new("RGBA", PLACEHOLDER_SIZE, (10, 200, 10, 255)).save(real)
|
|
||||||
assert is_placeholder_logo(real) is False
|
|
||||||
|
|
||||||
def test_missing_file_is_not_a_placeholder(self, tmp_path):
|
|
||||||
assert is_placeholder_logo(tmp_path / "nope.png") is False
|
|
||||||
|
|
||||||
def test_existing_real_logo_short_circuits_without_downloading(self, tmp_path):
|
|
||||||
real = tmp_path / "REAL.png"
|
|
||||||
Image.new("RGBA", (500, 500), (1, 2, 3, 255)).save(real)
|
|
||||||
with patch.object(LogoDownloader, "download_logo") as download:
|
|
||||||
assert download_missing_logo(
|
|
||||||
"afl", "1", "REAL", real, logo_url="http://example/x.png") is True
|
|
||||||
download.assert_not_called()
|
|
||||||
|
|
||||||
def _age_placeholder(self, path, seconds):
|
|
||||||
"""Rewrite a placeholder's marker so it reads as `seconds` old."""
|
|
||||||
metadata = PngInfo()
|
|
||||||
metadata.add_text(PLACEHOLDER_MARKER, str(time.time() - seconds))
|
|
||||||
with Image.open(path) as img:
|
|
||||||
img.copy().save(path, "PNG", pnginfo=metadata)
|
|
||||||
|
|
||||||
def test_stale_placeholder_triggers_a_retry(self, tmp_path):
|
|
||||||
path = self._placeholder(tmp_path)
|
|
||||||
self._age_placeholder(path, PLACEHOLDER_RETRY_SECONDS + 60)
|
|
||||||
assert placeholder_age_seconds(path) > PLACEHOLDER_RETRY_SECONDS
|
|
||||||
|
|
||||||
with patch.object(LogoDownloader, "download_logo", return_value=True) as download:
|
|
||||||
assert download_missing_logo(
|
|
||||||
"afl", "1", "COLL", path,
|
|
||||||
logo_url="http://example/coll.png") is True
|
|
||||||
download.assert_called_once()
|
|
||||||
|
|
||||||
def test_placeholder_age_survives_an_mtime_touch(self, tmp_path):
|
|
||||||
"""The age comes from the stamp, not the filesystem.
|
|
||||||
|
|
||||||
Anything that rewrites file times -- a backup restore, an rsync, a
|
|
||||||
permissions fix script -- would otherwise reset the retry clock.
|
|
||||||
"""
|
|
||||||
path = self._placeholder(tmp_path)
|
|
||||||
self._age_placeholder(path, PLACEHOLDER_RETRY_SECONDS + 60)
|
|
||||||
now = time.time()
|
|
||||||
os.utime(path, (now, now))
|
|
||||||
assert placeholder_age_seconds(path) > PLACEHOLDER_RETRY_SECONDS
|
|
||||||
|
|
||||||
def test_fresh_placeholder_does_not_retry(self, tmp_path):
|
|
||||||
"""Rate limiting: a placeholder written seconds ago must not re-download.
|
|
||||||
|
|
||||||
Without this the fix would trade a permanent grey box for an ESPN
|
|
||||||
request on every frame.
|
|
||||||
"""
|
|
||||||
path = self._placeholder(tmp_path)
|
|
||||||
with patch.object(LogoDownloader, "download_logo") as download:
|
|
||||||
assert download_missing_logo(
|
|
||||||
"afl", "1", "COLL", path,
|
|
||||||
logo_url="http://example/coll.png") is True
|
|
||||||
download.assert_not_called()
|
|
||||||
|
|
||||||
|
|
||||||
class TestDownloadEligibility:
|
|
||||||
"""One rule, shared by every download site.
|
|
||||||
|
|
||||||
The two bulk loops and the single-logo path each had their own idea of what
|
|
||||||
counted as "already have it", which is how one of them ended up retrying
|
|
||||||
fresh placeholders and the other skipping stale ones forever.
|
|
||||||
"""
|
|
||||||
|
|
||||||
def _placeholder(self, tmp_path, abbrev="COLL"):
|
|
||||||
assert LogoDownloader().create_placeholder_logo(abbrev, str(tmp_path))
|
|
||||||
return tmp_path / f"{abbrev}.png"
|
|
||||||
|
|
||||||
def _age(self, path, seconds):
|
|
||||||
metadata = PngInfo()
|
|
||||||
metadata.add_text(PLACEHOLDER_MARKER, str(time.time() - seconds))
|
|
||||||
with Image.open(path) as img:
|
|
||||||
img.copy().save(path, "PNG", pnginfo=metadata)
|
|
||||||
|
|
||||||
def test_missing_file_is_eligible(self, tmp_path):
|
|
||||||
assert should_attempt_download(tmp_path / "nope.png") is True
|
|
||||||
|
|
||||||
def test_real_logo_is_not_eligible(self, tmp_path):
|
|
||||||
real = tmp_path / "REAL.png"
|
|
||||||
Image.new("RGBA", (500, 500), (1, 2, 3, 255)).save(real)
|
|
||||||
assert should_attempt_download(real) is False
|
|
||||||
|
|
||||||
def test_force_download_beats_a_real_logo(self, tmp_path):
|
|
||||||
real = tmp_path / "REAL.png"
|
|
||||||
Image.new("RGBA", (500, 500), (1, 2, 3, 255)).save(real)
|
|
||||||
assert should_attempt_download(real, force_download=True) is True
|
|
||||||
|
|
||||||
def test_fresh_placeholder_is_not_eligible(self, tmp_path):
|
|
||||||
assert should_attempt_download(self._placeholder(tmp_path)) is False
|
|
||||||
|
|
||||||
def test_stale_placeholder_is_eligible(self, tmp_path):
|
|
||||||
path = self._placeholder(tmp_path)
|
|
||||||
self._age(path, PLACEHOLDER_RETRY_SECONDS + 60)
|
|
||||||
assert should_attempt_download(path) is True
|
|
||||||
|
|
||||||
def test_league_bulk_loop_skips_a_fresh_placeholder(self, tmp_path):
|
|
||||||
"""A bulk pass honours the same back-off as everything else."""
|
|
||||||
self._placeholder(tmp_path, "AAA")
|
|
||||||
downloader = LogoDownloader()
|
|
||||||
teams = [{"abbreviation": "AAA", "display_name": "A", "logo_url": "http://x/a.png"}]
|
|
||||||
with patch.object(LogoDownloader, "get_logo_directory", return_value=str(tmp_path)):
|
|
||||||
with patch.object(LogoDownloader, "fetch_teams_data", return_value={"sports": [{}]}):
|
|
||||||
with patch.object(LogoDownloader, "extract_teams_from_data", return_value=teams):
|
|
||||||
with patch.object(LogoDownloader, "download_logo") as download:
|
|
||||||
downloader.download_missing_logos_for_league("nfl")
|
|
||||||
download.assert_not_called()
|
|
||||||
|
|
||||||
def test_league_bulk_loop_retries_a_stale_placeholder(self, tmp_path):
|
|
||||||
path = self._placeholder(tmp_path, "AAA")
|
|
||||||
self._age(path, PLACEHOLDER_RETRY_SECONDS + 60)
|
|
||||||
downloader = LogoDownloader()
|
|
||||||
teams = [{"abbreviation": "AAA", "display_name": "A", "logo_url": "http://x/a.png"}]
|
|
||||||
with patch.object(LogoDownloader, "get_logo_directory", return_value=str(tmp_path)):
|
|
||||||
with patch.object(LogoDownloader, "fetch_teams_data", return_value={"sports": [{}]}):
|
|
||||||
with patch.object(LogoDownloader, "extract_teams_from_data", return_value=teams):
|
|
||||||
with patch.object(LogoDownloader, "download_logo", return_value=True) as download:
|
|
||||||
downloader.download_missing_logos_for_league("nfl")
|
|
||||||
download.assert_called_once()
|
|
||||||
|
|
||||||
def test_ncaa_bulk_loop_retries_a_stale_placeholder(self, tmp_path):
|
|
||||||
"""This loop skipped placeholders forever; it now shares the rule."""
|
|
||||||
path = self._placeholder(tmp_path, "AAA")
|
|
||||||
self._age(path, PLACEHOLDER_RETRY_SECONDS + 60)
|
|
||||||
downloader = LogoDownloader()
|
|
||||||
teams = [{"abbreviation": "AAA", "display_name": "A",
|
|
||||||
"logo_url": "http://x/a.png", "category": "FBS",
|
|
||||||
"conference": "SEC"}]
|
|
||||||
with patch.object(LogoDownloader, "get_logo_directory", return_value=str(tmp_path)):
|
|
||||||
with patch.object(LogoDownloader, "fetch_teams_data", return_value={"sports": [{}]}):
|
|
||||||
with patch.object(LogoDownloader, "extract_teams_from_data", return_value=teams):
|
|
||||||
with patch.object(LogoDownloader, "download_logo", return_value=True) as download:
|
|
||||||
downloader.download_all_ncaa_football_logos()
|
|
||||||
download.assert_called_once()
|
|
||||||
|
|
||||||
def test_ncaa_bulk_loop_skips_a_fresh_placeholder(self, tmp_path):
|
|
||||||
self._placeholder(tmp_path, "AAA")
|
|
||||||
downloader = LogoDownloader()
|
|
||||||
teams = [{"abbreviation": "AAA", "display_name": "A",
|
|
||||||
"logo_url": "http://x/a.png", "category": "FBS",
|
|
||||||
"conference": "SEC"}]
|
|
||||||
with patch.object(LogoDownloader, "get_logo_directory", return_value=str(tmp_path)):
|
|
||||||
with patch.object(LogoDownloader, "fetch_teams_data", return_value={"sports": [{}]}):
|
|
||||||
with patch.object(LogoDownloader, "extract_teams_from_data", return_value=teams):
|
|
||||||
with patch.object(LogoDownloader, "download_logo") as download:
|
|
||||||
downloader.download_all_ncaa_football_logos()
|
|
||||||
download.assert_not_called()
|
|
||||||
|
|
||||||
|
|
||||||
class TestRefreshPlaceholderTimestamp:
|
|
||||||
def test_restarts_the_back_off(self, tmp_path):
|
|
||||||
assert LogoDownloader().create_placeholder_logo("COLL", str(tmp_path))
|
|
||||||
path = tmp_path / "COLL.png"
|
|
||||||
metadata = PngInfo()
|
|
||||||
metadata.add_text(PLACEHOLDER_MARKER, str(time.time() - (PLACEHOLDER_RETRY_SECONDS + 60)))
|
|
||||||
with Image.open(path) as img:
|
|
||||||
img.copy().save(path, "PNG", pnginfo=metadata)
|
|
||||||
assert should_attempt_download(path) is True
|
|
||||||
|
|
||||||
assert refresh_placeholder_timestamp(path) is True
|
|
||||||
assert should_attempt_download(path) is False
|
|
||||||
|
|
||||||
def test_refuses_to_touch_a_real_logo(self, tmp_path):
|
|
||||||
real = tmp_path / "REAL.png"
|
|
||||||
Image.new("RGBA", (500, 500), (1, 2, 3, 255)).save(real)
|
|
||||||
before = real.read_bytes()
|
|
||||||
assert refresh_placeholder_timestamp(real) is False
|
|
||||||
assert real.read_bytes() == before
|
|
||||||
|
|
||||||
def test_missing_file_is_not_an_error(self, tmp_path):
|
|
||||||
assert refresh_placeholder_timestamp(tmp_path / "nope.png") is False
|
|
||||||
|
|||||||
@@ -421,76 +421,3 @@ class TestSessionConfiguration:
|
|||||||
def test_user_agent_and_accept_headers(self, helper):
|
def test_user_agent_and_accept_headers(self, helper):
|
||||||
assert helper.session.headers["User-Agent"] == "LEDMatrix-Common/1.0"
|
assert helper.session.headers["User-Agent"] == "LEDMatrix-Common/1.0"
|
||||||
assert helper.session.headers["Accept"] == "image/*"
|
assert helper.session.headers["Accept"] == "image/*"
|
||||||
|
|
||||||
|
|
||||||
class TestStalePlaceholderHandling:
|
|
||||||
"""load_logo_with_download must not be fooled by a cached placeholder.
|
|
||||||
|
|
||||||
A placeholder wears the real logo's filename, so both the file cache and
|
|
||||||
the in-memory cache can hold one and look like a hit.
|
|
||||||
"""
|
|
||||||
|
|
||||||
def _placeholder(self, tmp_path, abbrev="COLL"):
|
|
||||||
from src.logo_downloader import LogoDownloader
|
|
||||||
assert LogoDownloader().create_placeholder_logo(abbrev, str(tmp_path))
|
|
||||||
return tmp_path / f"{abbrev}.png"
|
|
||||||
|
|
||||||
def _make_stale(self, path):
|
|
||||||
import time
|
|
||||||
from PIL.PngImagePlugin import PngInfo
|
|
||||||
from src.logo_downloader import PLACEHOLDER_MARKER, PLACEHOLDER_RETRY_SECONDS
|
|
||||||
metadata = PngInfo()
|
|
||||||
metadata.add_text(PLACEHOLDER_MARKER, str(time.time() - (PLACEHOLDER_RETRY_SECONDS + 60)))
|
|
||||||
with Image.open(path) as img:
|
|
||||||
img.copy().save(path, "PNG", pnginfo=metadata)
|
|
||||||
|
|
||||||
def test_fresh_placeholder_is_served_without_a_download(self, helper, tmp_path):
|
|
||||||
path = self._placeholder(tmp_path)
|
|
||||||
with patch.object(LogoHelper, "_download_logo") as download:
|
|
||||||
assert helper.load_logo_with_download("COLL", path, "http://x/c.png") is not None
|
|
||||||
download.assert_not_called()
|
|
||||||
|
|
||||||
def test_stale_placeholder_triggers_a_download(self, helper, tmp_path):
|
|
||||||
path = self._placeholder(tmp_path)
|
|
||||||
self._make_stale(path)
|
|
||||||
with patch.object(LogoHelper, "_download_logo") as download:
|
|
||||||
helper.load_logo_with_download("COLL", path, "http://x/c.png")
|
|
||||||
download.assert_called_once()
|
|
||||||
|
|
||||||
def test_replacement_logo_is_not_masked_by_the_cached_placeholder(self, helper, tmp_path):
|
|
||||||
"""The bug this guards: load_logo answers from cache before the disk.
|
|
||||||
|
|
||||||
Without invalidation the freshly downloaded logo would not appear until
|
|
||||||
the process restarted.
|
|
||||||
"""
|
|
||||||
path = self._placeholder(tmp_path)
|
|
||||||
first = helper.load_logo_with_download("COLL", path, "http://x/c.png")
|
|
||||||
assert first is not None
|
|
||||||
|
|
||||||
self._make_stale(path)
|
|
||||||
|
|
||||||
def fake_download(_self, _url, file_path):
|
|
||||||
Image.new("RGB", (500, 500), (7, 8, 9)).save(file_path, format="PNG")
|
|
||||||
|
|
||||||
with patch.object(LogoHelper, "_download_logo", fake_download):
|
|
||||||
second = helper.load_logo_with_download("COLL", path, "http://x/c.png")
|
|
||||||
|
|
||||||
assert second is not None
|
|
||||||
from src.logo_downloader import is_placeholder_logo
|
|
||||||
assert is_placeholder_logo(path) is False
|
|
||||||
assert second.getpixel((0, 0))[:3] == (7, 8, 9)
|
|
||||||
|
|
||||||
def test_failed_retry_restarts_the_back_off(self, helper, tmp_path):
|
|
||||||
"""Otherwise a stale placeholder means a download attempt per call."""
|
|
||||||
from src.logo_downloader import should_attempt_download
|
|
||||||
path = self._placeholder(tmp_path)
|
|
||||||
self._make_stale(path)
|
|
||||||
assert should_attempt_download(path) is True
|
|
||||||
|
|
||||||
def boom(_self, _url, _file_path):
|
|
||||||
raise OSError("network down")
|
|
||||||
|
|
||||||
with patch.object(LogoHelper, "_download_logo", boom):
|
|
||||||
helper.load_logo_with_download("COLL", path, "http://x/c.png")
|
|
||||||
|
|
||||||
assert should_attempt_download(path) is False
|
|
||||||
|
|||||||
@@ -155,9 +155,8 @@ class TestInstallGate:
|
|||||||
rediscovering it. `update_plugin` also has a git branch that pulls in
|
rediscovering it. `update_plugin` also has a git branch that pulls in
|
||||||
place and never re-downloads; that one is gated separately by
|
place and never re-downloads; that one is gated separately by
|
||||||
`_gate_pulled_commit` and pinned in `TestGitPullGate` below. A third
|
`_gate_pulled_commit` and pinned in `TestGitPullGate` below. A third
|
||||||
route, `install_from_url` (sideloading from a URL), is gated in that
|
route, `install_from_url`, is still ungated — sideloading from a URL
|
||||||
function and pinned in `TestSideloadGate`. All three refuse on the same
|
checks required fields but not the floor.
|
||||||
rule.
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def _install_with_manifest(self, store, manifest, core_version, monkeypatch):
|
def _install_with_manifest(self, store, manifest, core_version, monkeypatch):
|
||||||
@@ -218,73 +217,6 @@ class TestInstallGate:
|
|||||||
assert (path / "manifest.json").exists()
|
assert (path / "manifest.json").exists()
|
||||||
|
|
||||||
|
|
||||||
class TestSideloadGate:
|
|
||||||
"""`install_from_url` never looked at the core version.
|
|
||||||
|
|
||||||
Sideloading is an explicit act rather than an automatic store update, so
|
|
||||||
the argument for gating it is different: not "the user did not choose
|
|
||||||
this", but that the floor states the plugin *cannot run here*. Letting it
|
|
||||||
through produces the same silent PluginState.ERROR at load that the store
|
|
||||||
gate exists to prevent, and the user who typed the URL is no better placed
|
|
||||||
to diagnose it than one who pressed Update.
|
|
||||||
"""
|
|
||||||
|
|
||||||
def _sideload(self, store, manifest, core_version, monkeypatch):
|
|
||||||
mgr, plugins_dir = store
|
|
||||||
plugin_id = manifest["id"]
|
|
||||||
|
|
||||||
def fake_clone(repo_url, target_path, branches=None):
|
|
||||||
target_path.mkdir(parents=True, exist_ok=True)
|
|
||||||
(target_path / "manifest.json").write_text(
|
|
||||||
json.dumps(manifest), encoding="utf-8")
|
|
||||||
(target_path / "manager.py").write_text(
|
|
||||||
"class P: pass\n", encoding="utf-8")
|
|
||||||
return "main"
|
|
||||||
|
|
||||||
monkeypatch.setattr(mgr, "_install_via_git", fake_clone)
|
|
||||||
monkeypatch.setattr(mgr, "_install_dependencies", lambda *a, **k: True)
|
|
||||||
|
|
||||||
import src
|
|
||||||
monkeypatch.setattr(src, "__version__", core_version)
|
|
||||||
return mgr.install_from_url("https://example.invalid/plugin"), \
|
|
||||||
plugins_dir / plugin_id
|
|
||||||
|
|
||||||
def test_refuses_a_plugin_that_needs_a_newer_core(self, store, monkeypatch):
|
|
||||||
manifest = {
|
|
||||||
"id": "sideload-newer", "name": "Sideload", "class_name": "P",
|
|
||||||
"display_modes": ["a"], "min_ledmatrix_version": "9.9.9",
|
|
||||||
}
|
|
||||||
result, path = self._sideload(store, manifest, "3.2.0", monkeypatch)
|
|
||||||
|
|
||||||
assert result["success"] is False
|
|
||||||
assert "9.9.9" in result["error"], result["error"]
|
|
||||||
assert not path.exists(), (
|
|
||||||
"a refused sideload must not leave the plugin installed")
|
|
||||||
|
|
||||||
def test_allows_a_compatible_plugin(self, store, monkeypatch):
|
|
||||||
"""The guard against over-refusing: a gate that blocks everything
|
|
||||||
passes the test above and breaks sideloading entirely."""
|
|
||||||
manifest = {
|
|
||||||
"id": "sideload-fine", "name": "Sideload", "class_name": "P",
|
|
||||||
"display_modes": ["a"], "min_ledmatrix_version": "3.0.0",
|
|
||||||
}
|
|
||||||
result, path = self._sideload(store, manifest, "3.2.0", monkeypatch)
|
|
||||||
|
|
||||||
assert result["success"] is True, result.get("error")
|
|
||||||
assert (path / "manifest.json").exists()
|
|
||||||
|
|
||||||
def test_untrustworthy_core_does_not_block_a_2_0_0_floor(
|
|
||||||
self, store, monkeypatch):
|
|
||||||
"""Same rule as the other two routes: a v3.1.0 release reports 1.0.0,
|
|
||||||
and nearly every published manifest floors at 2.0.0."""
|
|
||||||
manifest = {
|
|
||||||
"id": "sideload-floored", "name": "Sideload", "class_name": "P",
|
|
||||||
"display_modes": ["a"], "versions": [{"ledmatrix_min": "2.0.0"}],
|
|
||||||
}
|
|
||||||
result, _ = self._sideload(store, manifest, "1.0.0", monkeypatch)
|
|
||||||
|
|
||||||
assert result["success"] is True, result.get("error")
|
|
||||||
|
|
||||||
# --------------------------------------------------------------------------
|
# --------------------------------------------------------------------------
|
||||||
# The git-pull update path — the one route that does not re-download
|
# The git-pull update path — the one route that does not re-download
|
||||||
# --------------------------------------------------------------------------
|
# --------------------------------------------------------------------------
|
||||||
|
|||||||
@@ -1,195 +0,0 @@
|
|||||||
"""The card helpers the eight scoreboards now share.
|
|
||||||
|
|
||||||
These bodies lived in eight byte-identical copies. Moving them here means one
|
|
||||||
fix reaches every scoreboard — and that a mistake does too, which is what this
|
|
||||||
file guards. Each case below is one the plugins' own code already handled; the
|
|
||||||
point is that it keeps handling it.
|
|
||||||
|
|
||||||
The functions take ``config``/``logger``/``fonts`` as arguments rather than
|
|
||||||
reading them off an instance, so a plugin keeps its method and delegates the
|
|
||||||
body. That is what let all eight adopt this with byte-identical renders.
|
|
||||||
"""
|
|
||||||
|
|
||||||
import logging
|
|
||||||
import json
|
|
||||||
import os
|
|
||||||
|
|
||||||
import pytest
|
|
||||||
|
|
||||||
from src.common import sports_card as C
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
|
||||||
def log():
|
|
||||||
return logging.getLogger("test_sports_card")
|
|
||||||
|
|
||||||
|
|
||||||
class TestSettingsLookup:
|
|
||||||
def test_reads_the_scroll_card_block(self):
|
|
||||||
cfg = {"scroll_card": {"vs_text": "@"}}
|
|
||||||
assert C.scroll_card_option(cfg, "vs_text", "VS") == "@"
|
|
||||||
|
|
||||||
@pytest.mark.parametrize("cfg", [None, {}, {"scroll_card": None},
|
|
||||||
{"scroll_card": {"vs_text": None}}])
|
|
||||||
def test_missing_or_null_falls_back(self, cfg):
|
|
||||||
"""A null in config means "unset", not "empty string"."""
|
|
||||||
assert C.scroll_card_option(cfg, "vs_text", "VS") == "VS"
|
|
||||||
|
|
||||||
def test_upcoming_center_rejects_unknown_modes(self):
|
|
||||||
for bad in ("sideways", "", None, 7):
|
|
||||||
assert C.upcoming_center_mode({"scroll_card": {"upcoming_center": bad}}) == "vs"
|
|
||||||
assert C.upcoming_center_mode({"scroll_card": {"upcoming_center": "DATE_TIME"}}) == "date_time"
|
|
||||||
|
|
||||||
|
|
||||||
class TestColour:
|
|
||||||
def test_rgb_list_and_hex_both_work(self):
|
|
||||||
assert C.element_color({"customization": {"score_text": {"text_color": [1, 2, 3]}}},
|
|
||||||
"score_text") == (1, 2, 3)
|
|
||||||
assert C.element_color({"customization": {"score_text": {"text_color": "#ff8000"}}},
|
|
||||||
"score_text") == (255, 128, 0)
|
|
||||||
|
|
||||||
@pytest.mark.parametrize("value", ["nope", "#fff", [1, 2], None, ["a", "b", "c"]])
|
|
||||||
def test_unusable_colour_falls_back(self, value):
|
|
||||||
cfg = {"customization": {"score_text": {"text_color": value}}}
|
|
||||||
assert C.element_color(cfg, "score_text", (9, 9, 9)) == (9, 9, 9)
|
|
||||||
|
|
||||||
def test_coerce_rgb_clamps_rather_than_rejecting(self):
|
|
||||||
assert C.coerce_rgb([300, -5, 20], (0, 0, 0)) == (255, 0, 20)
|
|
||||||
|
|
||||||
def test_coerce_rgb_refuses_a_three_character_string(self):
|
|
||||||
""""123" would otherwise iterate into three digits and yield a colour."""
|
|
||||||
assert C.coerce_rgb("123", (7, 7, 7)) == (7, 7, 7)
|
|
||||||
|
|
||||||
def test_font_colour_is_resolved_by_identity(self):
|
|
||||||
a, b = object(), object()
|
|
||||||
cfg = {"customization": {"score_text": {"text_color": [4, 5, 6]}}}
|
|
||||||
assert C.font_color(cfg, {"score": a, "team": b}, a) == (4, 5, 6)
|
|
||||||
|
|
||||||
def test_a_shared_face_gives_up_rather_than_guessing(self):
|
|
||||||
"""One object used for two elements has no single right colour."""
|
|
||||||
shared = object()
|
|
||||||
cfg = {"customization": {"score_text": {"text_color": [4, 5, 6]}}}
|
|
||||||
assert C.font_color(cfg, {"score": shared, "team": shared}, shared) == (255, 255, 255)
|
|
||||||
|
|
||||||
|
|
||||||
class TestFavourites:
|
|
||||||
GAME = {"home_abbr": "TB", "away_abbr": "NO", "home_score": "21", "away_score": "17"}
|
|
||||||
|
|
||||||
def test_win_loss_and_tie(self):
|
|
||||||
cfg = {"favorite_teams": ["TB"]}
|
|
||||||
assert C.favorite_result(cfg, self.GAME) == "win"
|
|
||||||
assert C.favorite_result({"favorite_teams": ["NO"]}, self.GAME) == "loss"
|
|
||||||
tied = dict(self.GAME, home_score="3", away_score="3")
|
|
||||||
assert C.favorite_result(cfg, tied) == "tie"
|
|
||||||
|
|
||||||
def test_no_verdict_without_exactly_one_favourite_side(self):
|
|
||||||
assert C.favorite_result({}, self.GAME) is None
|
|
||||||
assert C.favorite_result({"favorite_teams": ["TB", "NO"]}, self.GAME) is None
|
|
||||||
assert C.favorite_result({"favorite_teams": ["SEA"]}, self.GAME) is None
|
|
||||||
|
|
||||||
def test_unusable_scores_give_no_verdict(self):
|
|
||||||
bad = dict(self.GAME, home_score="x")
|
|
||||||
assert C.favorite_result({"favorite_teams": ["TB"]}, bad) is None
|
|
||||||
|
|
||||||
def test_nested_payload_shape_is_read_too(self):
|
|
||||||
game = {"home_team": {"abbrev": "TB", "score": 9},
|
|
||||||
"away_team": {"abbrev": "NO", "score": 2}}
|
|
||||||
assert C.side_score(game, "home") == 9
|
|
||||||
assert C.side_is_favorite(game, "home", {"TB"}) is True
|
|
||||||
|
|
||||||
def test_matches_on_id_where_abbreviations_collide(self):
|
|
||||||
"""NRL keys favourites by ESPN id; abbreviations are not unique there."""
|
|
||||||
game = {"home_abbr": "SYD", "home_id": "4321"}
|
|
||||||
assert C.side_is_favorite(game, "home", {"4321"}) is True
|
|
||||||
|
|
||||||
def test_game_and_config_favourites_are_both_used(self):
|
|
||||||
"""Games carry resolved dynamic groups; config catches later edits."""
|
|
||||||
game = dict(self.GAME, favorite_teams=["NO"], league="nfl")
|
|
||||||
assert set(C.favorite_teams_for({"nfl": {"favorite_teams": ["TB"]}}, game)) == {"NO", "TB"}
|
|
||||||
|
|
||||||
|
|
||||||
class TestDateAndTime:
|
|
||||||
def test_date_formats(self, log):
|
|
||||||
for fmt, want in [("abbrev", "Sep 5"), ("numeric", "9/5"),
|
|
||||||
("day_first", "5 Sep"), ("numeric_day_first", "5/9")]:
|
|
||||||
cfg = {"scroll_card": {"date_format": fmt}}
|
|
||||||
assert C.format_game_date(cfg, log, "9/5") == want
|
|
||||||
|
|
||||||
@pytest.mark.parametrize("raw", ["", "garbage", "13/40", "no/slash/here"])
|
|
||||||
def test_unparseable_dates_pass_through(self, log, raw):
|
|
||||||
assert C.format_game_date({}, log, raw) == raw.strip()
|
|
||||||
|
|
||||||
def test_24h_conversion(self):
|
|
||||||
cfg = {"scroll_card": {"time_format": "24h"}}
|
|
||||||
assert C.format_game_time(cfg, "7:30 PM") == "19:30"
|
|
||||||
assert C.format_game_time(cfg, "12:00 AM") == "00:00"
|
|
||||||
assert C.format_game_time(cfg, "12:15 PM") == "12:15"
|
|
||||||
|
|
||||||
def test_12h_is_left_alone_and_junk_survives(self):
|
|
||||||
assert C.format_game_time({}, "7:30 PM") == "7:30 PM"
|
|
||||||
assert C.format_game_time({"scroll_card": {"time_format": "24h"}}, "soon") == "soon"
|
|
||||||
|
|
||||||
def test_a_bad_timezone_falls_back_to_utc(self, log):
|
|
||||||
"""A typo in config should not blank the card."""
|
|
||||||
from datetime import timezone
|
|
||||||
assert C.card_tzinfo({"timezone": "Not/AZone"}, log) is timezone.utc
|
|
||||||
|
|
||||||
|
|
||||||
class TestFontSizing:
|
|
||||||
def test_snaps_to_the_faces_pixel_grid(self):
|
|
||||||
assert C.crisp_size("4x6-font.ttf", 6) == 7 # 7px grid
|
|
||||||
assert C.crisp_size("PressStart2P-Regular.ttf", 10) == 8
|
|
||||||
assert C.crisp_size("PressStart2P-Regular.ttf", 13) == 16
|
|
||||||
|
|
||||||
def test_an_unknown_face_is_never_second_guessed(self):
|
|
||||||
assert C.crisp_size("SomeUserFont.ttf", 11) == 11
|
|
||||||
|
|
||||||
def test_aliases_resolve_before_the_grid_lookup(self):
|
|
||||||
assert C.crisp_size("four_by_six", 6) == C.crisp_size("4x6-font.ttf", 6)
|
|
||||||
|
|
||||||
@pytest.mark.parametrize("desired", [0, -3, None])
|
|
||||||
def test_unusable_sizes_pass_through_without_raising(self, desired):
|
|
||||||
"""None reached this in the field; football's variant raised TypeError."""
|
|
||||||
assert C.crisp_size("4x6-font.ttf", desired) == desired
|
|
||||||
|
|
||||||
def test_schema_cache_is_keyed_per_schema_not_globally(self, tmp_path):
|
|
||||||
"""Two plugins declaring different defaults must not share an answer."""
|
|
||||||
a, b = tmp_path / "a.json", tmp_path / "b.json"
|
|
||||||
for path, size in ((a, 11), (b, 22)):
|
|
||||||
path.write_text(json.dumps({"properties": {"customization": {"properties": {
|
|
||||||
"score_text": {"properties": {"font_size": {"default": size}}}}}}}))
|
|
||||||
assert C.schema_font_size(str(a), "score_text") == 11
|
|
||||||
assert C.schema_font_size(str(b), "score_text") == 22
|
|
||||||
|
|
||||||
def test_a_missing_schema_is_not_an_error(self, tmp_path):
|
|
||||||
assert C.schema_font_size(str(tmp_path / "nope.json"), "score_text") is None
|
|
||||||
|
|
||||||
def test_a_configured_size_matching_the_schema_default_is_not_a_choice(self, tmp_path):
|
|
||||||
"""The web UI writes the whole default block on every save, so
|
|
||||||
font_size == schema default carries no intent and must not pin the
|
|
||||||
install to an off-grid size forever."""
|
|
||||||
schema = tmp_path / "s.json"
|
|
||||||
schema.write_text(json.dumps({"properties": {"customization": {"properties": {
|
|
||||||
"score_text": {"properties": {"font_size": {"default": 10}}}}}}}))
|
|
||||||
got = C.resolve_font_size(str(schema), {"font_size": 10}, "score_text", 10,
|
|
||||||
"PressStart2P-Regular.ttf")
|
|
||||||
assert got == 8, "a default-valued size should snap to the grid"
|
|
||||||
|
|
||||||
def test_a_real_choice_wins(self, tmp_path):
|
|
||||||
schema = tmp_path / "s.json"
|
|
||||||
schema.write_text(json.dumps({"properties": {"customization": {"properties": {
|
|
||||||
"score_text": {"properties": {"font_size": {"default": 10}}}}}}}))
|
|
||||||
got = C.resolve_font_size(str(schema), {"font_size": 13}, "score_text", 10,
|
|
||||||
"PressStart2P-Regular.ttf")
|
|
||||||
assert got == 13, "an explicit size the user chose is not second-guessed"
|
|
||||||
|
|
||||||
|
|
||||||
class TestTables:
|
|
||||||
def test_every_font_key_maps_to_an_element(self):
|
|
||||||
assert set(C.ELEMENT_FOR_FONT) == {"score", "time", "team", "status", "detail", "rank"}
|
|
||||||
|
|
||||||
def test_result_colours_cover_every_verdict(self):
|
|
||||||
assert set(C.FAVORITE_RESULT_COLOR_DEFAULTS) == {"win", "loss", "tie"}
|
|
||||||
|
|
||||||
def test_month_and_weekday_tables_are_complete(self):
|
|
||||||
assert len(C.MONTH_ABBR) == 12 and len(C.WEEKDAY_ABBR) == 7
|
|
||||||
@@ -1,258 +0,0 @@
|
|||||||
"""The shared card geometry, exercised against a host that supplies only what
|
|
||||||
the mixin's contract names.
|
|
||||||
|
|
||||||
The point of these is the contract, not the arithmetic. The mixin reaches for
|
|
||||||
``display_width``, ``fonts``, ``config`` and six ``sports_card`` delegations
|
|
||||||
through ``self``, and the eight plugins are what actually provide them. A
|
|
||||||
stub host that provides exactly the documented surface and nothing else is
|
|
||||||
what catches the mixin quietly growing a dependency the plugins do not have.
|
|
||||||
"""
|
|
||||||
|
|
||||||
import pytest
|
|
||||||
from PIL import Image, ImageDraw, ImageFont
|
|
||||||
|
|
||||||
from src.common.sports_game_renderer import SportsGameRendererMixin
|
|
||||||
|
|
||||||
|
|
||||||
class Host(SportsGameRendererMixin):
|
|
||||||
"""The documented contract, and not one attribute more."""
|
|
||||||
|
|
||||||
def __init__(self, width=128, height=32, config=None, scroll=None):
|
|
||||||
self.display_width = width
|
|
||||||
self.display_height = height
|
|
||||||
self.config = config or {}
|
|
||||||
self.logger = _Logger()
|
|
||||||
self._team_rankings_cache = {}
|
|
||||||
self._scroll = scroll or {}
|
|
||||||
font = ImageFont.load_default()
|
|
||||||
self.fonts = {'score': font, 'time': font, 'detail': font}
|
|
||||||
self.drawn = []
|
|
||||||
|
|
||||||
# -- the six sports_card delegations the mixin calls --
|
|
||||||
def _scroll_card_option(self, key, default=None):
|
|
||||||
return self._scroll.get(key, default)
|
|
||||||
|
|
||||||
def _upcoming_center_mode(self):
|
|
||||||
return self._scroll.get('upcoming_center', 'vs')
|
|
||||||
|
|
||||||
def _vs_text(self):
|
|
||||||
return self._scroll.get('vs_text', 'VS')
|
|
||||||
|
|
||||||
def _element_color(self, element):
|
|
||||||
return (255, 255, 255)
|
|
||||||
|
|
||||||
def _format_game_date(self, raw, game):
|
|
||||||
return raw
|
|
||||||
|
|
||||||
def _format_game_time(self, raw):
|
|
||||||
return raw
|
|
||||||
|
|
||||||
# -- the one hook whose body genuinely varies per plugin --
|
|
||||||
def _draw_text_with_outline(self, draw, text, position, font,
|
|
||||||
fill=None, outline_color=(0, 0, 0)):
|
|
||||||
self.drawn.append((text, position))
|
|
||||||
|
|
||||||
|
|
||||||
class _Logger:
|
|
||||||
def debug(self, *a, **k):
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
def _draw():
|
|
||||||
return ImageDraw.Draw(Image.new("RGB", (256, 64)))
|
|
||||||
|
|
||||||
|
|
||||||
class TestCenterGap:
|
|
||||||
def test_explicit_center_gap_wins_outright(self):
|
|
||||||
assert Host(scroll={'center_gap': 31})._center_gap_width() == 31
|
|
||||||
|
|
||||||
def test_explicit_zero_restores_edge_to_edge_logos(self):
|
|
||||||
# 0 is a real setting, not a falsy miss -- the guard is `>= 0`.
|
|
||||||
assert Host(scroll={'center_gap': 0})._center_gap_width() == 0
|
|
||||||
|
|
||||||
def test_otherwise_it_scales_with_card_width_within_the_clamp(self):
|
|
||||||
h = Host(width=512)
|
|
||||||
# 512 * 0.28 = 143, clamped to the 40px ceiling.
|
|
||||||
assert h._center_gap_width() >= h.CENTER_GAP_MIN_PX
|
|
||||||
|
|
||||||
def test_the_gap_never_ends_up_narrower_than_the_score(self):
|
|
||||||
# This is the bug the measurement exists to prevent: a derived gap
|
|
||||||
# smaller than the rendered score drew the score over the logos.
|
|
||||||
h = Host(width=64)
|
|
||||||
assert h._center_gap_width() >= h._score_reserve_width()
|
|
||||||
|
|
||||||
def test_a_junk_ratio_falls_back_to_the_floor(self):
|
|
||||||
h = Host(scroll={'center_gap_ratio': 'wide'})
|
|
||||||
assert h._center_gap_width() == h.CENTER_GAP_MIN_PX
|
|
||||||
|
|
||||||
|
|
||||||
class TestNonFiniteSettings:
|
|
||||||
"""inf reaches int() and raises OverflowError, which the old
|
|
||||||
`except (TypeError, ValueError)` did not catch -- so one bad config value
|
|
||||||
aborted the entire card render rather than falling back."""
|
|
||||||
|
|
||||||
@pytest.mark.parametrize("bad", [float("inf"), float("-inf")])
|
|
||||||
def test_a_non_finite_center_gap_falls_back(self, bad):
|
|
||||||
h = Host(scroll={'center_gap': bad})
|
|
||||||
assert h._center_gap_width() >= h.CENTER_GAP_MIN_PX
|
|
||||||
|
|
||||||
@pytest.mark.parametrize("bad", [float("inf"), float("-inf"), float("nan")])
|
|
||||||
def test_a_non_finite_ratio_falls_back_to_the_floor(self, bad):
|
|
||||||
h = Host(scroll={'center_gap_ratio': bad})
|
|
||||||
assert h._center_gap_width() == h.CENTER_GAP_MIN_PX
|
|
||||||
|
|
||||||
@pytest.mark.parametrize("bad", [float("inf"), float("-inf")])
|
|
||||||
def test_non_finite_clamp_bounds_fall_back(self, bad):
|
|
||||||
h = Host(scroll={'center_gap_min': bad, 'center_gap_max': bad})
|
|
||||||
assert h._center_gap_width() == h.CENTER_GAP_MIN_PX
|
|
||||||
|
|
||||||
@pytest.mark.parametrize("bad", [float("inf"), float("-inf"), "inf", "-inf", "nan"])
|
|
||||||
def test_a_non_finite_layout_offset_gives_the_default(self, bad):
|
|
||||||
cfg = {'customization': {'layout': {'score': {'x_offset': bad}}}}
|
|
||||||
assert Host(config=cfg)._layout_offset('score', 'x_offset', 7) == 7
|
|
||||||
|
|
||||||
def test_a_finite_value_is_still_honoured(self):
|
|
||||||
# The guard must not swallow ordinary settings.
|
|
||||||
assert Host(scroll={'center_gap': 31})._center_gap_width() == 31
|
|
||||||
cfg = {'customization': {'layout': {'score': {'x_offset': -3}}}}
|
|
||||||
assert Host(config=cfg)._layout_offset('score', 'x_offset', 7) == -3
|
|
||||||
|
|
||||||
|
|
||||||
class TestScoreReserve:
|
|
||||||
def test_it_measures_the_probe_plus_both_gutters(self):
|
|
||||||
h = Host()
|
|
||||||
assert h._score_reserve_width() > 2 * h._SCORE_LOGO_GUTTER_PX
|
|
||||||
|
|
||||||
def test_a_wider_probe_reserves_more(self):
|
|
||||||
class Wide(Host):
|
|
||||||
_SCORE_PROBE = "000-000"
|
|
||||||
assert Wide()._score_reserve_width() > Host()._score_reserve_width()
|
|
||||||
|
|
||||||
def test_an_unmeasurable_font_reserves_nothing_rather_than_raising(self):
|
|
||||||
h = Host()
|
|
||||||
h.fonts = {'score': object()}
|
|
||||||
assert h._score_reserve_width() == 0
|
|
||||||
|
|
||||||
|
|
||||||
class TestLogoSlot:
|
|
||||||
def test_the_slot_is_what_is_left_after_the_gap(self):
|
|
||||||
h = Host(width=128, scroll={'center_gap': 40})
|
|
||||||
assert h._logo_slot_width() == 44
|
|
||||||
|
|
||||||
def test_it_is_not_capped_at_the_card_height(self):
|
|
||||||
# The height cap is what froze logos at 46px on a 128px card.
|
|
||||||
h = Host(width=512, height=32, scroll={'center_gap': 40})
|
|
||||||
assert h._logo_slot_width() > h.display_height
|
|
||||||
|
|
||||||
def test_a_gap_wider_than_the_card_still_leaves_a_usable_slot(self):
|
|
||||||
assert Host(width=64, scroll={'center_gap': 200})._logo_slot_width() == 8
|
|
||||||
|
|
||||||
def test_the_cache_key_is_scoped_to_the_slot_not_just_the_name(self):
|
|
||||||
# One cache dict is shared by renderers of different card widths.
|
|
||||||
narrow = Host(width=64, scroll={'center_gap': 20})._logo_cache_key("NYY")
|
|
||||||
wide = Host(width=256, scroll={'center_gap': 20})._logo_cache_key("NYY")
|
|
||||||
assert narrow != wide
|
|
||||||
|
|
||||||
|
|
||||||
class TestLayoutOffset:
|
|
||||||
def _host(self, value):
|
|
||||||
return Host(config={'customization': {'layout': {'score': {'x_offset': value}}}})
|
|
||||||
|
|
||||||
def test_it_reads_the_same_block_as_the_full_screen_scorebug(self):
|
|
||||||
assert self._host(5)._layout_offset('score', 'x_offset') == 5
|
|
||||||
|
|
||||||
def test_a_string_offset_from_the_web_ui_is_coerced(self):
|
|
||||||
assert self._host("-3")._layout_offset('score', 'x_offset') == -3
|
|
||||||
|
|
||||||
def test_a_bool_is_not_silently_an_offset_of_one(self):
|
|
||||||
assert self._host(True)._layout_offset('score', 'x_offset', 9) == 9
|
|
||||||
|
|
||||||
@pytest.mark.parametrize("cfg", [{}, {'customization': {}},
|
|
||||||
{'customization': {'layout': {}}}])
|
|
||||||
def test_a_missing_block_gives_the_default(self, cfg):
|
|
||||||
assert Host(config=cfg)._layout_offset('score', 'x_offset', 7) == 7
|
|
||||||
|
|
||||||
def test_an_unparseable_offset_gives_the_default(self):
|
|
||||||
assert self._host("left")._layout_offset('score', 'x_offset', 4) == 4
|
|
||||||
|
|
||||||
|
|
||||||
class TestUpcomingCenter:
|
|
||||||
def test_none_draws_nothing(self):
|
|
||||||
h = Host(scroll={'upcoming_center': 'none'})
|
|
||||||
h._draw_upcoming_center(_draw(), {})
|
|
||||||
assert h.drawn == []
|
|
||||||
|
|
||||||
def test_vs_is_the_default_and_never_a_score(self):
|
|
||||||
# An upcoming game has not started; the extractor's 0-0 is noise.
|
|
||||||
h = Host()
|
|
||||||
h._draw_upcoming_center(_draw(), {'home_score': 0, 'away_score': 0})
|
|
||||||
assert [t for t, _ in h.drawn] == ['VS']
|
|
||||||
|
|
||||||
def test_an_empty_vs_string_draws_nothing(self):
|
|
||||||
h = Host(scroll={'vs_text': ''})
|
|
||||||
h._draw_upcoming_center(_draw(), {})
|
|
||||||
assert h.drawn == []
|
|
||||||
|
|
||||||
def test_date_time_stacks_both_lines(self):
|
|
||||||
h = Host(scroll={'upcoming_center': 'date_time'})
|
|
||||||
h._draw_upcoming_center(_draw(), {'game_date': 'Sep 19', 'game_time': '7:00 PM'})
|
|
||||||
assert [t for t, _ in h.drawn] == ['Sep 19', '7:00 PM']
|
|
||||||
|
|
||||||
def test_hiding_both_lines_draws_nothing(self):
|
|
||||||
h = Host(scroll={'upcoming_center': 'date_time',
|
|
||||||
'show_date': False, 'show_time': False})
|
|
||||||
h._draw_upcoming_center(_draw(), {'game_date': 'Sep 19', 'game_time': '7:00 PM'})
|
|
||||||
assert h.drawn == []
|
|
||||||
|
|
||||||
|
|
||||||
class TestUpcomingStatus:
|
|
||||||
def test_time_on_top_and_date_below_by_default(self):
|
|
||||||
h = Host()
|
|
||||||
h._draw_upcoming_game_status(_draw(), {'game_date': 'Sep 19', 'game_time': '7:00 PM'})
|
|
||||||
assert [t for t, _ in h.drawn] == ['7:00 PM', 'Sep 19']
|
|
||||||
|
|
||||||
def test_swap_date_time_reverses_them(self):
|
|
||||||
h = Host(scroll={'swap_date_time': True})
|
|
||||||
h._draw_upcoming_game_status(_draw(), {'game_date': 'Sep 19', 'game_time': '7:00 PM'})
|
|
||||||
assert [t for t, _ in h.drawn] == ['Sep 19', '7:00 PM']
|
|
||||||
|
|
||||||
def test_it_stays_out_of_the_way_when_the_centre_already_has_them(self):
|
|
||||||
# Otherwise the date and time print twice on the same card.
|
|
||||||
h = Host(scroll={'upcoming_center': 'date_time'})
|
|
||||||
h._draw_upcoming_game_status(_draw(), {'game_date': 'Sep 19', 'game_time': '7:00 PM'})
|
|
||||||
assert h.drawn == []
|
|
||||||
|
|
||||||
def test_the_bottom_line_is_measured_not_a_fixed_offset(self):
|
|
||||||
# A fixed -7 ran "Sep 19" past the card wherever the detail font is
|
|
||||||
# 10px rather than 6px.
|
|
||||||
h = Host(height=64)
|
|
||||||
h._draw_upcoming_game_status(_draw(), {'game_date': 'Sep 19', 'game_time': '7:00 PM'})
|
|
||||||
bottom_y = h.drawn[1][1][1]
|
|
||||||
assert 0 <= bottom_y < h.display_height
|
|
||||||
|
|
||||||
|
|
||||||
class TestRankings:
|
|
||||||
def test_set_rankings_cache_replaces_the_cache(self):
|
|
||||||
h = Host()
|
|
||||||
h.set_rankings_cache({'UGA': 1})
|
|
||||||
assert h._team_rankings_cache == {'UGA': 1}
|
|
||||||
|
|
||||||
|
|
||||||
class TestContract:
|
|
||||||
def test_the_mixin_carries_no_state_of_its_own(self):
|
|
||||||
# Adoption must be one line on the class statement; a mixin with an
|
|
||||||
# __init__ would force eight constructors to cooperate.
|
|
||||||
assert '__init__' not in SportsGameRendererMixin.__dict__
|
|
||||||
|
|
||||||
def test_a_host_providing_the_documented_surface_needs_nothing_more(self):
|
|
||||||
# Host defines exactly what the module docstring names. If the mixin
|
|
||||||
# grows a new self.* dependency, this is what fails.
|
|
||||||
h = Host()
|
|
||||||
h._center_gap_width()
|
|
||||||
h._logo_slot_width()
|
|
||||||
h._logo_cache_key("X")
|
|
||||||
h._layout_offset('score', 'x_offset')
|
|
||||||
h._upcoming_date_and_time({})
|
|
||||||
h._draw_upcoming_center(_draw(), {})
|
|
||||||
h._draw_upcoming_game_status(_draw(), {})
|
|
||||||
h.set_rankings_cache({})
|
|
||||||
@@ -1,321 +0,0 @@
|
|||||||
"""The shared sports.py mixins: their host contract, and _plugin_dir.
|
|
||||||
|
|
||||||
Two things are worth testing here and the rest is not. The 45 method bodies
|
|
||||||
moved verbatim from the plugins, so they are covered by the plugins' own tests
|
|
||||||
and by 176 byte-identical safety-harness renders. What is genuinely new is:
|
|
||||||
|
|
||||||
1. The contract. Every ``self.<CONSTANT>`` a mixin reads must be defined on the
|
|
||||||
mixin, or a host that does not happen to declare it raises AttributeError at
|
|
||||||
runtime. Two were missed on the first pass (_QUALITY_CHOICES and
|
|
||||||
_RANKING_COVERAGE_SECONDS); the eight plugins all declare them, so nothing
|
|
||||||
failed -- it would only have bitten a ninth. The test derives the list rather
|
|
||||||
than restating it, so the next omission fails here instead of in the field.
|
|
||||||
|
|
||||||
2. ``_plugin_dir``. This is the only line of genuinely new logic in the move. In
|
|
||||||
sports.py these methods found config_schema.json with ``__file__``; here that
|
|
||||||
is src/common/, so the plugin directory has to be recovered from the
|
|
||||||
instance -- and getting it wrong is silent, costing grid-snapped font sizes
|
|
||||||
(measured at 81% anti-aliased edges) rather than raising.
|
|
||||||
"""
|
|
||||||
|
|
||||||
import ast
|
|
||||||
import os
|
|
||||||
import sys
|
|
||||||
import types
|
|
||||||
from abc import ABC
|
|
||||||
|
|
||||||
import pytest
|
|
||||||
|
|
||||||
from src.common import sports_shared
|
|
||||||
from src.common.sports_shared import (
|
|
||||||
SportsCoreSharedMixin, SportsLiveSharedMixin, SportsRecentSharedMixin)
|
|
||||||
|
|
||||||
MIXINS = (SportsCoreSharedMixin, SportsLiveSharedMixin, SportsRecentSharedMixin)
|
|
||||||
|
|
||||||
|
|
||||||
def _constants_read_by_mixins():
|
|
||||||
"""Every ALL-CAPS ``self.X`` the mixin bodies read, found by parsing them."""
|
|
||||||
tree = ast.parse(open(sports_shared.__file__).read())
|
|
||||||
names = set()
|
|
||||||
for node in ast.walk(tree):
|
|
||||||
if (isinstance(node, ast.Attribute)
|
|
||||||
and isinstance(node.value, ast.Name)
|
|
||||||
and node.value.id == "self"
|
|
||||||
and node.attr.upper() == node.attr):
|
|
||||||
names.add(node.attr)
|
|
||||||
return names
|
|
||||||
|
|
||||||
|
|
||||||
class TestHostContract:
|
|
||||||
def test_every_constant_read_is_also_defined(self):
|
|
||||||
# Otherwise a host that does not declare it raises AttributeError the
|
|
||||||
# first time the code path runs -- which for these is mid-render.
|
|
||||||
missing = sorted(
|
|
||||||
name for name in _constants_read_by_mixins()
|
|
||||||
if not any(hasattr(m, name) for m in MIXINS))
|
|
||||||
assert missing == [], (
|
|
||||||
f"read but never defined on a mixin: {missing}. Give each a default "
|
|
||||||
f"on SportsCoreSharedMixin and document it in the module docstring.")
|
|
||||||
|
|
||||||
@pytest.mark.parametrize("name,expected", [
|
|
||||||
("_QUALITY_CHOICES", frozenset({"any", "ranked"})),
|
|
||||||
("_RANKING_COVERAGE_SECONDS", 3600),
|
|
||||||
("_SCORE_PROBE_TEXT", "00-00"),
|
|
||||||
("_FONT_DESIGN_HEIGHT", 32),
|
|
||||||
])
|
|
||||||
def test_defaults_match_what_the_plugins_ship(self, name, expected):
|
|
||||||
# The eight plugins declare their own copies, which shadow these. The
|
|
||||||
# values must still agree, or a ninth plugin inheriting the default
|
|
||||||
# behaves differently from the eight.
|
|
||||||
assert getattr(SportsCoreSharedMixin, name) == expected
|
|
||||||
|
|
||||||
def test_only_the_recent_mixin_carries_a_constructor(self):
|
|
||||||
# SportsCore and SportsLive keep their own __init__ -- those differ per
|
|
||||||
# plugin. SportsRecent.__init__ was one of the 48 byte-identical bodies,
|
|
||||||
# so it moved with the rest; that is deliberate, not an oversight.
|
|
||||||
assert "__init__" not in SportsCoreSharedMixin.__dict__
|
|
||||||
assert "__init__" not in SportsLiveSharedMixin.__dict__
|
|
||||||
assert "__init__" in SportsRecentSharedMixin.__dict__
|
|
||||||
|
|
||||||
def test_the_recent_constructor_still_chains_to_the_host(self):
|
|
||||||
"""Its zero-arg super() binds to where it is DEFINED, not where it is used.
|
|
||||||
|
|
||||||
Moving a body containing bare ``super()`` is the one move that can
|
|
||||||
change meaning: the compiler closes over __class__ = the defining class,
|
|
||||||
so after the move that is SportsRecentSharedMixin rather than the
|
|
||||||
plugin's SportsRecent. It still works only because the mixin is listed
|
|
||||||
first, leaving the host class next in the MRO -- adopt it in the other
|
|
||||||
order and the chain silently skips the host's __init__.
|
|
||||||
"""
|
|
||||||
calls = []
|
|
||||||
|
|
||||||
class Host:
|
|
||||||
def __init__(self, config, display_manager, cache_manager, logger, sport_key):
|
|
||||||
calls.append(sport_key)
|
|
||||||
self.mode_config = {}
|
|
||||||
|
|
||||||
class Recent(SportsRecentSharedMixin, Host):
|
|
||||||
pass
|
|
||||||
|
|
||||||
inst = Recent({}, None, None, None, "nhl")
|
|
||||||
assert calls == ["nhl"], "the host constructor must still run"
|
|
||||||
assert inst.current_game_index == 0
|
|
||||||
assert inst.update_interval == 3600
|
|
||||||
assert inst._zero_clock_timestamps == {}
|
|
||||||
|
|
||||||
def test_adopting_the_recent_mixin_second_would_skip_the_host(self):
|
|
||||||
# The failure mode the ordering above prevents, pinned so nobody
|
|
||||||
# "tidies" the base list.
|
|
||||||
calls = []
|
|
||||||
|
|
||||||
class Host:
|
|
||||||
def __init__(self, *a):
|
|
||||||
calls.append(a)
|
|
||||||
self.mode_config = {}
|
|
||||||
|
|
||||||
class Wrong(Host, SportsRecentSharedMixin):
|
|
||||||
pass
|
|
||||||
|
|
||||||
Wrong({}, None, None, None, "nhl")
|
|
||||||
# Host.__init__ wins and the mixin's setup never runs at all.
|
|
||||||
assert not hasattr(Wrong({}, None, None, None, "nhl"), "current_game_index")
|
|
||||||
|
|
||||||
|
|
||||||
class _Host(SportsCoreSharedMixin):
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
def _write_plugin(tmp_path, name="fakeplug", schema=True):
|
|
||||||
"""A throwaway package on sys.path, with or without a config_schema.json."""
|
|
||||||
d = tmp_path / name
|
|
||||||
d.mkdir()
|
|
||||||
(d / "__init__.py").write_text("")
|
|
||||||
(d / "mod.py").write_text("class Leaf:\n pass\n")
|
|
||||||
if schema:
|
|
||||||
(d / "config_schema.json").write_text(
|
|
||||||
'{"properties": {"customization": {"properties": '
|
|
||||||
'{"score": {"properties": {"font_size": {"default": 16}}}}}}}')
|
|
||||||
return d
|
|
||||||
|
|
||||||
|
|
||||||
class TestPluginDir:
|
|
||||||
def test_it_finds_the_directory_holding_config_schema_json(self, tmp_path, monkeypatch):
|
|
||||||
d = _write_plugin(tmp_path)
|
|
||||||
monkeypatch.syspath_prepend(str(tmp_path))
|
|
||||||
mod = __import__("fakeplug.mod", fromlist=["Leaf"])
|
|
||||||
host = type("H", (mod.Leaf, SportsCoreSharedMixin), {})()
|
|
||||||
assert host._plugin_dir() == str(d)
|
|
||||||
|
|
||||||
def test_a_class_built_by_type_still_resolves(self, tmp_path, monkeypatch):
|
|
||||||
# SportsCore is an ABC, so type(name, bases, ns) reports __module__ as
|
|
||||||
# "abc" rather than the plugin -- which is exactly what the plugins'
|
|
||||||
# own tests build. Walking the MRO is what steps past it.
|
|
||||||
d = _write_plugin(tmp_path, "abcplug")
|
|
||||||
monkeypatch.syspath_prepend(str(tmp_path))
|
|
||||||
mod = __import__("abcplug.mod", fromlist=["Leaf"])
|
|
||||||
|
|
||||||
class Base(SportsCoreSharedMixin, mod.Leaf, ABC):
|
|
||||||
pass
|
|
||||||
|
|
||||||
synthetic = type("Probe", (Base,), {})
|
|
||||||
assert synthetic.__module__ == "abc", "precondition: the trap this guards"
|
|
||||||
assert synthetic.__new__(synthetic)._plugin_dir() == str(d)
|
|
||||||
|
|
||||||
def test_it_returns_none_when_no_schema_is_anywhere_on_the_mro(self, tmp_path, monkeypatch):
|
|
||||||
d = _write_plugin(tmp_path, "noschema", schema=False)
|
|
||||||
monkeypatch.syspath_prepend(str(tmp_path))
|
|
||||||
mod = __import__("noschema.mod", fromlist=["Leaf"])
|
|
||||||
host = type("H", (mod.Leaf, SportsCoreSharedMixin), {})()
|
|
||||||
# None rather than a wrong guess: _schema_font_size then caches empty
|
|
||||||
# and every element keeps its own default.
|
|
||||||
assert host._plugin_dir() is None
|
|
||||||
|
|
||||||
def test_it_never_returns_the_core_module_directory(self):
|
|
||||||
# The bug this replaced: __file__ pointed at src/common/, so the schema
|
|
||||||
# was never found and font sizes silently stopped snapping to the grid.
|
|
||||||
host = _Host()
|
|
||||||
core_common = os.path.dirname(os.path.abspath(sports_shared.__file__))
|
|
||||||
assert host._plugin_dir() != core_common
|
|
||||||
|
|
||||||
def test_a_module_with_no_file_is_skipped_not_crashed_on(self, monkeypatch):
|
|
||||||
# Namespace packages and some frozen/dynamic modules have no __file__.
|
|
||||||
ghost = types.ModuleType("ghost_no_file")
|
|
||||||
if hasattr(ghost, "__file__"):
|
|
||||||
del ghost.__file__
|
|
||||||
monkeypatch.setitem(sys.modules, "ghost_no_file", ghost)
|
|
||||||
cls = type("H", (SportsCoreSharedMixin,), {"__module__": "ghost_no_file"})
|
|
||||||
assert cls.__new__(cls)._plugin_dir() is None
|
|
||||||
|
|
||||||
|
|
||||||
class TestSchemaFontSize:
|
|
||||||
def test_it_reads_the_plugin_schema_not_the_cores(self, tmp_path, monkeypatch):
|
|
||||||
d = _write_plugin(tmp_path, "sizeplug")
|
|
||||||
monkeypatch.syspath_prepend(str(tmp_path))
|
|
||||||
mod = __import__("sizeplug.mod", fromlist=["Leaf"])
|
|
||||||
host = type("H", (mod.Leaf, SportsCoreSharedMixin), {})()
|
|
||||||
assert host._schema_font_size("score") == 16
|
|
||||||
|
|
||||||
def test_an_unknown_element_is_none(self, tmp_path, monkeypatch):
|
|
||||||
_write_plugin(tmp_path, "unkplug")
|
|
||||||
monkeypatch.syspath_prepend(str(tmp_path))
|
|
||||||
mod = __import__("unkplug.mod", fromlist=["Leaf"])
|
|
||||||
host = type("H", (mod.Leaf, SportsCoreSharedMixin), {})()
|
|
||||||
assert host._schema_font_size("nonesuch") is None
|
|
||||||
|
|
||||||
def test_an_empty_key_is_none_without_touching_the_disk(self):
|
|
||||||
assert _Host()._schema_font_size("") is None
|
|
||||||
|
|
||||||
def test_a_missing_schema_degrades_to_none_rather_than_raising(self, tmp_path, monkeypatch):
|
|
||||||
_write_plugin(tmp_path, "bareplug", schema=False)
|
|
||||||
monkeypatch.syspath_prepend(str(tmp_path))
|
|
||||||
mod = __import__("bareplug.mod", fromlist=["Leaf"])
|
|
||||||
host = type("H", (mod.Leaf, SportsCoreSharedMixin), {})()
|
|
||||||
assert host._schema_font_size("score") is None
|
|
||||||
|
|
||||||
|
|
||||||
class _LiveHost(SportsLiveSharedMixin):
|
|
||||||
"""The documented contract for the live mixin, and nothing else."""
|
|
||||||
|
|
||||||
def __init__(self, no_data_interval=300, stale_game_timeout=600, over=()):
|
|
||||||
self.no_data_interval = no_data_interval
|
|
||||||
self.stale_game_timeout = stale_game_timeout
|
|
||||||
self.game_update_timestamps = {}
|
|
||||||
self._over = set(over)
|
|
||||||
|
|
||||||
class _L:
|
|
||||||
def __getattr__(self, _n):
|
|
||||||
return lambda *a, **k: None
|
|
||||||
self.logger = _L()
|
|
||||||
|
|
||||||
def _is_game_really_over(self, game):
|
|
||||||
return game.get("id") in self._over
|
|
||||||
|
|
||||||
|
|
||||||
class TestLiveMixin:
|
|
||||||
"""These three moved to the core, so they are tested here.
|
|
||||||
|
|
||||||
They were already covered by hockey's and lacrosse's own tests, but those
|
|
||||||
two plugins disable live mode in their safety-harness fixtures, so the 176
|
|
||||||
renders never exercise this path. Testing the mixin directly means the
|
|
||||||
coverage no longer depends on which plugin happens to have a unit test.
|
|
||||||
"""
|
|
||||||
|
|
||||||
def test_a_stale_game_is_dropped_and_forgotten(self):
|
|
||||||
h = _LiveHost(stale_game_timeout=600)
|
|
||||||
import time as _t
|
|
||||||
h.game_update_timestamps["g1"] = {"last_seen": _t.time() - 5000}
|
|
||||||
games = [{"id": "g1", "home_abbr": "H", "away_abbr": "A"}]
|
|
||||||
h._detect_stale_games(games)
|
|
||||||
assert games == []
|
|
||||||
assert "g1" not in h.game_update_timestamps, "its timestamp must go too"
|
|
||||||
|
|
||||||
def test_a_fresh_game_survives(self):
|
|
||||||
h = _LiveHost(stale_game_timeout=600)
|
|
||||||
import time as _t
|
|
||||||
h.game_update_timestamps["g1"] = {"last_seen": _t.time() - 5}
|
|
||||||
games = [{"id": "g1"}]
|
|
||||||
h._detect_stale_games(games)
|
|
||||||
assert len(games) == 1
|
|
||||||
|
|
||||||
def test_a_game_never_seen_is_not_treated_as_stale(self):
|
|
||||||
# last_seen 0 means "no reading", not "seen at the epoch".
|
|
||||||
h = _LiveHost()
|
|
||||||
games = [{"id": "g1"}]
|
|
||||||
h._detect_stale_games(games)
|
|
||||||
assert len(games) == 1
|
|
||||||
|
|
||||||
def test_a_game_with_no_id_is_left_alone(self):
|
|
||||||
h = _LiveHost()
|
|
||||||
games = [{"home_abbr": "H"}]
|
|
||||||
h._detect_stale_games(games)
|
|
||||||
assert len(games) == 1
|
|
||||||
|
|
||||||
def test_a_finished_game_is_dropped_even_when_fresh(self):
|
|
||||||
h = _LiveHost(over=("g2",))
|
|
||||||
games = [{"id": "g1"}, {"id": "g2"}]
|
|
||||||
h._detect_stale_games(games)
|
|
||||||
assert [g["id"] for g in games] == ["g1"]
|
|
||||||
|
|
||||||
def test_removing_several_does_not_skip_any(self):
|
|
||||||
# It iterates a copy for exactly this reason; mutating the live list
|
|
||||||
# while looping would step over the element after each removal.
|
|
||||||
h = _LiveHost(over=("g1", "g2", "g3"))
|
|
||||||
games = [{"id": "g1"}, {"id": "g2"}, {"id": "g3"}]
|
|
||||||
h._detect_stale_games(games)
|
|
||||||
assert games == []
|
|
||||||
|
|
||||||
def test_the_idle_interval_escalates_with_the_empty_streak(self):
|
|
||||||
h = _LiveHost(no_data_interval=60)
|
|
||||||
h.live_idle_max_interval = 100000
|
|
||||||
base = h._idle_live_interval()
|
|
||||||
h._empty_live_streak = 6
|
|
||||||
short = h._idle_live_interval()
|
|
||||||
h._empty_live_streak = 24
|
|
||||||
long = h._idle_live_interval()
|
|
||||||
assert base < short < long
|
|
||||||
|
|
||||||
def test_the_ceiling_bounds_even_the_unescalated_interval(self):
|
|
||||||
# base > ceiling is a reachable config: the two settings are
|
|
||||||
# independent integers with no cross-validation. Returning base
|
|
||||||
# unclamped made the wait SHRINK as the streak grew.
|
|
||||||
h = _LiveHost(no_data_interval=3600)
|
|
||||||
h.live_idle_max_interval = 900
|
|
||||||
h._empty_live_streak = 0
|
|
||||||
assert h._idle_live_interval() == 900
|
|
||||||
h._empty_live_streak = 24
|
|
||||||
assert h._idle_live_interval() == 900
|
|
||||||
|
|
||||||
def test_finding_a_live_game_resets_the_streak(self):
|
|
||||||
h = _LiveHost()
|
|
||||||
h._note_live_fetch(False)
|
|
||||||
h._note_live_fetch(False)
|
|
||||||
assert h._empty_live_streak == 2
|
|
||||||
h._note_live_fetch(True)
|
|
||||||
assert h._empty_live_streak == 0
|
|
||||||
|
|
||||||
def test_the_streak_starts_from_absent_state(self):
|
|
||||||
# The host is not required to pre-declare _empty_live_streak.
|
|
||||||
h = _LiveHost()
|
|
||||||
assert not hasattr(h, "_empty_live_streak")
|
|
||||||
h._note_live_fetch(False)
|
|
||||||
assert h._empty_live_streak == 1
|
|
||||||
Reference in New Issue
Block a user