mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-04 14:25:08 +00:00
ChuckandClaude Opus 5.5
c00bf5e8e6
* fix(plugin-system): unload/update race, failed-load module cleanup, limits validation, schema lookup, install rollback, op-queue dedupe - unload_plugin takes the per-plugin lock (5s bounded) before cleanup(), and an update() that finishes after its plugin was unloaded no longer sets the state back to ENABLED. - A load that fails after import drops plugin_<id> and its submodules and forgets its manager fonts, so a fixed plugin reloads new code. - Resource limits are validated as non-negative numbers: 400 at POST /plugins/limits, bad cached records ignored with one warning. Route docstrings note health/metrics reset and limits only change the web process's view. - SchemaManager.get_schema_path resolves each search dir via resolve_plugin_dir (manifest id, ledmatrix-<id>) before the literal paths; plugins/ still before plugin-repos/. Misses cached 30s and logged once at DEBUG. - install_from_url sets an existing copy aside and restores it if the move fails, under the per-plugin reinstall lock. - Operation queue refuses a second pending op for a plugin and trims _operations with history. - get_vegas_render_width reads display_manager.width first. - get_logger in store/schema/health/resource/saved_repositories; UTF-8 reads in store_manager and state_manager. - Docs: update_interval precedence (manifest over config) stated where users are told to set it in config. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(web): build the limits 400 message from the field name, not an exception CodeQL flagged str(e) flowing into the response. invalid_limit_field() returns the offending field without raising, and limits_from_dict uses it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1082 lines
44 KiB
Python
1082 lines
44 KiB
Python
"""
|
|
Base Plugin Interface
|
|
|
|
All LEDMatrix plugins must inherit from BasePlugin and implement
|
|
the required abstract methods: update() and display().
|
|
|
|
API Version: 1.0.0
|
|
Stability: Stable - maintains backward compatibility
|
|
"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from enum import Enum
|
|
from typing import Dict, Any, Optional, List
|
|
import os
|
|
import sys
|
|
from src.logging_config import get_logger
|
|
|
|
|
|
_shared_fallback_font_manager: Optional[Any] = None
|
|
|
|
#: Distinguishes "not looked up yet" from "looked up and not found", so a
|
|
#: plugin with no schema does not re-scan the disk on every frame.
|
|
_UNSET_SCHEMA_PATH = object()
|
|
|
|
|
|
class _NullStyleResolver:
|
|
"""Stand-in for ElementStyleResolver when the module is unavailable.
|
|
|
|
Only reachable on a core that predates src.element_style, which
|
|
``styles`` degrades to rather than raising: every lookup returns the
|
|
caller's classic values, which is what the plugin drew before styling
|
|
existed.
|
|
"""
|
|
|
|
def __init__(self, config: Any) -> None:
|
|
self._config = config
|
|
|
|
def style(self, element_key: str, classic_font: str = None,
|
|
classic_size: int = 8, classic_color: Any = None,
|
|
mode: Optional[str] = None) -> Any:
|
|
from types import SimpleNamespace
|
|
return SimpleNamespace(
|
|
font=None, color=classic_color or (255, 255, 255), offset=(0, 0),
|
|
font_name=classic_font, font_size=classic_size,
|
|
user_forced=False, user_forced_color=False,
|
|
visible=True, align=None, scale=1.0)
|
|
|
|
def offset(self, element_key: str, mode: Optional[str] = None) -> tuple:
|
|
return (0, 0)
|
|
|
|
def offset_value(self, element_key: str, axis: str, default: int = 0,
|
|
mode: Optional[str] = None) -> int:
|
|
return default
|
|
|
|
|
|
def _fallback_font_manager() -> Any:
|
|
"""Shared FontManager for environments (unit tests, mocks) where the
|
|
plugin manager doesn't carry one. Scans assets/fonts like the real one."""
|
|
global _shared_fallback_font_manager
|
|
if _shared_fallback_font_manager is None:
|
|
from src.font_manager import FontManager
|
|
_shared_fallback_font_manager = FontManager({})
|
|
return _shared_fallback_font_manager
|
|
|
|
|
|
class VegasDisplayMode(Enum):
|
|
"""
|
|
Display mode for Vegas scroll integration.
|
|
|
|
Determines how a plugin's content behaves within the continuous scroll:
|
|
|
|
- SCROLL: Content scrolls continuously within the stream.
|
|
Best for multi-item plugins like sports scores, odds tickers, news feeds.
|
|
Plugin provides multiple frames via get_vegas_content().
|
|
|
|
- FIXED_SEGMENT: Content is a fixed-width block that scrolls BY with
|
|
the rest of the content. Best for static info like clock, weather.
|
|
Plugin provides a single image sized to vegas_panel_count panels.
|
|
|
|
- STATIC: Scroll pauses, plugin displays for its duration, then scroll
|
|
resumes. Best for important alerts or detailed views that need attention.
|
|
Plugin uses standard display() method during the pause.
|
|
"""
|
|
SCROLL = "scroll"
|
|
FIXED_SEGMENT = "fixed"
|
|
STATIC = "static"
|
|
|
|
|
|
class BasePlugin(ABC):
|
|
"""
|
|
Base class that all plugins must inherit from.
|
|
Provides standard interface and helper methods.
|
|
|
|
This is the core plugin interface that all plugins must implement.
|
|
Provides common functionality for logging, configuration, and
|
|
integration with the LEDMatrix core system.
|
|
"""
|
|
|
|
API_VERSION = "1.0.0"
|
|
|
|
#: Which ``customization.modes.<mode>`` overrides :attr:`styles` applies.
|
|
#: A plugin with one instance per display mode (the scoreboards' live /
|
|
#: upcoming / recent classes) sets this and every existing style lookup
|
|
#: becomes mode-aware without changing a call site.
|
|
STYLE_MODE: Optional[str] = None
|
|
|
|
def __init__(
|
|
self,
|
|
plugin_id: str,
|
|
config: Dict[str, Any],
|
|
display_manager: Any,
|
|
cache_manager: Any,
|
|
plugin_manager: Any,
|
|
) -> None:
|
|
"""
|
|
Standard initialization for all plugins.
|
|
|
|
Args:
|
|
plugin_id: Unique identifier for this plugin instance
|
|
config: Plugin-specific configuration dictionary
|
|
display_manager: Shared display manager instance for rendering
|
|
cache_manager: Shared cache manager instance for data persistence
|
|
plugin_manager: Reference to plugin manager for inter-plugin communication
|
|
"""
|
|
self.plugin_id: str = plugin_id
|
|
self.config: Dict[str, Any] = config
|
|
self.display_manager: Any = display_manager
|
|
self.cache_manager: Any = cache_manager
|
|
self.plugin_manager: Any = plugin_manager
|
|
# get_logger returns a PluginLoggerAdapter here (plugin_id given), which
|
|
# stamps every record with plugin_id so it survives into formatted output.
|
|
self.logger = get_logger(f"plugin.{plugin_id}", plugin_id=plugin_id)
|
|
self.enabled: bool = config.get("enabled", True)
|
|
|
|
self.logger.info("Initialized plugin: %s", plugin_id)
|
|
|
|
@abstractmethod
|
|
def update(self) -> None:
|
|
"""
|
|
Fetch/update data for this plugin.
|
|
|
|
This method is called based on update_interval specified in the
|
|
plugin's manifest. It should fetch any necessary data from APIs,
|
|
databases, or other sources and prepare it for display.
|
|
|
|
Use the cache_manager for caching API responses to avoid
|
|
excessive requests.
|
|
|
|
Example:
|
|
def update(self):
|
|
cache_key = f"{self.plugin_id}_data"
|
|
cached = self.cache_manager.get(cache_key, max_age=3600)
|
|
if cached:
|
|
self.data = cached
|
|
return
|
|
|
|
self.data = self._fetch_from_api()
|
|
self.cache_manager.set(cache_key, self.data)
|
|
"""
|
|
raise NotImplementedError("Plugins must implement update()")
|
|
|
|
@abstractmethod
|
|
def display(self, force_clear: bool = False) -> None:
|
|
"""
|
|
Render this plugin's display.
|
|
|
|
This method is called during the display rotation or when the plugin
|
|
is explicitly requested to render. It should use the display_manager
|
|
to draw content on the LED matrix.
|
|
|
|
Args:
|
|
force_clear: If True, clear display before rendering
|
|
|
|
Example:
|
|
def display(self, force_clear=False):
|
|
if force_clear:
|
|
self.display_manager.clear()
|
|
|
|
self.display_manager.draw_text(
|
|
"Hello, World!",
|
|
x=5, y=15,
|
|
color=(255, 255, 255)
|
|
)
|
|
|
|
self.display_manager.update_display()
|
|
"""
|
|
raise NotImplementedError("Plugins must implement display()")
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Global (whole-device) configuration
|
|
# -------------------------------------------------------------------------
|
|
@property
|
|
def global_config(self) -> Dict[str, Any]:
|
|
"""
|
|
The full LEDMatrix configuration, for reading device-wide settings.
|
|
|
|
``self.config`` is only this plugin's own slice, so cross-cutting
|
|
settings — ``target_fps``, ``timezone``, ``location`` — were previously
|
|
unreachable from a plugin without reaching into a manager by hand.
|
|
|
|
Resolution order mirrors the timezone helpers the sports plugins
|
|
already ship: ``plugin_manager.config_manager`` first (the cores that
|
|
hang it there), then ``cache_manager.config_manager``. Returns ``{}``
|
|
when neither is available, so callers can use plain ``.get()`` without
|
|
guarding, and a plugin on a core that predates this property still
|
|
loads — ``getattr(self, 'global_config', {})`` simply yields the
|
|
default.
|
|
|
|
Treat as read-only: the returned dict is the live config the core is
|
|
using, so mutating it edits every other consumer's view and can be
|
|
persisted back to disk.
|
|
|
|
Assignment is still allowed and wins over the resolved value. Several
|
|
shipped plugins (news, stock-news, ledmatrix-stocks, ledmatrix-
|
|
elections, ledmatrix-leaderboard, nfl-draft) set
|
|
``self.global_config`` to their own ``config['global']`` sub-dict; a
|
|
property without a setter would raise AttributeError and stop those
|
|
plugins loading.
|
|
|
|
Example:
|
|
fps = self.global_config.get('target_fps')
|
|
"""
|
|
override = getattr(self, '_global_config_override', None)
|
|
if override is not None:
|
|
return override
|
|
for owner in (self.plugin_manager, self.cache_manager):
|
|
config_manager = getattr(owner, 'config_manager', None)
|
|
if config_manager is None:
|
|
continue
|
|
try:
|
|
config = config_manager.get_config()
|
|
except Exception:
|
|
# A broken or unreadable config must never stop a plugin from
|
|
# loading; fall through to the next source, then to {}.
|
|
self.logger.debug(
|
|
"Could not read global config from %s",
|
|
type(owner).__name__, exc_info=True,
|
|
)
|
|
continue
|
|
# Only a real mapping is usable: callers do .get() on this and feed
|
|
# the result to numeric code, so handing back whatever a stub or a
|
|
# half-built manager returned would fail later and further away.
|
|
#
|
|
# An empty dict is treated as "nothing here yet" rather than a
|
|
# valid answer, so resolution continues to the next source. Both
|
|
# managers default to the same config/config.json, so falling
|
|
# through cannot pick up a different file's settings -- but it does
|
|
# rescue the case where the first manager simply hasn't loaded yet,
|
|
# which would otherwise return {} and silently disable every
|
|
# setting read through this property.
|
|
if isinstance(config, dict) and config:
|
|
return config
|
|
return {}
|
|
|
|
@global_config.setter
|
|
def global_config(self, value: Dict[str, Any]) -> None:
|
|
"""Let a plugin substitute its own view (see the getter's docstring)."""
|
|
self._global_config_override = value
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Adaptive layout support (opt-in)
|
|
# -------------------------------------------------------------------------
|
|
@property
|
|
def layout(self) -> Any:
|
|
"""
|
|
LayoutContext for the current logical display size.
|
|
|
|
Lazily built and rebuilt automatically when the display size changes
|
|
(e.g. Vegas segment widths, double-sided logical screens). Provides
|
|
Region carving (self.layout.bounds), breakpoint tiers, a geometry
|
|
scale factor vs. the manifest's display.design_size, and fit-text
|
|
queries against font ladders. See src/adaptive_layout.py.
|
|
|
|
Example:
|
|
rows = self.layout.bounds.inset(1).split_v(3, 1, gap=1)
|
|
self.draw_fit(big_text, rows[0], ladder=LADDER_ARCADE)
|
|
self.draw_fit(small_text, rows[1])
|
|
"""
|
|
from src.adaptive_layout import LayoutContext
|
|
|
|
width = getattr(self.display_manager, "width", None)
|
|
height = getattr(self.display_manager, "height", None)
|
|
if not width or not height:
|
|
matrix = getattr(self.display_manager, "matrix", None)
|
|
width = getattr(matrix, "width", 128)
|
|
height = getattr(matrix, "height", 32)
|
|
|
|
font_manager = self._get_font_manager()
|
|
generation = getattr(font_manager, "cache_generation", 0)
|
|
cached = getattr(self, "_layout_context", None)
|
|
if (cached is not None
|
|
and (cached.width, cached.height) == (width, height)
|
|
and getattr(self, "_layout_font_generation", None) == generation):
|
|
return cached
|
|
|
|
context = LayoutContext(
|
|
width, height, font_manager,
|
|
design_size=self._get_design_size(),
|
|
)
|
|
self._layout_context = context
|
|
self._layout_font_generation = generation
|
|
return context
|
|
|
|
@property
|
|
def styles(self) -> Any:
|
|
"""
|
|
The user's per-element styling: fonts, sizes, colours, offsets,
|
|
visibility, alignment and scale, resolved against this plugin's own
|
|
config_schema.json.
|
|
|
|
Every consumer of src.element_style used to repeat the same three
|
|
things -- a guarded import, finding its own schema file, and
|
|
rebuilding the resolver when on_config_change swapped the config
|
|
dict. This is those three things, once.
|
|
|
|
Ask for a style by element name, passing what the plugin drew before
|
|
the user could customise anything::
|
|
|
|
title = self.styles.style('title_text',
|
|
classic_font='PressStart2P-Regular.ttf',
|
|
classic_size=8,
|
|
classic_color=(255, 255, 255))
|
|
x, y = title.offset
|
|
self.display_manager.draw_text(text, x=x, y=y,
|
|
font=title.font, color=title.color)
|
|
|
|
The classic_* arguments matter: when the user has chosen nothing,
|
|
they come back verbatim, so a plugin that adopts this renders
|
|
identically until someone actually changes a setting.
|
|
|
|
A plugin whose display has modes (a scoreboard's live/upcoming/
|
|
recent, weather's current/hourly/daily) sets ``STYLE_MODE`` on the
|
|
class, and every lookup here honours the matching
|
|
``customization.modes.<mode>`` overrides without any call site
|
|
passing a mode. Use :meth:`styles_for` for a one-off mode.
|
|
|
|
Never raises: with no schema on disk, or with the element-style
|
|
module unavailable, lookups fall back to the classic values.
|
|
"""
|
|
resolver = getattr(self, "_style_resolver", None)
|
|
# The config dict is swapped wholesale by on_config_change, so
|
|
# identity is the invalidation signal -- the same check the sports
|
|
# base classes use.
|
|
if resolver is not None and resolver._config is self.config:
|
|
return resolver
|
|
resolver = self._build_style_resolver(getattr(self, "STYLE_MODE", None))
|
|
self._style_resolver = resolver
|
|
return resolver
|
|
|
|
def styles_for(self, mode: Optional[str]) -> Any:
|
|
""":attr:`styles`, bound to ``mode`` instead of ``STYLE_MODE``.
|
|
|
|
For a plugin that renders more than one mode from one instance. A
|
|
plugin with an instance per mode should set ``STYLE_MODE`` instead
|
|
and leave its call sites alone.
|
|
"""
|
|
cache = getattr(self, "_style_resolvers_by_mode", None)
|
|
if cache is None or getattr(self, "_style_resolver_config", None) is not self.config:
|
|
cache = {}
|
|
self._style_resolvers_by_mode = cache
|
|
self._style_resolver_config = self.config
|
|
if mode not in cache:
|
|
cache[mode] = self._build_style_resolver(mode)
|
|
return cache[mode]
|
|
|
|
def _build_style_resolver(self, mode: Optional[str]) -> Any:
|
|
"""Construct a resolver for this plugin's config and schema."""
|
|
try:
|
|
from src.element_style import (ElementStyleResolver,
|
|
defaults_from_schema_file)
|
|
except ImportError: # pragma: no cover - core always ships it
|
|
return _NullStyleResolver(self.config)
|
|
|
|
schema_path = self._config_schema_path()
|
|
defaults = (defaults_from_schema_file(schema_path) if schema_path
|
|
else {})
|
|
return ElementStyleResolver(self.config, defaults, mode=mode)
|
|
|
|
def _config_schema_path(self) -> Optional[str]:
|
|
"""This plugin's config_schema.json, or None.
|
|
|
|
Looked up from the concrete class's own module rather than from this
|
|
file: a plugin's subclass lives in its plugin directory, while this
|
|
module lives in src/plugin_system, where no plugin schema exists.
|
|
Falls back to the configured plugins directory, including the
|
|
ledmatrix- prefix form the loader accepts.
|
|
|
|
Returning None is safe, not fatal -- the resolver then has no
|
|
defaults to compare against, so every configured value counts as a
|
|
deliberate override, which is the conservative reading.
|
|
"""
|
|
cached = getattr(self, "_config_schema_path_cache", _UNSET_SCHEMA_PATH)
|
|
if cached is not _UNSET_SCHEMA_PATH:
|
|
return cached
|
|
|
|
path = None
|
|
try:
|
|
for candidate in self._schema_path_candidates():
|
|
if candidate and os.path.isfile(candidate):
|
|
path = candidate
|
|
break
|
|
except Exception as exc: # pragma: no cover - defensive
|
|
self.logger.debug("Could not locate config_schema.json: %s", exc)
|
|
self._config_schema_path_cache = path
|
|
return path
|
|
|
|
def _schema_path_candidates(self) -> list:
|
|
"""Where a plugin's schema might be, best guess first."""
|
|
candidates = []
|
|
|
|
module = sys.modules.get(type(self).__module__)
|
|
module_file = getattr(module, "__file__", None)
|
|
if module_file:
|
|
candidates.append(os.path.join(
|
|
os.path.dirname(os.path.abspath(module_file)),
|
|
"config_schema.json"))
|
|
|
|
plugins_dir = getattr(self.plugin_manager, "plugins_dir", None)
|
|
if plugins_dir:
|
|
for plugin_id in (self.plugin_id, f"ledmatrix-{self.plugin_id}"):
|
|
candidates.append(os.path.join(
|
|
str(plugins_dir), os.path.basename(plugin_id),
|
|
"config_schema.json"))
|
|
return candidates
|
|
|
|
def draw_fit(self, text: str, box: Any,
|
|
color: tuple = (255, 255, 255),
|
|
ladder: Optional[Any] = None,
|
|
align: str = "center", valign: str = "center") -> Any:
|
|
"""
|
|
Fit text to a Region with the largest crisp font that fits, then draw
|
|
it aligned within that region via the display manager.
|
|
|
|
Args:
|
|
text: Text to display (ellipsized if even the smallest rung is too wide)
|
|
box: Region (or (w, h) tuple anchored at 0,0) to fit and align within
|
|
color: RGB color tuple
|
|
ladder: FontLadder to walk (default LADDER_GRID; use LADDER_ARCADE
|
|
for headline text like clocks and scores)
|
|
align/valign: alignment of the text ink within the box
|
|
|
|
Returns:
|
|
FitResult (font, family, size_px, text, ink metrics, fits flag)
|
|
"""
|
|
from src.adaptive_layout import LADDER_DEFAULT, draw_fitted_text
|
|
|
|
fit = self.layout.fit_text(text, box, ladder=ladder or LADDER_DEFAULT)
|
|
draw_fitted_text(self.display_manager, fit, box,
|
|
color=color, align=align, valign=valign)
|
|
return fit
|
|
|
|
def draw_image(self, img: Any, box: Any, *,
|
|
mode: str = "contain", align: str = "center",
|
|
valign: str = "center", crop_to_ink: bool = False,
|
|
anchor: str = "center", resample: Optional[Any] = None,
|
|
cache_key: Optional[Any] = None,
|
|
offset: tuple = (0, 0)) -> Any:
|
|
"""
|
|
Fit an image into a Region and paste it aligned within that region
|
|
onto the display canvas — the image counterpart to draw_fit().
|
|
|
|
Args:
|
|
img: Source PIL image (logos, art, icons)
|
|
box: Region (or (w, h) tuple) to fit and align within
|
|
mode: "contain" (letterbox), "cover" (crop-to-fill),
|
|
"fill_height" (logo-style), "stretch"
|
|
crop_to_ink: Trim transparent padding before fitting
|
|
anchor: "center" or "top" for cover crops
|
|
resample: PIL filter; default LANCZOS. Use RESAMPLE_NEAREST
|
|
(from src.adaptive_images) for pixel art/flags
|
|
cache_key: Stable identity (e.g. "logo:KC") for cross-reload
|
|
caching; defaults to the image object's identity
|
|
offset: Final (dx, dy) translation — the hook for user
|
|
x/y-offset customization
|
|
|
|
Returns:
|
|
ImageFitResult (processed image + dimensions + scale)
|
|
"""
|
|
from src.adaptive_images import draw_fitted_image
|
|
|
|
ifit = self.layout.fit_image(img, box, mode=mode,
|
|
crop_to_ink=crop_to_ink, anchor=anchor,
|
|
resample=resample, cache_key=cache_key)
|
|
draw_fitted_image(self.display_manager, ifit, box,
|
|
align=align, valign=valign, offset=offset)
|
|
return ifit
|
|
|
|
def _get_font_manager(self) -> Any:
|
|
"""The shared FontManager, or a module-level fallback when running
|
|
under mocks/harnesses that don't provide one."""
|
|
font_manager = getattr(self.plugin_manager, "font_manager", None)
|
|
if font_manager is not None and hasattr(font_manager, "get_font"):
|
|
return font_manager
|
|
return _fallback_font_manager()
|
|
|
|
def _get_design_size(self) -> tuple:
|
|
"""Panel size this plugin's layout was authored against, from the
|
|
manifest's optional display.design_size (defaults to 128x32)."""
|
|
from src.adaptive_layout import DEFAULT_DESIGN_SIZE
|
|
|
|
if self.plugin_manager and hasattr(self.plugin_manager, "plugin_manifests"):
|
|
manifest = self.plugin_manager.plugin_manifests.get(self.plugin_id, {})
|
|
declared = manifest.get("display", {}).get("design_size", {})
|
|
width, height = declared.get("width"), declared.get("height")
|
|
if width and height:
|
|
return (int(width), int(height))
|
|
return DEFAULT_DESIGN_SIZE
|
|
|
|
def get_display_duration(self) -> float:
|
|
"""
|
|
Get the display duration for this plugin instance.
|
|
|
|
Uses, in order, the first positive number among:
|
|
1. ``self.display_duration`` (a common pattern in scoreboard plugins)
|
|
2. ``self.config["display_duration"]``
|
|
3. 15.0
|
|
|
|
Numeric strings count as numbers. Can be overridden by plugins to
|
|
provide dynamic durations based on content (e.g., longer duration for
|
|
more complex displays).
|
|
|
|
Returns:
|
|
Duration in seconds to display this plugin's content
|
|
"""
|
|
try:
|
|
duration = getattr(self, 'display_duration', None)
|
|
except (TypeError, ValueError, AttributeError) as e:
|
|
# A plugin may define display_duration as a property that raises.
|
|
self.logger.warning(
|
|
"Error reading display_duration instance variable: %s, using config fallback", e)
|
|
duration = None
|
|
if duration is not None:
|
|
seconds = self._positive_seconds(duration, "display_duration instance variable")
|
|
if seconds is not None:
|
|
return seconds
|
|
|
|
seconds = self._positive_seconds(
|
|
self.config.get("display_duration", 15.0), "config display_duration")
|
|
return seconds if seconds is not None else 15.0
|
|
|
|
def _positive_seconds(self, value: Any, source: str) -> Optional[float]:
|
|
"""``value`` as a positive float, or None (with a log line) if it is not one.
|
|
|
|
bool is rejected although it is an int subclass: True would otherwise
|
|
read as a 1-second duration.
|
|
"""
|
|
if isinstance(value, bool) or not isinstance(value, (int, float, str)):
|
|
self.logger.warning("%s has unexpected type %s (value: %s), ignoring it",
|
|
source, type(value).__name__, value)
|
|
return None
|
|
try:
|
|
seconds = float(value)
|
|
except ValueError:
|
|
self.logger.warning("%s has invalid value %r, ignoring it", source, value)
|
|
return None
|
|
if seconds > 0:
|
|
return seconds
|
|
self.logger.debug("%s is non-positive (%s), ignoring it", source, value)
|
|
return None
|
|
|
|
# ---------------------------------------------------------------------
|
|
# Dynamic duration support hooks
|
|
# ---------------------------------------------------------------------
|
|
def _get_dynamic_duration_config(self) -> Dict[str, Any]:
|
|
"""
|
|
Retrieve dynamic duration configuration block from plugin config.
|
|
|
|
Returns:
|
|
Dict with configuration values or empty dict if not configured.
|
|
"""
|
|
value = self.config.get("dynamic_duration", {})
|
|
if isinstance(value, dict):
|
|
return value
|
|
return {}
|
|
|
|
def supports_dynamic_duration(self) -> bool:
|
|
"""
|
|
Determine whether this plugin should use dynamic display durations.
|
|
|
|
Plugins can override to implement custom logic. By default this reads the
|
|
`dynamic_duration.enabled` flag from plugin configuration.
|
|
"""
|
|
config = self._get_dynamic_duration_config()
|
|
return bool(config.get("enabled", False))
|
|
|
|
def get_dynamic_duration_cap(self) -> Optional[float]:
|
|
"""
|
|
Return the maximum duration (in seconds) the controller should wait for
|
|
this plugin to complete its display cycle when using dynamic duration.
|
|
|
|
Returns:
|
|
Positive float value for explicit cap, or None to indicate no
|
|
additional cap beyond global defaults.
|
|
"""
|
|
config = self._get_dynamic_duration_config()
|
|
cap_value = config.get("max_duration_seconds")
|
|
if cap_value is None:
|
|
return None
|
|
try:
|
|
cap = float(cap_value)
|
|
if cap <= 0:
|
|
return None
|
|
return cap
|
|
except (TypeError, ValueError):
|
|
self.logger.warning(
|
|
"Invalid dynamic_duration.max_duration_seconds for %s: %s",
|
|
self.plugin_id,
|
|
cap_value,
|
|
)
|
|
return None
|
|
|
|
def is_cycle_complete(self) -> bool:
|
|
"""
|
|
Indicate whether the plugin has completed a full display cycle.
|
|
|
|
The display controller calls this after each display iteration when
|
|
dynamic duration is enabled. Plugins that render multi-step content
|
|
should override this method and return True only after all content has
|
|
been shown once.
|
|
|
|
Returns:
|
|
True if the plugin cycle is complete (default behaviour).
|
|
"""
|
|
return True
|
|
|
|
def reset_cycle_state(self) -> None:
|
|
"""
|
|
Reset any internal counters/state related to cycle tracking.
|
|
|
|
Called by the display controller before beginning a new dynamic-duration
|
|
session. Override in plugins that maintain custom tracking data.
|
|
"""
|
|
return
|
|
|
|
def get_update_interval(self) -> Optional[float]:
|
|
"""
|
|
How often this plugin wants update() called, right now, in seconds.
|
|
|
|
The manifest's ``update_interval`` is a single static number, which
|
|
cannot say "poll me every 15 seconds while a game is in progress and
|
|
every 15 minutes when nothing is on". Only the plugin knows which is
|
|
true at any moment, so override this to say so.
|
|
|
|
Return None (the default) to accept the manifest/config value.
|
|
|
|
Two constraints, both because the scheduler calls this on every tick of
|
|
the render loop:
|
|
|
|
- It must be cheap. Attribute reads only -- no config lookups, no I/O,
|
|
no locks that a fetch might be holding.
|
|
- It must not raise. A raising hook is ignored and the static interval
|
|
used, but a hook that raises every tick also logs every tick.
|
|
|
|
Values below PluginManager.MIN_DYNAMIC_UPDATE_INTERVAL are clamped up:
|
|
a plugin asking for 0 would otherwise busy-wait against its own API.
|
|
|
|
Example::
|
|
|
|
def get_update_interval(self):
|
|
# Fast while something is actually live, manifest default otherwise.
|
|
if any(m.live_games for m in self._live_managers):
|
|
return self.config.get("live_update_interval", 15)
|
|
return None
|
|
"""
|
|
return None
|
|
|
|
def has_live_priority(self) -> bool:
|
|
"""
|
|
Check if this plugin has live priority enabled.
|
|
|
|
Live priority allows a plugin to take over the display when it has
|
|
live/urgent content (e.g., live sports games, breaking news).
|
|
|
|
Returns:
|
|
True if live priority is enabled in config, False otherwise
|
|
"""
|
|
return self.config.get("live_priority", False)
|
|
|
|
def has_live_content(self) -> bool:
|
|
"""
|
|
Check if this plugin currently has live content to display.
|
|
|
|
Override this method in your plugin to implement live content detection.
|
|
This is called by the display controller to determine if a live priority
|
|
plugin should take over the display.
|
|
|
|
Returns:
|
|
True if plugin has live content, False otherwise
|
|
|
|
Example (sports plugin):
|
|
def has_live_content(self):
|
|
# Check if there are any live games
|
|
return hasattr(self, 'live_games') and len(self.live_games) > 0
|
|
|
|
Example (news plugin):
|
|
def has_live_content(self):
|
|
# Check if there's breaking news
|
|
return hasattr(self, 'breaking_news') and self.breaking_news
|
|
"""
|
|
return False
|
|
|
|
def get_vegas_priority_weight(self) -> Optional[int]:
|
|
"""How many slots per Vegas cycle this plugin should get, or None.
|
|
|
|
The Vegas ticker is otherwise a strict round robin: every plugin
|
|
appears exactly once per cycle. With a dozen plugins enabled that puts
|
|
minutes between a live score and its next appearance. A weight of N
|
|
gives the plugin N slots per cycle, spread evenly through it rather
|
|
than clumped together.
|
|
|
|
Return ``None`` (the default) to let the core decide. It gives a
|
|
plugin ``vegas_scroll.live_weight`` when ``has_live_priority()`` and
|
|
``has_live_content()`` are both true, and 1 otherwise -- so live sports
|
|
already get extra turns without implementing this at all.
|
|
|
|
Implement it only when the plugin knows something the core cannot. The
|
|
motivating case is favorite teams: the core can see *that* a game is
|
|
live but not *whose*, so a scoreboard that wants its favorite's game
|
|
shown more often than other live games has to say so::
|
|
|
|
def get_vegas_priority_weight(self):
|
|
if not (self.has_live_priority() and self.has_live_content()):
|
|
return None # let the core decide
|
|
cfg = self.global_config.get('display', {}).get('vegas_scroll', {})
|
|
if self._favorite_is_live():
|
|
return cfg.get('favorite_live_weight', 5)
|
|
return cfg.get('live_weight', 3)
|
|
|
|
The weight is per *plugin*, not per game. A scoreboard showing four
|
|
live games still occupies one slot at a time and rotates its own games
|
|
within that slot; this controls how often the plugin itself comes
|
|
round.
|
|
|
|
Raising is safe: the core logs it and falls back to its own
|
|
live-content check, so a broken weight calculation costs the plugin
|
|
the favorite distinction but not the live boost.
|
|
|
|
Returns:
|
|
Slots per cycle (clamped to 1..10 by the caller), or None to
|
|
defer to the core's own live-content weighting.
|
|
"""
|
|
return None
|
|
|
|
def get_live_modes(self) -> List[str]:
|
|
"""
|
|
Get list of display modes that should be used during live priority takeover.
|
|
|
|
Override this method to specify which modes should be shown when this
|
|
plugin has live content. By default, returns all display modes from manifest.
|
|
|
|
Returns:
|
|
List of mode names to display during live priority
|
|
|
|
Example:
|
|
def get_live_modes(self):
|
|
# Only show live game mode, not upcoming/recent
|
|
return ['nhl_live', 'nba_live']
|
|
"""
|
|
# Get display modes from manifest via plugin manager
|
|
if self.plugin_manager and hasattr(self.plugin_manager, "plugin_manifests"):
|
|
manifest = self.plugin_manager.plugin_manifests.get(self.plugin_id, {})
|
|
return manifest.get("display_modes", [self.plugin_id])
|
|
return [self.plugin_id]
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Vegas scroll mode support
|
|
# -------------------------------------------------------------------------
|
|
def get_vegas_render_width(self) -> int:
|
|
"""
|
|
Width the Vegas ticker wants this plugin's content to occupy.
|
|
|
|
On a wide panel a layout built to fill the screen reads as sparse in a
|
|
ticker — a forecast spread over five columns, a progress bar drawn at
|
|
100% width, a stat block with the panel's whole width between its
|
|
elements. Vegas asks for a narrower render so the plugin can choose a
|
|
tighter arrangement instead of being cropped afterwards.
|
|
|
|
Vegas also narrows ``display_manager`` for the duration of the call, so
|
|
a plugin that already sizes itself from ``display_manager.width`` needs
|
|
no changes. Read this only when you size content some other way.
|
|
|
|
Controlled by the plugin's own ``vegas_width_pct`` config value, else
|
|
the global ``display.vegas_scroll.render_width_pct``.
|
|
|
|
Returns:
|
|
Target width in pixels. Outside a Vegas content request, the full
|
|
display width: ``display_manager.width``, which falls back to the
|
|
canvas size when ``matrix`` is None (hardware init failed).
|
|
"""
|
|
requested = getattr(self, '_vegas_render_width', None)
|
|
if isinstance(requested, int) and requested > 0:
|
|
return requested
|
|
|
|
# display_manager.width first, as CLAUDE.md asks of every plugin: it
|
|
# already reads matrix.width when there is a matrix. matrix.width is
|
|
# only the fallback for a display_manager without a width (a test
|
|
# double, an older wrapper).
|
|
display_manager = getattr(self, 'display_manager', None)
|
|
width = getattr(display_manager, 'width', None)
|
|
if callable(width):
|
|
width = width()
|
|
if width:
|
|
return int(width)
|
|
matrix = getattr(display_manager, 'matrix', None)
|
|
if matrix is not None and getattr(matrix, 'width', None):
|
|
return int(matrix.width)
|
|
return 128
|
|
|
|
def get_vegas_content(self) -> Optional[Any]:
|
|
"""
|
|
Get content for Vegas-style continuous scroll mode.
|
|
|
|
Override this method to provide optimized content for continuous scrolling.
|
|
Plugins can return:
|
|
- A single PIL Image: Displayed as a static block in the scroll
|
|
- A list of PIL Images: Each image becomes a separate item in the scroll
|
|
- None: Vegas mode will fall back to capturing display() output
|
|
|
|
Multi-item plugins (sports scores, odds) should return individual game/item
|
|
images so they scroll smoothly with other plugins.
|
|
|
|
Returns:
|
|
PIL Image, list of PIL Images, or None
|
|
|
|
Example (sports plugin):
|
|
def get_vegas_content(self):
|
|
# Return individual game cards for smooth scrolling
|
|
return [self._render_game(game) for game in self.games]
|
|
|
|
Example (static plugin):
|
|
def get_vegas_content(self):
|
|
# Return current display as single block
|
|
return self._render_current_view()
|
|
"""
|
|
return None
|
|
|
|
def get_vegas_content_type(self) -> str:
|
|
"""
|
|
Indicate the type of content this plugin provides for Vegas scroll.
|
|
|
|
Override this to specify how Vegas mode should treat this plugin's content.
|
|
|
|
Returns:
|
|
'multi' - Plugin has multiple scrollable items (sports, odds, news)
|
|
'static' - Plugin is a static block (clock, weather, music)
|
|
'none' - Plugin should not appear in Vegas scroll mode
|
|
|
|
Example:
|
|
def get_vegas_content_type(self):
|
|
return 'multi' # We have multiple games to scroll
|
|
"""
|
|
return 'static'
|
|
|
|
def get_vegas_display_mode(self) -> VegasDisplayMode:
|
|
"""
|
|
Get the display mode for Vegas scroll integration.
|
|
|
|
This method determines how the plugin's content behaves within Vegas mode:
|
|
- SCROLL: Content scrolls continuously (multi-item plugins)
|
|
- FIXED_SEGMENT: Fixed block that scrolls by (clock, weather)
|
|
- STATIC: Pause scroll to display (alerts, detailed views)
|
|
|
|
Override to change default behavior. By default, reads from config
|
|
or maps legacy get_vegas_content_type() for backward compatibility.
|
|
|
|
Returns:
|
|
VegasDisplayMode enum value
|
|
|
|
Example:
|
|
def get_vegas_display_mode(self):
|
|
return VegasDisplayMode.SCROLL
|
|
"""
|
|
# Check for explicit config setting first
|
|
config_mode = self.config.get("vegas_mode")
|
|
if config_mode:
|
|
try:
|
|
return VegasDisplayMode(config_mode)
|
|
except ValueError:
|
|
self.logger.warning(
|
|
"Invalid vegas_mode '%s' for %s, using default",
|
|
config_mode, self.plugin_id
|
|
)
|
|
|
|
# Fall back to mapping legacy content_type. 'none' (excluded from
|
|
# Vegas) also maps to FIXED_SEGMENT: exclusion is decided by checking
|
|
# get_vegas_content_type() separately.
|
|
if self.get_vegas_content_type() == 'multi':
|
|
return VegasDisplayMode.SCROLL
|
|
return VegasDisplayMode.FIXED_SEGMENT
|
|
|
|
def get_supported_vegas_modes(self) -> List[VegasDisplayMode]:
|
|
"""
|
|
Return list of Vegas display modes this plugin supports.
|
|
|
|
Not currently consulted by core: neither Vegas mode nor the web UI
|
|
calls it. It is kept, and plugins override it, as the declared set of
|
|
modes a future mode picker would offer.
|
|
|
|
By default:
|
|
- 'multi' content type plugins support SCROLL and FIXED_SEGMENT
|
|
- 'static' content type plugins support FIXED_SEGMENT and STATIC
|
|
- 'none' content type plugins return empty list (excluded from Vegas)
|
|
|
|
Returns:
|
|
List of VegasDisplayMode values this plugin can use
|
|
|
|
Example:
|
|
def get_supported_vegas_modes(self):
|
|
# This plugin only makes sense as a scrolling ticker
|
|
return [VegasDisplayMode.SCROLL]
|
|
"""
|
|
content_type = self.get_vegas_content_type()
|
|
|
|
if content_type == 'none':
|
|
return []
|
|
elif content_type == 'multi':
|
|
return [VegasDisplayMode.SCROLL, VegasDisplayMode.FIXED_SEGMENT]
|
|
else: # 'static'
|
|
return [VegasDisplayMode.FIXED_SEGMENT, VegasDisplayMode.STATIC]
|
|
|
|
def get_vegas_segment_width(self) -> Optional[int]:
|
|
"""
|
|
Get the preferred width for this plugin in Vegas FIXED_SEGMENT mode.
|
|
|
|
Not currently consulted by core: Vegas mode sizes a card from the
|
|
``vegas_width_pct`` / ``vegas_scroll.render_width_pct`` settings
|
|
(see get_vegas_render_width()). Kept because plugins override it.
|
|
|
|
Returns the number of panels this plugin should occupy when displayed
|
|
as a fixed segment. The actual pixel width is calculated as:
|
|
width = panels * single_panel_width
|
|
|
|
Where single_panel_width comes from display.hardware.cols in config.
|
|
|
|
Override to provide dynamic sizing based on content.
|
|
Returns None to use the default (1 panel).
|
|
|
|
Returns:
|
|
Number of panels, or None for default (1 panel)
|
|
|
|
Example:
|
|
def get_vegas_segment_width(self):
|
|
# Clock needs 2 panels to show time clearly
|
|
return 2
|
|
"""
|
|
raw_value = self.config.get("vegas_panel_count", None)
|
|
if raw_value is None:
|
|
return None
|
|
|
|
try:
|
|
panel_count = int(raw_value)
|
|
if panel_count > 0:
|
|
return panel_count
|
|
else:
|
|
self.logger.warning(
|
|
"vegas_panel_count must be positive, got %s; using default",
|
|
raw_value
|
|
)
|
|
return None
|
|
except (ValueError, TypeError):
|
|
self.logger.warning(
|
|
"Invalid vegas_panel_count value '%s'; using default",
|
|
raw_value
|
|
)
|
|
return None
|
|
|
|
def validate_config(self) -> bool:
|
|
"""
|
|
Validate plugin configuration against schema.
|
|
|
|
Called during plugin loading to ensure configuration is valid.
|
|
Override this method to implement custom validation logic.
|
|
|
|
Returns:
|
|
True if config is valid, False otherwise
|
|
|
|
Example:
|
|
def validate_config(self):
|
|
required_fields = ['api_key', 'city']
|
|
for field in required_fields:
|
|
if field not in self.config:
|
|
self.logger.error("Missing required field: %s", field)
|
|
return False
|
|
return True
|
|
"""
|
|
# Basic validation - check that enabled is a boolean if present
|
|
if "enabled" in self.config:
|
|
if not isinstance(self.config["enabled"], bool):
|
|
self.logger.error("'enabled' must be a boolean")
|
|
return False
|
|
|
|
# Check display_duration if present. bool is excluded explicitly:
|
|
# it's an int subclass, and get_display_duration rejects it too.
|
|
if "display_duration" in self.config:
|
|
duration = self.config["display_duration"]
|
|
if (not isinstance(duration, (int, float))
|
|
or isinstance(duration, bool) or duration <= 0):
|
|
self.logger.error("'display_duration' must be a positive number")
|
|
return False
|
|
|
|
return True
|
|
|
|
def cleanup(self) -> None:
|
|
"""
|
|
Cleanup resources when plugin is unloaded.
|
|
|
|
Override this method to clean up any resources (e.g., close
|
|
file handles, terminate threads, close network connections).
|
|
|
|
This method is called when the plugin is unloaded or when the
|
|
system is shutting down.
|
|
|
|
Example:
|
|
def cleanup(self):
|
|
if hasattr(self, 'api_client'):
|
|
self.api_client.close()
|
|
if hasattr(self, 'worker_thread'):
|
|
self.worker_thread.stop()
|
|
"""
|
|
self.logger.info("Cleaning up plugin: %s", self.plugin_id)
|
|
|
|
def on_config_change(self, new_config: Dict[str, Any]) -> None:
|
|
"""
|
|
Called after the plugin configuration has been updated via the web API.
|
|
|
|
Plugins may override this to apply changes immediately without a restart.
|
|
The default implementation updates the in-memory config.
|
|
|
|
Args:
|
|
new_config: The full, merged configuration for this plugin (including
|
|
any secret-derived values that are merged at runtime).
|
|
"""
|
|
# Update config reference
|
|
self.config = new_config or {}
|
|
|
|
# Update simple flags
|
|
self.enabled = self.config.get("enabled", self.enabled)
|
|
|
|
def get_info(self) -> Dict[str, Any]:
|
|
"""
|
|
Return plugin info for display in web UI.
|
|
|
|
Override this method to provide additional information about
|
|
the plugin's current state.
|
|
|
|
Returns:
|
|
Dict with plugin information including id, enabled status, and config
|
|
|
|
Example:
|
|
def get_info(self):
|
|
info = super().get_info()
|
|
info['games_count'] = len(self.games)
|
|
info['last_update'] = self.last_update_time
|
|
return info
|
|
"""
|
|
return {
|
|
"id": self.plugin_id,
|
|
"enabled": self.enabled,
|
|
"config": self.config,
|
|
"api_version": self.API_VERSION,
|
|
}
|
|
|
|
def on_enable(self) -> None:
|
|
"""
|
|
Called when plugin is enabled.
|
|
|
|
Override this method to perform any actions needed when the
|
|
plugin is enabled (e.g., start background tasks, open connections).
|
|
"""
|
|
self.enabled = True
|
|
self.logger.info("Plugin enabled: %s", self.plugin_id)
|
|
|
|
def on_disable(self) -> None:
|
|
"""
|
|
Called when plugin is disabled.
|
|
|
|
Override this method to perform any actions needed when the
|
|
plugin is disabled (e.g., stop background tasks, close connections).
|
|
"""
|
|
self.enabled = False
|
|
self.logger.info("Plugin disabled: %s", self.plugin_id)
|