mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-04 14:25:08 +00:00
feat(common): favorite_team_check and sports_timezone, promoted from the scoreboards (sports consolidation stage 2) (#665)
* feat(common): favorite_team_check and sports_timezone, promoted from the scoreboards (sports consolidation stage 2) Two new hardware-free modules, taken from files the scoreboard plugins carry as copies: - src/common/favorite_team_check.py: FavoriteTeamCheck(logger, leagues), the seven byte-identical <sport>_favorite_check.py copies. Same code; the only additions are two type annotations (for the mypy ratchet). - src/common/sports_timezone.py: resolve_timezone_name(), resolve_timezone(), system_timezone_name(), from the ten <sport>_timezone.py copies. They differed only in the plugin label named in the nothing-resolved warning and the write-back-bug values, which become keyword-only arguments (plugin_label, writeback_fixed_in). Same resolution order and log text. Tests are ported from the plugins' own (test_favorite_check.py, test_schedule_note_uses_game_dates.py, test_timezone_resolution.py; the timezone ones run once per plugin's values and pin the exact warning text). Both modules are on the mypy ratchet, in src/common/README.md, the CHANGELOG's Unreleased section and SPORTS_UNIFICATION's module table. Nothing in core uses them yet. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * docs(common): bdf_font and json_body shipped in 3.5.0 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * chore(common): annotate the favourite check's deliberate except/pass for Bandit Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Chuck
+27
-2
@@ -24,11 +24,12 @@ Rules for the package:
|
||||
| Module | For | Plugins import it? | Since |
|
||||
|---|---|---|---|
|
||||
| [`api_helper`](#api_helper) | HTTP GET/POST with caching and rate limiting | Yes | — |
|
||||
| [`bdf_font`](#bdf_font) | Load and draw BDF bitmap fonts | Yes, if drawing BDF text directly | Unreleased |
|
||||
| [`bdf_font`](#bdf_font) | Load and draw BDF bitmap fonts | Yes, if drawing BDF text directly | 3.5.0 |
|
||||
| [`espn_dates`](#espn_dates) | Fetch ESPN scoreboards across a date range | Yes (scoreboards) | 3.5.0 |
|
||||
| [`favorite_team_check`](#favorite_team_check) | Log why a favourite team code shows nothing | Yes (scoreboards) | Unreleased |
|
||||
| [`font_layout`](#font_layout) | Reproducible TrueType loading, crisp sizes | Yes | 3.4.0 |
|
||||
| [`frame_timing`](#frame_timing) | Timing of every presented frame, stall watchdog | No, core-internal | n/a |
|
||||
| [`json_body`](#json_body) | Parse a response body as JSON, with orjson if installed | Optional (large payloads) | Unreleased |
|
||||
| [`json_body`](#json_body) | Parse a response body as JSON, with orjson if installed | Optional (large payloads) | 3.5.0 |
|
||||
| [`logo_helper`](#logo_helper) | Load, resize and cache team logos | Yes | — |
|
||||
| [`path_safety`](#path_safety) | Turn request-supplied names into safe paths | No, core-internal | n/a |
|
||||
| [`permission_utils`](#permission_utils) | File modes and shared-group ownership | Rarely | — |
|
||||
@@ -41,6 +42,7 @@ Rules for the package:
|
||||
| [`sports_helpers`](#sports_helpers) | Small helpers every scoreboard `sports.py` copies | Yes (scoreboards) | 3.5.0 |
|
||||
| [`sports_scroll`](#sports_scroll) | Scoreboard scroll-display orchestration | Yes (scoreboards) | 3.2.0 |
|
||||
| [`sports_shared`](#sports_shared) | Sport-independent `sports.py` methods | Yes (scoreboards) | 3.3.0 |
|
||||
| [`sports_timezone`](#sports_timezone) | Which timezone a scoreboard draws start times in | Yes (scoreboards) | Unreleased |
|
||||
| [`sync_manager`](#sync_manager) | Leader/follower sync between two displays | No, core-internal | n/a |
|
||||
| [`text_helper`](#text_helper) | Outlined text, wrapping, measurement | Yes | — |
|
||||
|
||||
@@ -100,6 +102,17 @@ results; `espn_date_chunks()`, `fetch_espn_date_chunks()`,
|
||||
`clamp_espn_limit()` and `merge_scoreboard_payloads()` are the pieces.
|
||||
Scoreboard plugins also bundle a copy for older cores.
|
||||
|
||||
### favorite_team_check
|
||||
|
||||
[`favorite_team_check.py`](favorite_team_check.py).
|
||||
`FavoriteTeamCheck(logger, leagues)`, where `leagues` maps a league key to
|
||||
`(display name, ESPN sport/league path)`. `schedule(league_key, favorites)`
|
||||
checks the configured favourite team codes against ESPN's team list once per
|
||||
league, on a daemon thread, and logs a bad code with the nearest real one, or
|
||||
says the league has nothing on yet; `reset()` re-arms it after a config edit.
|
||||
Diagnostics only: every failure is swallowed. Scoreboard plugins also bundle
|
||||
a copy for older cores.
|
||||
|
||||
### font_layout
|
||||
|
||||
[`font_layout.py`](font_layout.py). `load_truetype(path, size)` is
|
||||
@@ -237,6 +250,18 @@ fonts, colours, dates, the switch-mode upcoming card). The docstring lists
|
||||
the attributes the host class must have and the three methods deliberately
|
||||
left out.
|
||||
|
||||
### sports_timezone
|
||||
|
||||
[`sports_timezone.py`](sports_timezone.py).
|
||||
`resolve_timezone_name(config, plugin_manager, cache_manager, log, *,
|
||||
plugin_label, writeback_fixed_in=None)` and `resolve_timezone(...)` (the same
|
||||
as a pytz zone): the plugin's own `timezone`, then the global one via either
|
||||
manager's `config_manager`, then the host's zone (`system_timezone_name()`),
|
||||
then UTC. `plugin_label` names the plugin in the warning logged when nothing
|
||||
resolves; `writeback_fixed_in` is for a plugin that once wrote `"UTC"` into
|
||||
the saved config (a bare plugin-level `"UTC"` is then ignored when another
|
||||
source disagrees). Scoreboard plugins also bundle a copy for older cores.
|
||||
|
||||
### sync_manager
|
||||
|
||||
[`sync_manager.py`](sync_manager.py). Core-internal. `DisplaySyncManager`
|
||||
|
||||
@@ -0,0 +1,325 @@
|
||||
"""
|
||||
Explain an empty screen: a wrong team code, or a season that has not started.
|
||||
|
||||
Favourite teams are matched by exact ESPN abbreviation, so a plausible-looking
|
||||
code silently matches nothing and the plugin shows an empty screen with no hint
|
||||
that the code is at fault. The codes are not always guessable — ESPN calls
|
||||
Alabama ``ALA`` rather than ``BAMA``, and Golden State ``GS`` rather than
|
||||
``GSW``. Between seasons a perfectly correct code produces the same empty
|
||||
screen for a completely different reason, and the two were indistinguishable
|
||||
from the logs.
|
||||
|
||||
This module is diagnostics only. It runs on a daemon thread, once per league per
|
||||
process, and every failure is swallowed: it must never delay a frame or change
|
||||
what is displayed.
|
||||
"""
|
||||
|
||||
import difflib
|
||||
import logging
|
||||
import re
|
||||
import threading
|
||||
from datetime import datetime, timezone
|
||||
from typing import Dict, Iterable, List, Optional, Set, Tuple
|
||||
|
||||
TEAMS_URL = "https://site.api.espn.com/apis/site/v2/sports/{path}/teams?limit=1000"
|
||||
SCOREBOARD_URL = "https://site.api.espn.com/apis/site/v2/sports/{path}/scoreboard"
|
||||
REQUEST_TIMEOUT = 15
|
||||
|
||||
|
||||
class FavoriteTeamCheck:
|
||||
"""
|
||||
Validates configured favourite team codes against ESPN, and says so in the log.
|
||||
|
||||
``leagues`` maps the plugin's own league key to a
|
||||
``(human readable name, ESPN sport/league path)`` pair, e.g.
|
||||
``{'nhl': ('NHL', 'hockey/nhl')}``.
|
||||
"""
|
||||
|
||||
# How far out the next fixture has to be before it is worth mentioning.
|
||||
# An off day or two is normal mid-season and saying so would just be noise.
|
||||
GAP_DAYS = 3
|
||||
|
||||
def __init__(self, logger: Optional[logging.Logger],
|
||||
leagues: Dict[str, Tuple[str, str]]) -> None:
|
||||
self.logger = logger or logging.getLogger(__name__)
|
||||
self.leagues = leagues
|
||||
self._checked: Set[str] = set()
|
||||
self._lock = threading.Lock()
|
||||
|
||||
def reset(self) -> None:
|
||||
"""Re-check on the next call, e.g. after the user edits the config."""
|
||||
with self._lock:
|
||||
self._checked.clear()
|
||||
|
||||
def schedule(self, league_key: str, favorites: Iterable[str]) -> None:
|
||||
"""Check one league in the background, at most once per process."""
|
||||
try:
|
||||
favorites = [str(f) for f in (favorites or []) if str(f).strip()]
|
||||
if not favorites or league_key not in self.leagues:
|
||||
return
|
||||
with self._lock:
|
||||
if league_key in self._checked:
|
||||
return
|
||||
self._checked.add(league_key)
|
||||
threading.Thread(
|
||||
target=self._run, args=(league_key, favorites),
|
||||
name="favorite-team-check", daemon=True,
|
||||
).start()
|
||||
except Exception:
|
||||
pass # nosec B110 - a diagnostic must never be the reason an update fails # nosemgrep
|
||||
|
||||
def _run(self, league_key: str, favorites) -> None:
|
||||
try:
|
||||
self._check(league_key, favorites)
|
||||
except Exception as exc:
|
||||
self.logger.debug("Favorite team check failed for %s: %s",
|
||||
league_key, exc)
|
||||
|
||||
def _check(self, league_key: str, favorites) -> None:
|
||||
name, path = self.leagues[league_key]
|
||||
|
||||
try:
|
||||
teams = self._fetch_teams(path)
|
||||
except Exception as exc:
|
||||
self.logger.debug("Could not verify %s favorite teams: %s", name, exc)
|
||||
return
|
||||
if not teams:
|
||||
# Some ESPN endpoints (college lacrosse) return no teams at all.
|
||||
# Nothing can be concluded, so say nothing.
|
||||
return
|
||||
|
||||
# Dynamic groups like AP_TOP_25 are expanded elsewhere; they are not
|
||||
# team codes and must not be reported as bad ones.
|
||||
codes = [f for f in favorites if not self._is_dynamic(f)]
|
||||
recognised = [f for f in codes if f in teams]
|
||||
unknown = [f for f in codes if f not in teams]
|
||||
|
||||
for code in unknown:
|
||||
self.logger.warning(
|
||||
"%s favorite team %r is not a %s team code.%s "
|
||||
"Every code this league accepts is listed at %s.",
|
||||
name, code, name, self._suggest(code, teams),
|
||||
TEAMS_URL.format(path=path),
|
||||
)
|
||||
|
||||
if codes and not recognised:
|
||||
self.logger.warning(
|
||||
"%s has no recognised favorite teams, so nothing will be shown "
|
||||
"for it. Codes must be ESPN abbreviations, e.g. %s.",
|
||||
name, ", ".join("{} ({})".format(a, n)
|
||||
for a, n in list(sorted(teams.items()))[:3]),
|
||||
)
|
||||
return
|
||||
|
||||
if not recognised:
|
||||
return
|
||||
|
||||
# Codes are fine, so check the other cause of an empty screen.
|
||||
try:
|
||||
note = self._schedule_note(path)
|
||||
except Exception as exc:
|
||||
self.logger.debug("Could not check the %s schedule: %s", name, exc)
|
||||
return
|
||||
|
||||
if note:
|
||||
self.logger.info(
|
||||
"%s favorite teams %s look correct, but %s. An empty display "
|
||||
"until then is expected, not a configuration problem.",
|
||||
name, ", ".join(recognised), note,
|
||||
)
|
||||
else:
|
||||
self.logger.info("%s favorite teams recognised: %s",
|
||||
name, ", ".join(recognised))
|
||||
|
||||
@staticmethod
|
||||
def _is_dynamic(code: str) -> bool:
|
||||
upper = (code or "").strip().upper()
|
||||
return upper.startswith("AP_") or upper.startswith("TOP_") or "TOP_" in upper
|
||||
|
||||
@staticmethod
|
||||
def _fetch_teams(path: str) -> Dict[str, str]:
|
||||
"""ESPN's {abbreviation: display name} for a league.
|
||||
|
||||
``limit=1000`` is required: the default page size truncates the NCAA
|
||||
responses to roughly half their teams, which makes valid codes look wrong.
|
||||
"""
|
||||
import requests
|
||||
|
||||
payload = requests.get(TEAMS_URL.format(path=path),
|
||||
timeout=REQUEST_TIMEOUT).json()
|
||||
entries = payload['sports'][0]['leagues'][0]['teams']
|
||||
return {
|
||||
t['team']['abbreviation']: t['team']['displayName']
|
||||
for t in entries if t.get('team', {}).get('abbreviation')
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def _schedule_note(cls, path: str) -> Optional[str]:
|
||||
"""
|
||||
Why the league has nothing to show, as a clause, or ``None`` if it does.
|
||||
|
||||
Two things make this harder than reading ``events``:
|
||||
|
||||
* An out-of-season league does not come back empty. ESPN rolls the
|
||||
scoreboard forward to the next day that has fixtures, so in July the
|
||||
NHL endpoint returns seven September games. Emptiness cannot be the
|
||||
signal; the date of those games is, and it is more useful anyway.
|
||||
* A *finished* season rolls nowhere and returns its last game instead,
|
||||
months in the past — so dates have to be filtered to the future
|
||||
before the soonest one means anything.
|
||||
"""
|
||||
import requests
|
||||
|
||||
payload = requests.get(SCOREBOARD_URL.format(path=path),
|
||||
timeout=REQUEST_TIMEOUT).json()
|
||||
|
||||
event_dates = [cls._parse_date(e.get('date'))
|
||||
for e in payload.get('events') or []]
|
||||
calendar_dates = []
|
||||
for entry in (payload.get('leagues') or [{}])[0].get('calendar') or []:
|
||||
calendar_dates.append(cls._parse_date(
|
||||
entry if isinstance(entry, str) else entry.get('startDate')))
|
||||
|
||||
# Count the last day as current, rather than filtering on "later than
|
||||
# right now": a game that began a few hours ago still means the league
|
||||
# has something on, and dropping it would report a live slate as a
|
||||
# finished season. A day's grace also keeps this correct whatever the
|
||||
# user's timezone, since these timestamps are UTC.
|
||||
now = datetime.now(timezone.utc)
|
||||
|
||||
def future(candidates):
|
||||
return sorted(d for d in candidates if d and (now - d).days < 1)
|
||||
|
||||
# Events are fixtures; the calendar is week and phase boundaries,
|
||||
# which routinely open days before their first game (an NFL week 1
|
||||
# calendar entry starts the weekend before the Thursday opener).
|
||||
# Reading the two together reported the earliest boundary as a game
|
||||
# date -- "nothing on until 06 September" for a league whose first
|
||||
# snap is the 10th. The calendar only gets a say when the scoreboard
|
||||
# has no events at all to roll forward to: events that exist but are
|
||||
# all in the past mean the season is over, and an offseason calendar
|
||||
# phase must not be dressed up as its next game.
|
||||
if any(event_dates):
|
||||
upcoming = future(event_dates)
|
||||
else:
|
||||
upcoming = future(calendar_dates)
|
||||
if not upcoming:
|
||||
if not any(event_dates) and not any(calendar_dates):
|
||||
return None # Nothing published either way; draw no conclusion.
|
||||
return ("the season has finished and the next one's fixtures are "
|
||||
"not published yet")
|
||||
|
||||
# A day or two out is just an off day, and saying so would be noise.
|
||||
if (upcoming[0] - now).days < cls.GAP_DAYS:
|
||||
return None
|
||||
return "the league has nothing on until {}".format(
|
||||
upcoming[0].strftime('%d %B %Y'))
|
||||
|
||||
@staticmethod
|
||||
def _parse_date(raw) -> Optional[datetime]:
|
||||
if not raw or not isinstance(raw, str):
|
||||
return None
|
||||
try:
|
||||
return datetime.fromisoformat(raw.replace('Z', '+00:00'))
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def _suggest(cls, code: str, teams: Dict[str, str]) -> str:
|
||||
"""Nearest matching code for a typo, as a ready-to-log clause."""
|
||||
upper = (code or "").strip().upper()
|
||||
if not upper or code in teams:
|
||||
return ""
|
||||
|
||||
# Right code, wrong case — matching is case-sensitive. Guard on the case
|
||||
# actually differing, so a valid code never draws this message.
|
||||
for abbr in teams:
|
||||
if abbr.upper() == upper:
|
||||
return " Codes are case-sensitive; use {!r} ({}).".format(
|
||||
abbr, teams[abbr])
|
||||
|
||||
ranked = cls._rank(upper, (a for a, n in teams.items()
|
||||
if cls._abbreviates(upper, n)), teams)
|
||||
if not ranked:
|
||||
# Nicknames are often a fragment of a word rather than its initials:
|
||||
# 'BAMA' sits inside 'Alabama' but abbreviates nothing in it. Require
|
||||
# three characters, since shorter fragments match far too much.
|
||||
if len(upper) >= 3:
|
||||
ranked = cls._rank(
|
||||
upper,
|
||||
(a for a, n in teams.items()
|
||||
if any(upper in w for w in cls._words(n))),
|
||||
teams)
|
||||
|
||||
if len(ranked) == 1:
|
||||
return " Closest match is {!r} ({}).".format(
|
||||
ranked[0], teams[ranked[0]])
|
||||
if ranked:
|
||||
return " Did you mean {}?".format(", ".join(
|
||||
"{!r} ({})".format(a, teams[a]) for a in ranked[:3]))
|
||||
|
||||
# Otherwise fall back to similarity, against names before codes: a name
|
||||
# gives more characters to compare and so produces fewer ties.
|
||||
names = {n.upper(): a for a, n in teams.items()}
|
||||
hits = difflib.get_close_matches(upper, list(names), n=1, cutoff=0.6)
|
||||
if hits:
|
||||
abbr = names[hits[0]]
|
||||
return " Closest match is {!r} ({}).".format(abbr, teams[abbr])
|
||||
|
||||
code_hits = difflib.get_close_matches(upper, list(teams), n=1, cutoff=0.6)
|
||||
if code_hits:
|
||||
return " Closest match is {!r} ({}).".format(
|
||||
code_hits[0], teams[code_hits[0]])
|
||||
return ""
|
||||
|
||||
@staticmethod
|
||||
def _words(name: str):
|
||||
return [w for w in re.split(r'[^A-Za-z0-9]+', (name or '').upper()) if w]
|
||||
|
||||
@classmethod
|
||||
def _rank(cls, code: str, candidates, teams: Dict[str, str]):
|
||||
"""
|
||||
Order candidate codes best-first.
|
||||
|
||||
A code that picks up the *first* word of the name wins, because that is
|
||||
how people shorten team names: 'SCAR' for South Carolina starts at
|
||||
'South', whereas for Rutgers Scarlet Knights it starts mid-name. Without
|
||||
this the tie is broken alphabetically and the obvious answer can land
|
||||
third in the list.
|
||||
"""
|
||||
def key(abbr):
|
||||
words = cls._words(teams.get(abbr, ''))
|
||||
first_word_hit = bool(words) and words[0].startswith(code[:1])
|
||||
return (not first_word_hit, len(abbr), abbr)
|
||||
|
||||
return sorted(set(candidates), key=key)
|
||||
|
||||
@staticmethod
|
||||
def _abbreviates(code: str, name: str) -> bool:
|
||||
"""
|
||||
Whether ``code`` reads as an abbreviation of ``name``.
|
||||
|
||||
Each part of the code must be a prefix of one of the name's words, taken
|
||||
in order — which is how people actually shorten team names. Plain string
|
||||
similarity is no use for three-letter codes: 'MUN' scores identically
|
||||
against 'MAN' and 'SUN', so Manchester United and Sunderland tie and the
|
||||
suggestion is a coin flip. This rule separates them, because 'MUN'
|
||||
splits as M-anchester UN-ited while Sunderland has no word starting M.
|
||||
"""
|
||||
words = [w for w in re.split(r'[^A-Za-z0-9]+', (name or '').upper()) if w]
|
||||
|
||||
def consume(rest: str, remaining: List[str]) -> bool:
|
||||
if not rest:
|
||||
return True
|
||||
if not remaining:
|
||||
return False
|
||||
head, tail = remaining[0], remaining[1:]
|
||||
# Skip this word entirely, as in "Manchester United" -> "UTD".
|
||||
if consume(rest, tail):
|
||||
return True
|
||||
for size in range(1, min(len(rest), len(head)) + 1):
|
||||
if head.startswith(rest[:size]) and consume(rest[size:], tail):
|
||||
return True
|
||||
return False
|
||||
|
||||
return consume((code or "").strip().upper(), words)
|
||||
@@ -0,0 +1,262 @@
|
||||
"""Timezone resolution for the scoreboard plugins.
|
||||
|
||||
Game start times arrive from ESPN in UTC and have to be converted to the
|
||||
user's local zone before they are drawn. This module owns the "which zone?"
|
||||
decision so every part of a scoreboard (its scorebug, its scroll-mode game
|
||||
card and its plugin manager) agrees.
|
||||
|
||||
The scoreboards each carried a copy of this module as ``<sport>_timezone.py``.
|
||||
The copies were identical apart from two per-plugin values, which are
|
||||
keyword-only arguments here: ``plugin_label``, the name the final warning
|
||||
tells the user to open, and ``writeback_fixed_in`` (see below).
|
||||
|
||||
Resolution order, first valid wins:
|
||||
|
||||
1. ``timezone`` in the plugin's own config (explicit per-plugin override)
|
||||
2. The LEDMatrix global timezone via ``plugin_manager.config_manager``
|
||||
3. The LEDMatrix global timezone via ``cache_manager.config_manager``
|
||||
4. The host system's zone (``TZ``, ``/etc/timezone``, ``/etc/localtime``)
|
||||
5. UTC
|
||||
|
||||
Steps 2 and 3 matter because the core does not consistently hang
|
||||
``config_manager`` off both objects -- reading only one of them is what made
|
||||
a scoreboard fall through to UTC while the clock plugin (which checks
|
||||
``plugin_manager`` first) showed the right time on the same device. Step 4 is
|
||||
the backstop for cores that expose no ``config_manager`` at all: a Pi with its
|
||||
system clock set correctly should never end up rendering UTC.
|
||||
|
||||
Two traps this module exists to avoid, both of which render every start time
|
||||
in UTC on a correctly-configured device:
|
||||
|
||||
* **The stale ``"UTC"`` artifact.** Some plugins once wrote
|
||||
``"timezone": "UTC"`` into the *saved* config whenever resolution failed, and
|
||||
that write-back stuck -- thereafter shadowing the real global timezone. Such
|
||||
a plugin passes ``writeback_fixed_in`` (the release that fixed it), and step 1
|
||||
then treats a bare ``"UTC"`` as suspect: it is honored only when nothing
|
||||
downstream disagrees. ``Etc/UTC`` is the unambiguous spelling for "I really
|
||||
do want UTC"; the bug never produced it, so it is always honored. A plugin
|
||||
that never had the bug leaves ``writeback_fixed_in`` as ``None``, and its
|
||||
plugin-level ``"UTC"`` is honored verbatim.
|
||||
* **``get_timezone()``'s own default.** The core's
|
||||
``ConfigManager.get_timezone()`` is ``self.config.get('timezone', 'UTC')``, so
|
||||
it hands back ``"UTC"`` for a config that simply has no ``timezone`` key.
|
||||
Steps 2 and 3 read the raw config dict instead, so an absent key falls
|
||||
through to the system zone rather than latching onto that default.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
from typing import Any, Dict, Iterable, Optional, Tuple
|
||||
|
||||
import pytz
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _from_config_manager(config_manager: Any, log: logging.Logger) -> Optional[str]:
|
||||
"""Pull the global timezone out of a core ConfigManager, if it has one.
|
||||
|
||||
Reads the raw config dict in preference to ``get_timezone()``. The core's
|
||||
``ConfigManager.get_timezone()`` is ``self.config.get('timezone', 'UTC')`` --
|
||||
it substitutes its own ``"UTC"`` when the key is absent, which is
|
||||
indistinguishable from the user deliberately choosing UTC. Taking that at
|
||||
face value would mask a missing global setting and stop resolution ever
|
||||
reaching the host system zone. So: if the raw config is readable and has no
|
||||
``timezone`` key, report "nothing here" and let the caller fall through.
|
||||
``get_timezone()`` is only consulted for cores that expose no raw config.
|
||||
"""
|
||||
if config_manager is None:
|
||||
return None
|
||||
|
||||
raw_readable = False
|
||||
for loader_name in ("get_config", "load_config"):
|
||||
loader = getattr(config_manager, loader_name, None)
|
||||
if not callable(loader):
|
||||
continue
|
||||
try:
|
||||
main_config = loader()
|
||||
except Exception:
|
||||
log.debug("config_manager.%s() failed", loader_name, exc_info=True)
|
||||
continue
|
||||
if isinstance(main_config, dict):
|
||||
raw_readable = True
|
||||
name: Optional[str] = main_config.get("timezone")
|
||||
if name:
|
||||
return name
|
||||
|
||||
if raw_readable:
|
||||
return None
|
||||
|
||||
getter = getattr(config_manager, "get_timezone", None)
|
||||
if callable(getter):
|
||||
try:
|
||||
name = getter()
|
||||
if name:
|
||||
return name
|
||||
except Exception:
|
||||
log.debug("config_manager.get_timezone() failed", exc_info=True)
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def system_timezone_name() -> Optional[str]:
|
||||
"""Best-effort IANA name for the host's configured timezone."""
|
||||
name = os.environ.get("TZ")
|
||||
if name:
|
||||
return name
|
||||
|
||||
# Debian / Raspberry Pi OS record the zone name here.
|
||||
try:
|
||||
with open("/etc/timezone", "r", encoding="utf-8") as handle:
|
||||
name = handle.read().strip()
|
||||
if name:
|
||||
return name
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
# Otherwise /etc/localtime is a symlink into the zoneinfo tree.
|
||||
try:
|
||||
path = os.path.realpath("/etc/localtime")
|
||||
marker = "zoneinfo" + os.sep
|
||||
if marker in path:
|
||||
return path.split(marker, 1)[1]
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _validated(name: object, source: str, log: logging.Logger) -> Optional[str]:
|
||||
"""Return a usable IANA name, or None if blank/absent/not a real zone."""
|
||||
if not isinstance(name, str):
|
||||
return None
|
||||
name = name.strip()
|
||||
if not name:
|
||||
return None
|
||||
try:
|
||||
pytz.timezone(name)
|
||||
except pytz.UnknownTimeZoneError:
|
||||
log.warning("Ignoring invalid timezone %r from %s", name, source)
|
||||
return None
|
||||
except Exception:
|
||||
# Not an unknown-zone error, so something else went wrong inside pytz.
|
||||
# Log it loudly rather than silently reclassifying it as "invalid" --
|
||||
# but still don't propagate: this runs in the render path, and a
|
||||
# mislabelled zone beats taking the whole display down.
|
||||
log.warning(
|
||||
"Unexpected error validating timezone %r from %s; ignoring it",
|
||||
name, source, exc_info=True,
|
||||
)
|
||||
return None
|
||||
return name
|
||||
|
||||
|
||||
def resolve_timezone_name(
|
||||
config: Optional[Dict[str, Any]] = None,
|
||||
plugin_manager: Any = None,
|
||||
cache_manager: Any = None,
|
||||
log: Optional[logging.Logger] = None,
|
||||
*,
|
||||
plugin_label: str,
|
||||
writeback_fixed_in: Optional[str] = None,
|
||||
) -> str:
|
||||
"""Return the IANA timezone name to render game times in.
|
||||
|
||||
Never raises and never returns an empty string; falls back to ``"UTC"``
|
||||
only when every source is missing or invalid.
|
||||
|
||||
``plugin_label`` names the plugin in the warning logged when nothing
|
||||
resolves (e.g. ``"hockey scoreboard"``). ``writeback_fixed_in`` is the
|
||||
plugin release that stopped writing ``"UTC"`` back into the saved config,
|
||||
for a plugin that ever did; ``None`` (the default) otherwise.
|
||||
"""
|
||||
log = log or logger
|
||||
|
||||
def downstream():
|
||||
"""Yield (source, name) for everything except the plugin's own config.
|
||||
|
||||
Lazy: ``SportsCore._get_timezone()`` runs this once per game and the
|
||||
answer is almost always already in the plugin config, so evaluating on
|
||||
demand keeps the common case from calling into both config managers and
|
||||
stat-ing the host timezone files every time.
|
||||
"""
|
||||
yield (
|
||||
"plugin_manager.config_manager",
|
||||
_from_config_manager(getattr(plugin_manager, "config_manager", None), log),
|
||||
)
|
||||
yield (
|
||||
"cache_manager.config_manager",
|
||||
_from_config_manager(getattr(cache_manager, "config_manager", None), log),
|
||||
)
|
||||
yield "system timezone", system_timezone_name()
|
||||
|
||||
def first_valid(
|
||||
sources: Iterable[Tuple[str, Any]],
|
||||
) -> Tuple[Optional[str], Optional[str]]:
|
||||
for source, name in sources:
|
||||
name = _validated(name, source, log)
|
||||
if name:
|
||||
return source, name
|
||||
return None, None
|
||||
|
||||
plugin_value = _validated((config or {}).get("timezone"), "plugin config", log)
|
||||
|
||||
if writeback_fixed_in is not None and plugin_value and plugin_value.lower() == "utc":
|
||||
# Before writeback_fixed_in this plugin wrote "timezone": "UTC" into
|
||||
# the saved config whenever it failed to resolve a global timezone, and
|
||||
# that write-back persisted. A bare "UTC" is therefore far more likely
|
||||
# to be that artifact than a deliberate choice -- it only ever appeared
|
||||
# on failure. Honor it only when nothing downstream disagrees; a user
|
||||
# who genuinely wants UTC writes the unambiguous "Etc/UTC", which the
|
||||
# bug never produced and which falls through to the normal path below.
|
||||
source, downstream_name = first_valid(downstream())
|
||||
if downstream_name and downstream_name.lower() not in ("utc", "etc/utc"):
|
||||
log.warning(
|
||||
"Ignoring the plugin-level timezone 'UTC': it is almost "
|
||||
"certainly left over from the write-back bug fixed in %s, and "
|
||||
"%s says %s. Using %s. If you really do want UTC here, set "
|
||||
"this plugin's timezone to 'Etc/UTC' instead.",
|
||||
writeback_fixed_in, source, downstream_name, downstream_name,
|
||||
)
|
||||
return downstream_name
|
||||
log.debug("Plugin-level timezone 'UTC' agrees with %s; using UTC", source or "no other source")
|
||||
return "UTC"
|
||||
|
||||
if plugin_value:
|
||||
log.debug("Resolved timezone %s from plugin config", plugin_value)
|
||||
return plugin_value
|
||||
|
||||
source, name = first_valid(downstream())
|
||||
if name:
|
||||
log.debug("Resolved timezone %s from %s", name, source)
|
||||
return name
|
||||
|
||||
log.warning(
|
||||
"Could not determine a timezone from the plugin config, the LEDMatrix "
|
||||
"config or the system; game times will be shown in UTC. Set a timezone "
|
||||
"in the %s's Advanced Settings to override.",
|
||||
plugin_label,
|
||||
)
|
||||
return "UTC"
|
||||
|
||||
|
||||
def resolve_timezone(
|
||||
config: Optional[Dict[str, Any]] = None,
|
||||
plugin_manager: Any = None,
|
||||
cache_manager: Any = None,
|
||||
log: Optional[logging.Logger] = None,
|
||||
*,
|
||||
plugin_label: str,
|
||||
writeback_fixed_in: Optional[str] = None,
|
||||
):
|
||||
"""``resolve_timezone_name`` as a ready-to-use tzinfo object."""
|
||||
return pytz.timezone(
|
||||
resolve_timezone_name(
|
||||
config=config,
|
||||
plugin_manager=plugin_manager,
|
||||
cache_manager=cache_manager,
|
||||
log=log,
|
||||
plugin_label=plugin_label,
|
||||
writeback_fixed_in=writeback_fixed_in,
|
||||
)
|
||||
)
|
||||
Reference in New Issue
Block a user