mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-04 22:35:08 +00:00
* ci: mypy ratchet -- keep type-clean modules clean mypy-clean.txt lists the 71 modules under src/ that type-check clean; scripts/check_types.py runs mypy (--follow-imports=silent) on exactly those files and fails on any error or a missing/unsorted/duplicate entry. A new "Type check (mypy ratchet)" CI job runs it with mypy 1.20.2 and pinned stubs; the manual pre-commit mypy hook now runs the same script (a local hook, so mypy sees the installed requirements like CI does). 35 modules were made clean with annotation-only fixes: hints, typing.cast, TYPE_CHECKING imports, implicit-Optional defaults made explicit, and annotations widened (never guards removed) where mypy called a defensive isinstance check unreachable. No runtime behaviour change. mypy.ini: numpy and orjson are treated as Any (follow_imports=skip, also for stubs). numpy 2.3+ stubs use 3.12 `type` statements that mypy won't parse at python_version 3.10, and orjson is optional, so seeing its stubs made the result depend on whether it was installed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * chore: annotate check_types.py's list-form mypy subprocess Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
472 lines
20 KiB
Python
472 lines
20 KiB
Python
"""
|
|
Logo Helper
|
|
|
|
Handles logo loading, caching, resizing, and management for LED matrix displays.
|
|
Extracted from LEDMatrix core to provide reusable functionality for plugins.
|
|
"""
|
|
|
|
import logging
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional, Tuple, Union
|
|
|
|
import requests
|
|
from PIL import Image, ImageDraw
|
|
from src.common.api_helper import USER_AGENT
|
|
from src.common.permission_utils import (
|
|
ensure_directory_permissions,
|
|
get_assets_dir_mode,
|
|
)
|
|
|
|
# How long a missing logo stays remembered as missing.
|
|
#
|
|
# This was 600s, and measured on a live rig that turned out to suppress nothing:
|
|
# the display rotation is ~618s, so every recheck landed just as the plugin came
|
|
# round again and the warning rate was unchanged at ~6/hour. A TTL has to be long
|
|
# relative to the loop that does the asking, not merely "a while".
|
|
#
|
|
# An hour is safe because the TTL is not the main way an entry clears. A download
|
|
# through load_logo_with_download() drops it immediately, and clear_cache() drops
|
|
# all of them; the TTL only covers a file that appeared some other way -- someone
|
|
# copying one in by hand. Waiting up to an hour for that, or restarting, is a fair
|
|
# trade for not re-warning about a file nobody is going to add.
|
|
MISSING_LOGO_RECHECK_SECONDS = 3600.0
|
|
|
|
# Well above any real team logo; bounds what a remote URL can write to disk.
|
|
# The cap for every logo download: src.logo_downloader.fetch_logo uses it too.
|
|
MAX_LOGO_BYTES = 10 * 1024 * 1024
|
|
|
|
#: A logo's default bounding box, as a multiple of the panel's width and
|
|
#: height, when the caller gives no max_width / max_height.
|
|
DEFAULT_LOGO_BOX_FACTOR = 1.5
|
|
|
|
|
|
class LogoHelper:
|
|
"""
|
|
Helper class for logo loading, caching, and resizing.
|
|
|
|
Provides functionality for:
|
|
- Loading logos from files
|
|
- Caching loaded logos in memory
|
|
- Resizing logos to fit display dimensions
|
|
- Handling logo variations and fallbacks
|
|
- Downloading missing logos from URLs
|
|
"""
|
|
|
|
def __init__(self, display_width: int, display_height: int,
|
|
cache_size: int = 100, logger: Optional[logging.Logger] = None):
|
|
"""
|
|
Initialize the LogoHelper.
|
|
|
|
Args:
|
|
display_width: Width of the LED matrix display
|
|
display_height: Height of the LED matrix display
|
|
cache_size: Maximum number of logos to cache in memory
|
|
logger: Optional logger instance
|
|
"""
|
|
self.display_width = display_width
|
|
self.display_height = display_height
|
|
self.cache_size = cache_size
|
|
self.logger = logger or logging.getLogger(__name__)
|
|
|
|
# In-memory logo cache
|
|
self._logo_cache: Dict[str, Image.Image] = {}
|
|
self._cache_order: List[str] = [] # For LRU cache management
|
|
|
|
# Misses, so an absent file is stat'd and warned about once rather than
|
|
# on every call. Without this a permanently missing logo produced a
|
|
# warning per rotation forever -- measured at 114 lines in 24 hours for
|
|
# a single missing ticker icon, for a file nobody was going to add.
|
|
# Time-bounded rather than permanent so a logo that appears later (the
|
|
# downloader writes them at runtime) is still picked up.
|
|
self._missing_logos: Dict[str, float] = {}
|
|
|
|
# Failed downloads by logo path. A logo that is absent (not a stale
|
|
# placeholder) has no on-disk timestamp to back off on, so without this
|
|
# every call retried the download -- up to a 30s timeout each time.
|
|
self._download_failures: Dict[str, float] = {}
|
|
|
|
# Session for HTTP requests
|
|
self.session = requests.Session()
|
|
self.session.headers.update({
|
|
'User-Agent': USER_AGENT,
|
|
'Accept': 'image/*',
|
|
})
|
|
|
|
def load_logo(self, team_abbr: str, logo_path: Union[str, Path],
|
|
max_width: Optional[int] = None,
|
|
max_height: Optional[int] = None,
|
|
scale: float = 1.0) -> Optional[Image.Image]:
|
|
"""
|
|
Load and resize a team logo.
|
|
|
|
Args:
|
|
team_abbr: Team abbreviation for caching
|
|
logo_path: Path to the logo file
|
|
max_width: Maximum width (default display_width *
|
|
DEFAULT_LOGO_BOX_FACTOR)
|
|
max_height: Maximum height (default display_height *
|
|
DEFAULT_LOGO_BOX_FACTOR)
|
|
scale: User's size multiplier for this image, from
|
|
``customization.layout.<element>.scale``; 1.0 leaves the box
|
|
as is. Callers hold the config, so they resolve the element
|
|
name; this only applies the number, clamped to
|
|
src.element_style's MIN_ELEMENT_SCALE..MAX_ELEMENT_SCALE.
|
|
A value that is not a finite positive number means 1.0.
|
|
|
|
Returns:
|
|
PIL Image object or None if loading fails
|
|
|
|
Note: for new adaptive-layout code prefer ``BasePlugin.draw_image``
|
|
/ ``LayoutContext.fit_image`` (src/adaptive_images.py) for the
|
|
fitting step — LogoHelper remains useful for its download and
|
|
placeholder logic.
|
|
"""
|
|
# Resolve the effective target size BEFORE the cache lookup so the
|
|
# key is size-qualified — a panel-size change must not return a
|
|
# logo resized for the old dimensions.
|
|
max_width, max_height, scale = self._scaled_box(max_width, max_height, scale)
|
|
# The key carries the scaled box, so two elements scaled differently
|
|
# cannot be served each other's image.
|
|
cache_key = f"{team_abbr}_{logo_path}_{max_width}x{max_height}"
|
|
if cache_key in self._logo_cache:
|
|
self.logger.debug(f"Using cached logo for {team_abbr}")
|
|
# Update LRU order (move to end)
|
|
if cache_key in self._cache_order:
|
|
self._cache_order.remove(cache_key)
|
|
self._cache_order.append(cache_key)
|
|
return self._logo_cache[cache_key]
|
|
|
|
# A known-missing file: skip the stat and stay quiet until the entry
|
|
# ages out. Checked after the positive cache so a logo that has since
|
|
# been loaded always wins.
|
|
missed_at = self._missing_logos.get(cache_key)
|
|
if missed_at is not None:
|
|
if time.time() - missed_at < MISSING_LOGO_RECHECK_SECONDS:
|
|
return None
|
|
del self._missing_logos[cache_key]
|
|
|
|
try:
|
|
logo_path = Path(logo_path)
|
|
if not logo_path.exists():
|
|
self._missing_logos[cache_key] = time.time()
|
|
self.logger.warning(f"Logo not found for {team_abbr} at {logo_path}")
|
|
return None
|
|
|
|
# Load image
|
|
logo: Image.Image = Image.open(logo_path)
|
|
if logo.mode != 'RGBA':
|
|
logo = logo.convert('RGBA')
|
|
|
|
# Resize if needed
|
|
logo = self._resize_logo(logo, max_width, max_height,
|
|
allow_upscale=scale > 1.0)
|
|
|
|
# Cache the logo
|
|
self._cache_logo(cache_key, logo)
|
|
|
|
self.logger.debug(f"Loaded logo for {team_abbr} from {logo_path}")
|
|
return logo
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error loading logo for {team_abbr}: {e}")
|
|
return None
|
|
|
|
def load_logo_with_download(self, team_abbr: str, logo_path: Union[str, Path],
|
|
logo_url: Optional[str] = None,
|
|
max_width: Optional[int] = None,
|
|
max_height: Optional[int] = None,
|
|
scale: float = 1.0) -> Optional[Image.Image]:
|
|
"""
|
|
Load logo with automatic download if missing.
|
|
|
|
Args:
|
|
team_abbr: Team abbreviation
|
|
logo_path: Local path to store/load logo
|
|
logo_url: URL to download logo from if local file missing
|
|
max_width: Maximum width for resizing
|
|
max_height: Maximum height for resizing
|
|
|
|
Returns:
|
|
PIL Image object or None if loading fails
|
|
"""
|
|
logo_path = Path(logo_path)
|
|
|
|
# Try to load existing logo first. A placeholder written by a previous
|
|
# failed download does not count: it wears the real logo's filename, so
|
|
# 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,
|
|
scale)
|
|
|
|
# Download if URL provided and file doesn't exist, unless the last
|
|
# attempt for this path failed recently.
|
|
failed_at = self._download_failures.get(str(logo_path))
|
|
if (logo_url and failed_at is not None
|
|
and time.time() - failed_at < MISSING_LOGO_RECHECK_SECONDS):
|
|
logo_url = None
|
|
if logo_url:
|
|
try:
|
|
self.logger.info(f"Downloading logo for {team_abbr} from {logo_url}")
|
|
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,
|
|
scale)
|
|
except Exception as e:
|
|
self.logger.error(f"Failed to download logo for {team_abbr}: {e}")
|
|
self._download_failures[str(logo_path)] = time.time()
|
|
# 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. Sized to the same scaled box
|
|
# a real logo gets, so a scaled element doesn't jump in size while
|
|
# its logo is missing.
|
|
box_width, box_height, _ = self._scaled_box(max_width, max_height, scale)
|
|
return self._create_placeholder_logo(team_abbr, box_width, box_height)
|
|
|
|
def _scaled_box(self, max_width: Optional[int], max_height: Optional[int],
|
|
scale: float) -> Tuple[int, int, float]:
|
|
"""The logo box after defaults and the user's scale are applied.
|
|
|
|
Returns ``(width, height, coerced_scale)``.
|
|
"""
|
|
if max_width is None:
|
|
max_width = int(self.display_width * DEFAULT_LOGO_BOX_FACTOR)
|
|
if max_height is None:
|
|
max_height = int(self.display_height * DEFAULT_LOGO_BOX_FACTOR)
|
|
# Imported here: src.element_style imports src.common (for bdf_font),
|
|
# whose __init__ imports this module.
|
|
from src.element_style import coerce_scale
|
|
scale = coerce_scale(scale, 1.0)
|
|
if scale != 1.0:
|
|
max_width = max(1, int(round(max_width * scale)))
|
|
max_height = max(1, int(round(max_height * scale)))
|
|
return max_width, max_height, scale
|
|
|
|
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)
|
|
# The file exists now, so any record of it being missing is wrong --
|
|
# and load_logo() consults that record before it stats the disk, so
|
|
# leaving it would hide a logo we just downloaded.
|
|
for key in [k for k in self._missing_logos if k.startswith(prefix)]:
|
|
del self._missing_logos[key]
|
|
self._download_failures.pop(str(logo_path), None)
|
|
|
|
@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]:
|
|
"""
|
|
Get possible filename variations for a team abbreviation.
|
|
|
|
Args:
|
|
team_abbr: Team abbreviation
|
|
|
|
Returns:
|
|
List of possible filename variations
|
|
"""
|
|
variations = [team_abbr]
|
|
|
|
# Common variations
|
|
if '&' in team_abbr:
|
|
variations.append(team_abbr.replace('&', 'AND'))
|
|
if 'AND' in team_abbr:
|
|
variations.append(team_abbr.replace('AND', '&'))
|
|
|
|
# Handle special cases
|
|
special_cases = {
|
|
'TA&M': ['TAMU', 'TEXASAM'],
|
|
'UCLA': ['UCLA'],
|
|
'USC': ['USC'],
|
|
'LSU': ['LSU'],
|
|
}
|
|
|
|
if team_abbr in special_cases:
|
|
variations.extend(special_cases[team_abbr])
|
|
|
|
return variations
|
|
|
|
def normalize_abbreviation(self, team_abbr: str) -> str:
|
|
"""
|
|
Normalize team abbreviation for consistent filename usage.
|
|
|
|
NOTE: this deliberately differs from
|
|
LogoDownloader.normalize_abbreviation (src/logo_downloader.py),
|
|
which replaces filesystem-unsafe characters (/ \\ : * ? " < > |)
|
|
but does not strip spaces. Plugins call the LogoDownloader
|
|
version; changing either implementation changes which logo
|
|
filenames resolve on existing installs.
|
|
|
|
Args:
|
|
team_abbr: Raw team abbreviation
|
|
|
|
Returns:
|
|
Normalized abbreviation
|
|
"""
|
|
# Remove spaces and convert to uppercase
|
|
normalized = team_abbr.strip().upper()
|
|
|
|
# Handle special characters
|
|
normalized = normalized.replace('&', 'AND')
|
|
normalized = normalized.replace(' ', '')
|
|
|
|
return normalized
|
|
|
|
def clear_cache(self) -> None:
|
|
"""Clear the logo cache."""
|
|
self._logo_cache.clear()
|
|
self._cache_order.clear()
|
|
self._missing_logos.clear()
|
|
self._download_failures.clear()
|
|
self.logger.debug("Logo cache cleared")
|
|
|
|
def get_cache_stats(self) -> Dict[str, float]:
|
|
"""
|
|
Get cache statistics.
|
|
|
|
Returns:
|
|
Dictionary with cache statistics
|
|
"""
|
|
return {
|
|
'cached_logos': len(self._logo_cache),
|
|
'cache_size_limit': self.cache_size,
|
|
'cache_usage_percent': (
|
|
(len(self._logo_cache) / self.cache_size) * 100
|
|
if self.cache_size else 0
|
|
),
|
|
}
|
|
|
|
def _resize_logo(self, logo: Image.Image, max_width: Optional[int] = None,
|
|
max_height: Optional[int] = None,
|
|
allow_upscale: bool = False) -> Image.Image:
|
|
"""Resize logo to fit display dimensions.
|
|
|
|
``allow_upscale`` is only set when the user asked for a scale above 1:
|
|
the fit rule is "never larger than the box", and growing an image
|
|
nobody asked to grow would change every existing render.
|
|
"""
|
|
if max_width is None:
|
|
max_width = int(self.display_width * DEFAULT_LOGO_BOX_FACTOR)
|
|
if max_height is None:
|
|
max_height = int(self.display_height * DEFAULT_LOGO_BOX_FACTOR)
|
|
|
|
# Only resize if necessary
|
|
if logo.width <= max_width and logo.height <= max_height:
|
|
if not allow_upscale or not logo.width or not logo.height:
|
|
return logo
|
|
ratio = min(max_width / logo.width, max_height / logo.height)
|
|
if ratio <= 1:
|
|
return logo
|
|
return logo.resize((max(1, int(logo.width * ratio)),
|
|
max(1, int(logo.height * ratio))),
|
|
Image.Resampling.LANCZOS)
|
|
|
|
# Maintain aspect ratio
|
|
logo.thumbnail((max_width, max_height), Image.Resampling.LANCZOS)
|
|
return logo
|
|
|
|
def _cache_logo(self, cache_key: str, logo: Image.Image) -> None:
|
|
"""Cache a logo with LRU eviction."""
|
|
# Remove oldest if cache is full
|
|
if len(self._logo_cache) >= self.cache_size:
|
|
if self._cache_order:
|
|
oldest_key = self._cache_order.pop(0)
|
|
del self._logo_cache[oldest_key]
|
|
|
|
# Add to cache
|
|
self._logo_cache[cache_key] = logo
|
|
self._cache_order.append(cache_key)
|
|
|
|
def _download_logo(self, url: str, file_path: Path) -> None:
|
|
"""Download a logo from ``url`` to ``file_path``; raises on failure.
|
|
|
|
Delegates to ``src.logo_downloader.fetch_logo``, the same hardened
|
|
download the scoreboard plugins use: streamed and capped at
|
|
``MAX_LOGO_BYTES``, ``image/*`` only, decoded by Pillow, stored as an
|
|
RGBA PNG, and moved into place atomically -- a failure leaves neither
|
|
a partial file nor a temp file. Uses this helper's own session.
|
|
|
|
Imported lazily: src.logo_downloader imports src.common, so a
|
|
module-level import here would be circular.
|
|
"""
|
|
from src.logo_downloader import fetch_logo
|
|
|
|
# Ensure directory exists with proper permissions
|
|
ensure_directory_permissions(file_path.parent, get_assets_dir_mode())
|
|
fetch_logo(self.session, url, file_path, timeout=30,
|
|
max_bytes=MAX_LOGO_BYTES)
|
|
self.logger.debug(f"Downloaded logo to {file_path}")
|
|
|
|
def _create_placeholder_logo(self, team_abbr: str,
|
|
max_width: Optional[int] = None,
|
|
max_height: Optional[int] = None) -> Optional[Image.Image]:
|
|
"""
|
|
A stand-in for a logo that could not be loaded or downloaded: a
|
|
translucent grey box with a light outline, filling the logo box.
|
|
No text is drawn; ``team_abbr`` is only used in log messages.
|
|
|
|
Args:
|
|
team_abbr: Team the placeholder stands in for
|
|
max_width: Width (default display_width * DEFAULT_LOGO_BOX_FACTOR)
|
|
max_height: Height (default display_height * DEFAULT_LOGO_BOX_FACTOR)
|
|
|
|
Returns:
|
|
The RGBA placeholder, or None if it could not be created
|
|
"""
|
|
try:
|
|
if max_width is None:
|
|
max_width = int(self.display_width * DEFAULT_LOGO_BOX_FACTOR)
|
|
if max_height is None:
|
|
max_height = int(self.display_height * DEFAULT_LOGO_BOX_FACTOR)
|
|
|
|
placeholder = Image.new('RGBA', (max_width, max_height), (0, 0, 0, 0))
|
|
draw = ImageDraw.Draw(placeholder)
|
|
draw.rectangle([0, 0, max_width-1, max_height-1],
|
|
fill=(100, 100, 100, 200), outline=(200, 200, 200, 255))
|
|
|
|
self.logger.debug(f"Created placeholder logo for {team_abbr}")
|
|
return placeholder
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error creating placeholder for {team_abbr}: {e}")
|
|
return None
|