mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-05 14:55:08 +00:00
ChuckandClaude Opus 5.5
701c220ad0
Vegas bakes each plugin's pictures into one strip, so a card already on its way across the panel keeps what it showed when it was drawn. This adds the API and bookkeeping for content that can be updated in place; the worker that redraws and swaps it follows separately. No shipped plugin implements the hook yet, so nothing changes for users. Plugin API (core 3.8.0), all no-ops by default: - BasePlugin.get_vegas_elements() -> [VegasElement(key, image, version, live, refresh_hz)]: named, fixed-width pieces of Vegas content. - BasePlugin.redraw_vegas_element(key, width, height, at): a lock-free redraw for content that changes with time. - BasePlugin.notify_vegas_data_changed(): data that lands outside update(). - src/plugin_system/vegas_elements.py (VegasElement, re-exported from base_plugin). Core: - PluginAdapter asks a plugin that implements the hook for elements on the background fetch only (under its lock, on its own canvas); every other path keeps get_vegas_content(). Live elements are pinned (padded with content_padding, never trimmed), tagged with their key, digest and data epoch in Image.info so the existing cache and group plumbing carry them unchanged, and untagged if a width budget crops them. - RenderPipeline records where each live element lands (ElementRecord), in absolute strip columns a trim does not move; the block-start arithmetic is shared with the STATIC markers. - PluginManager update listeners (add/remove_update_listener, notify_data_changed): told the moment update() completes, not at the next ~4s Vegas poll. The coordinator uses one to move each plugin's data epoch on. - vegas_scroll.live_refresh (kill switch), live_max_hz, live_min_interval, live_lead_screens; per-plugin core-owned vegas_live. Live elements are off under multi-display sync, in swap mode and with offscreen_prefetch off. - scripts/check_plugin.py checks the element contract (src/plugin_system/testing/vegas.py); test/fixtures/plugins/vegas-live-stub is a working example. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
138 lines
4.2 KiB
Python
138 lines
4.2 KiB
Python
"""When Vegas live elements are on, and that everywhere else nothing changes.
|
|
|
|
Live elements run only when the config allows them and nothing rules them
|
|
out: multi-display sync (the follower mirrors whole strips only), swap mode,
|
|
the offscreen kill switch, or a display manager with no off-screen canvas.
|
|
While they are on, the coordinator listens for plugin updates and moves each
|
|
plugin's data epoch on; while off, the adapter never asks for elements.
|
|
"""
|
|
import sys
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
from PIL import Image
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from src.vegas_mode.coordinator import VegasModeCoordinator # noqa: E402
|
|
|
|
|
|
class _DM:
|
|
width, height = 128, 32
|
|
|
|
def __init__(self, offscreen=True):
|
|
self.image = Image.new("RGB", (128, 32))
|
|
if offscreen:
|
|
self.offscreen = self._offscreen
|
|
|
|
@contextmanager
|
|
def _offscreen(self, width=None, height=None):
|
|
yield SimpleNamespace(image=Image.new("RGB", (width or 128, 32)))
|
|
|
|
def set_scrolling_state(self, *a, **k):
|
|
pass
|
|
|
|
def update_display(self):
|
|
pass
|
|
|
|
|
|
class _PM:
|
|
def __init__(self):
|
|
self.plugins = {}
|
|
self.listeners = []
|
|
|
|
def add_update_listener(self, fn):
|
|
if fn not in self.listeners:
|
|
self.listeners.append(fn)
|
|
|
|
def remove_update_listener(self, fn):
|
|
self.listeners = [f for f in self.listeners if f != fn]
|
|
|
|
|
|
def _coordinator(dm=None, **vegas):
|
|
vegas.setdefault("enabled", True)
|
|
config = {"display": {"vegas_scroll": vegas}}
|
|
return VegasModeCoordinator(config, dm or _DM(), _PM())
|
|
|
|
|
|
def test_on_by_default():
|
|
c = _coordinator()
|
|
c._apply_live_state()
|
|
assert c.live_active
|
|
assert c.plugin_adapter.live_elements_enabled
|
|
assert c._on_plugin_data_changed in c.plugin_manager.listeners
|
|
|
|
|
|
@pytest.mark.parametrize("why,vegas,dm", [
|
|
("switched off", {"live_refresh": False}, None),
|
|
("swap mode", {"continuous_scroll": False}, None),
|
|
("offscreen kill switch", {"offscreen_prefetch": False}, None),
|
|
("no offscreen canvas", {}, _DM(offscreen=False)),
|
|
])
|
|
def test_off_when_ruled_out(why, vegas, dm):
|
|
c = _coordinator(dm, **vegas)
|
|
c._apply_live_state()
|
|
assert not c.live_active
|
|
assert not c.plugin_adapter.live_elements_enabled
|
|
assert c.plugin_manager.listeners == []
|
|
|
|
|
|
@pytest.mark.parametrize("role", ["leader", "follower"])
|
|
def test_off_whenever_sync_is_configured(role):
|
|
c = _coordinator()
|
|
c.set_sync_manager(SimpleNamespace(role=role))
|
|
c._apply_live_state()
|
|
assert not c.live_active
|
|
assert not c.plugin_adapter.live_elements_enabled
|
|
|
|
|
|
def test_a_standalone_sync_manager_does_not_count():
|
|
from src.common.sync_manager import SyncRole
|
|
c = _coordinator()
|
|
c.set_sync_manager(SimpleNamespace(role=SyncRole.STANDALONE))
|
|
c._apply_live_state()
|
|
assert c.live_active
|
|
|
|
|
|
def test_stopping_switches_it_off_and_stops_listening():
|
|
c = _coordinator()
|
|
c._apply_live_state()
|
|
c._is_active = True
|
|
c.stop()
|
|
assert not c.live_active and not c.plugin_adapter.live_elements_enabled
|
|
assert c.plugin_manager.listeners == []
|
|
|
|
|
|
def test_a_config_change_can_switch_it_off_mid_run():
|
|
c = _coordinator()
|
|
c._apply_live_state()
|
|
c._is_active = True
|
|
c.stream_manager.refresh = MagicMock()
|
|
c.update_config({"display": {"vegas_scroll": {"enabled": True,
|
|
"live_refresh": False}}})
|
|
c._apply_pending_config()
|
|
assert not c.live_active
|
|
assert c.plugin_manager.listeners == []
|
|
|
|
|
|
def test_an_update_moves_the_plugins_epoch_on():
|
|
c = _coordinator()
|
|
c._apply_live_state()
|
|
before = c.live_epochs.get("p")
|
|
for listener in c.plugin_manager.listeners:
|
|
listener("p")
|
|
assert c.live_epochs.get("p") > before
|
|
assert c.plugin_adapter.live_epochs is c.live_epochs
|
|
|
|
|
|
def test_the_state_change_is_logged_once(caplog):
|
|
c = _coordinator(live_refresh=False)
|
|
with caplog.at_level("INFO"):
|
|
c._apply_live_state()
|
|
c._apply_live_state()
|
|
lines = [r.message for r in caplog.records if "live elements" in r.message]
|
|
assert lines == ["Vegas live elements off: switched off (vegas_scroll.live_refresh)"]
|