Files
LEDMatrix/src/plugin_system/base_plugin.py
T
08265c1135 feat(vegas): let live content keep its place in the ticker (#457)
* feat(vegas): let live content keep its place in the ticker

Live content used to preempt Vegas outright: while any plugin reported
live priority the display controller refused to run the ticker at all and
showed a full-screen scoreboard instead. Keeping the marquee meant not
seeing live scores; seeing live scores meant losing the marquee.

Two changes, both off by default.

vegas_scroll.live_in_ticker keeps the ticker running through a live game.
Three places assumed the takeover and all three now honour it: the
controller's gate, the coordinator's per-frame pause, and the rotation
switch that would otherwise move current_mode_index underneath a ticker
that never yields.

And the rotation is no longer a strict round robin. It was one slot per
plugin per cycle, so with a dozen plugins enabled a live score came round
once a lap and could be minutes old on screen. A plugin can now hold
several slots, placed by Smooth Weighted Round-Robin -- the same
scheduler the sports plugins already use to rotate their own games. The
property that matters is that repeats are spread through the cycle
rather than clumped: three in a row and then silence would be worse than
no boost at all.

Weight comes from the plugin first, via a new optional
get_vegas_priority_weight(), then from the core: live content earns
live_weight, everything else 1. So existing plugins gain the behaviour
without changes, and the hook exists for the one thing the core cannot
work out -- the core can see that a game is live but not whose, so only
the plugin can say a favorite is playing.

Documented in ADVANCED_FEATURES (worked example, why weights are per
plugin not per game, and that frequency is not freshness),
CONFIG_REFERENCE, PLUGIN_API_REFERENCE, and the config template.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Udr6MfaFLUPhX5Fgo67Jf5

* fix(vegas): carry the new keys through config, and correct two docs

Three findings from CodeRabbit, all valid.

to_dict() and update() enumerate keys explicitly and had not learned the
three new ones, so get_status() never reported them and a live config
change never applied -- turning live_in_ticker on in the web UI would
have done nothing until a restart. update() clamps the weights exactly
as from_config does.

The vegas_scroll key count in ADVANCED_FEATURES said 29; the template
has 30. My arithmetic, not the reviewer's.

The third was a documentation error rather than a code one, and I have
fixed it the other way round. The docs claimed a raising
get_vegas_priority_weight() is treated as weight 1. The code instead
falls through to the core's own live-content check, and that is the
better behaviour: the hook is only how a plugin asks for *more* than
live_weight, and has_live_priority/has_live_content are separate methods
guarded separately, so a plugin with a broken weight calculation should
lose the favorite distinction and keep the live boost. Said so in the
code, the base-plugin docstring and the API reference.

The test fake now fails in each place independently, because the two
failures mean different things: a broken hook still earns live_weight, a
plugin that cannot say whether it is live has nothing to fall back on
and weighs 1.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ui/code/session_01Udr6MfaFLUPhX5Fgo67Jf5

* fix(vegas): stop the heaviest plugin doubling across the cycle seam

Smooth Weighted Round-Robin spaces repeats well within a pass, but it
schedules the heaviest item first and usually last as well. The strip
loops, so those two are neighbours: the marquee showed the same plugin
twice running at exactly the one join a within-cycle check cannot see.
Observed on a live rig at 28 slots -- gaps of 6, 7, 7, 7 and then 1.

Rotating the list does not fix it. Rotation preserves the cyclic order
exactly, so it moves where the seam is drawn rather than the adjacency
itself; the trailing entry has to be swapped with one from the middle.

The first version swapped with the first slot that merely fitted, which
undid the spacing this exists to protect -- it moved a repeat from a gap
of 7 into a gap of 2, more clumped than the seam had ever been. It now
picks the candidate furthest from any other appearance, so the repeat
lands in the widest gap.

Left alone when no candidate exists. A plugin holding most of the slots
has to neighbour itself, and scheduling it is better than refusing to.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Udr6MfaFLUPhX5Fgo67Jf5

* fix(vegas): stop the seam repair creating the duplicate it removes

Swapping the trailing repeat with a middle slot moves two elements, and
the candidate filter only guarded one of them. It checked the neighbours
`repeated` would acquire at j, but not what the displaced element would
sit beside at the end -- so ['a','b','c','d','x','y','x','a'] came back
as [...,'x','x'], the seam duplicate traded for a fresh one. Reported by
CodeRabbit with that exact case.

Adding the missing condition fixed it and immediately broke something
else: schedule[j] is schedule[-2] when j is the second-to-last slot, so
that candidate was always excluded, and ['a','b','c','a'] lost the only
repair it has. The same class of mistake twice, from reasoning about
which neighbours two moved elements end up with.

So it no longer reasons. It performs each candidate swap, counts the
cyclic duplicates in the result, and keeps the best one that has none --
preferring whichever leaves the boosted plugin most evenly spread. When
no such swap exists the schedule is returned untouched, which is the
unavoidable case: a plugin holding most of the slots has to neighbour
itself.

Fuzzed across 6,956 seam schedules: none made worse, none lost an entry.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Udr6MfaFLUPhX5Fgo67Jf5

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-13 17:02:36 -04:00

932 lines
37 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 logging
from src.logging_config import get_logger
_shared_fallback_font_manager: Optional[Any] = None
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"
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
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.
Automatically detects duration from:
1. self.display_duration instance variable (if exists)
2. self.config.get("display_duration", 15.0) (fallback)
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
"""
# Check for instance variable first (common pattern in scoreboard plugins)
if hasattr(self, 'display_duration'):
try:
duration = getattr(self, 'display_duration')
# Handle None case
if duration is None:
pass # Fall through to config
# Try to convert to float if it's a number or numeric string.
# bool is excluded: it's an int subclass, and True would
# otherwise read as a 1-second duration.
elif isinstance(duration, (int, float)) and not isinstance(duration, bool):
if duration > 0:
return float(duration)
else:
self.logger.debug(
"display_duration instance variable is non-positive (%s), using config fallback",
duration
)
# Try converting string representations of numbers
elif isinstance(duration, str):
try:
duration_float = float(duration)
if duration_float > 0:
return duration_float
else:
self.logger.debug(
"display_duration string value is non-positive (%s), using config fallback",
duration
)
except (ValueError, TypeError):
self.logger.warning(
"display_duration instance variable has invalid string value '%s', using config fallback",
duration
)
else:
self.logger.warning(
"display_duration instance variable has unexpected type %s (value: %s), using config fallback",
type(duration).__name__, duration
)
except (TypeError, ValueError, AttributeError) as e:
self.logger.warning(
"Error reading display_duration instance variable: %s, using config fallback",
e
)
# Fall back to config
config_duration = self.config.get("display_duration", 15.0)
try:
# Ensure config value is also a valid float (bool excluded — an
# int subclass that would otherwise read True as 1 second)
if isinstance(config_duration, (int, float)) and not isinstance(config_duration, bool):
if config_duration > 0:
return float(config_duration)
else:
self.logger.debug(
"Config display_duration is non-positive (%s), using default 15.0",
config_duration
)
return 15.0
elif isinstance(config_duration, str):
try:
duration_float = float(config_duration)
if duration_float > 0:
return duration_float
else:
self.logger.debug(
"Config display_duration string is non-positive (%s), using default 15.0",
config_duration
)
return 15.0
except ValueError:
self.logger.warning(
"Config display_duration has invalid string value '%s', using default 15.0",
config_duration
)
return 15.0
else:
self.logger.warning(
"Config display_duration has unexpected type %s (value: %s), using default 15.0",
type(config_duration).__name__, config_duration
)
except (ValueError, TypeError) as e:
self.logger.warning(
"Error processing config display_duration: %s, using default 15.0",
e
)
return 15.0
# ---------------------------------------------------------------------
# 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 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 ``matrix.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.
"""
requested = getattr(self, '_vegas_render_width', None)
if isinstance(requested, int) and requested > 0:
return requested
display_manager = getattr(self, 'display_manager', None)
matrix = getattr(display_manager, 'matrix', None)
if matrix is not None and getattr(matrix, 'width', None):
return int(matrix.width)
width = getattr(display_manager, 'width', None)
if callable(width):
width = width()
return int(width) if width else 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
content_type = self.get_vegas_content_type()
if content_type == 'multi':
return VegasDisplayMode.SCROLL
elif content_type == 'static':
return VegasDisplayMode.FIXED_SEGMENT
elif content_type == 'none':
# 'none' means excluded - return FIXED_SEGMENT as default
# The exclusion is handled by checking get_vegas_content_type() separately
return VegasDisplayMode.FIXED_SEGMENT
return VegasDisplayMode.FIXED_SEGMENT
def get_supported_vegas_modes(self) -> List[VegasDisplayMode]:
"""
Return list of Vegas display modes this plugin supports.
Used by the web UI to show available mode options for user configuration.
Override to customize which modes are available for this plugin.
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.
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)