Compare commits

..
Author SHA1 Message Date
68d03ea260 feat: universal element style system — resolver, schema expansion, live preview (#394)
* feat(element-style): universal per-element style resolver

One shared implementation of the three things every customizable plugin
re-invented: a font loader, the customization.layout x/y-offset reader,
and the "did the user actually override this font?" check.

The override check is the load-bearing piece: the web UI's save flow
writes full schema defaults into config.json on every save, and the
plugin manager merges defaults again before instantiation, so key
presence never means user intent. The resolver compares against the
plugin's own schema defaults (via schema_manager), degrading to
caller-supplied classic defaults when unavailable. This retires the
hand-maintained _CLASSIC_FONT_DEFAULTS dicts that shipped broken twice.

The loader is a superset of the four per-plugin variants: alias
resolution (baseball), truetype for TTF/OTF/BDF (FreeType loads BDF at
native size), .pil sidecar fallback for BDF (football), fallback font,
PIL default — never raises. Also introduces the text_color convention
([r,g,b], absent = keep the plugin's hardcoded color).

BasePlugin gains a lazy style_resolver property (invalidated on config
change) and element_style() sugar; standalone helpers receive the
resolver from their owning plugin.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam

* feat(element-style): x-style-elements schema expansion + color provenance

Plugins can now declare styleable display elements once, compactly, on
their customization schema ("x-style-elements") instead of hand-copying
the ~50-line font/font_size/text_color/offset property blocks (currently
duplicated 52x across the plugin monorepo). SchemaManager.load_schema
expands declarations before caching, so the config form, save path,
validation, and defaults generation all see the same shape; the single
expansion implementation lives in src.element_style and is also applied
by defaults_from_schema_file, keeping the web UI's view and a plugin's
raw-schema-file view of the defaults provably identical (parity test).

Generated blocks use only widgets the config form already renders
(font-selector, color-picker, number inputs) and update x-propertyOrder
when present (the template only renders listed keys). Expansion is
idempotent, never mutates its input or the cached/on-disk schema, and a
hand-written block for the same element always wins.

Color gets the same provenance rule as fonts: the web form always posts
the RGB inputs, so a saved config carries the schema-default color
whether or not the user touched it — the resolver now only honors a
color that DIFFERS from the schema default, and keeps the plugin's
classic (possibly state-dependent, e.g. gold-on-touchdown) color
otherwise.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam

* feat(web): live plugin preview on the config page

Adds POST /api/v3/plugins/preview: renders a plugin headlessly with the
CANDIDATE (unsaved) config and returns a PNG — so users can see exactly
what the panel will show before saving, at their real panel size (from
display.hardware) or a chosen test size.

Two extractions make it drift-proof rather than parallel-implemented:
- dev_server's _render_once moves to
  src/plugin_system/testing/render_service.py (pure PIL via
  VisualTestDisplayManager, install_deps=False always — safe in the web
  process, which never touches display hardware); the dev server now
  wraps it.
- save_plugin_config's ~350-line form->config conversion is extracted
  verbatim as parse_plugin_config_form and shared by the preview
  endpoint, so preview and save can never interpret the form
  differently.

update() is skipped by default (no network on the request thread);
plugins with a test/harness.json get their mock-data fixture primed
instead, and ?skip_update=0 opts into a live update. The config page
gains a Live Preview panel (HTMX hx-include of the existing form, size
selector, pixelated img fragment) — works for every plugin with zero
per-plugin code, including the x-style-elements font/size/color fields.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam

* fix(web): expand x-style-elements in the config-page form render too

pages_v3's plugin-config partial loads config_schema.json directly from
disk rather than through SchemaManager.load_schema, so declared style
elements expanded everywhere EXCEPT the form the user actually sees.
Found live on the devpi: the API served the expanded schema and the
save path validated against it, but the config page rendered no
font/color/offset fields. Apply the same (idempotent) expansion here.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam

* fix(web): preview size selector — htmx caches hx-post, use hx-vals

Found live on the devpi: selecting a preview size did nothing because
htmx snapshots the request path when it processes the button, so the
size dropdown's onchange mutation of hx-post never took effect. The size
now travels as a __preview_size=WxH form field attached via hx-vals
(evaluated at request time); the endpoint honors it (query args still
win for API callers) and strips it before form parsing so it can't leak
into the candidate config.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam

* fix(web): allowlist plugin_id in the preview endpoint's dir lookup

Same defense as the dev server's find_plugin_dir (and pages_v3's
existing pattern): plugin_id arrives in request input and is used to
build filesystem paths — reject anything outside ^[a-zA-Z0-9_-]{1,64}$
before touching the filesystem.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam

* fix: harden preview + resolver edges found in self-review

- extract_schema_defaults now matches SchemaManager's array handling
  ([] for arrays without defaults, [item] for item-level defaults) —
  the documented parity guarantee was false for array-typed properties;
  parity test extended to cover them.
- render_plugin_once calls plugin cleanup() in a finally (image captured
  first — cleanup may clear the canvas): preview instances could leak
  sessions/threads per request in the long-running web process.
- Preview JSON path deep-merges the candidate onto the saved config,
  matching the form path — a shallow update() silently dropped saved
  sibling values in any nested section the candidate touched.
- Preview render is bounded (15s, 504 on timeout, daemonized worker):
  a hanging plugin display() no longer pins a web worker forever.
- ElementStyleResolver.is_for(config): public staleness check for
  callers that cache a resolver, instead of poking _config.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam

* fix(render-service): adopt the dev server's exception-detail policy

Port c6962701's convention into the shared render service (which
supersedes the inline _render_once it patched): full tracebacks to the
server log via exc_info, only the exception class name to the client.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam

---------

Co-authored-by: Chuck <chuck@example.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-07-12 10:40:03 -04:00
ChuckandClaude Fable 5 a55504c0af fix(layout): bound the fit cache; never alias the source image in fits
Two latent issues found in a self-review pass:

- LayoutContext._fit_cache was an unbounded dict (the image cache got an
  LRU cap, the text-fit cache didn't). Cache keys embed the fitted TEXT,
  so a plugin fitting changing strings — a live game clock, a ticker —
  on a 24/7 service grows it forever. Now LRU-bounded at 512 entries via
  the same pattern as the image cache.

- fit_image returned the caller's ORIGINAL image object when the source
  was already RGBA at target size (contain/fill_height, no ink crop).
  ImageFitResult is documented as an independent copy, and LayoutContext
  caches results — an aliased image lets later mutations of the source
  corrupt cached fits (or vice versa). Copy in that branch.

Both covered by new regression tests.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam
2026-07-12 10:19:47 -04:00
ChuckBuilds bd17fad578 fix: remove unused imports flagged by Codacy
Union in adaptive_images.py and field in adaptive_layout.py are both
imported but never used -- the last two Codacy findings on this PR,
matching the same fix already applied on PR #396.
2026-07-12 10:07:18 -04:00
ChuckBuilds 7d5044c315 fix(dev-server): remove conditional-reassignment ambiguity in plugin_dir resolution
CodeQL's path-injection flow still traced through _parse_render_request
after the scandir fix -- the tainted find_plugin_dir() result and the
scandir-derived _trusted_plugin_dir() result shared the same variable
name (plugin_dir), reassigned only on the truthy branch. That merge
point apparently isn't treated as a barrier by the flow analysis, so it
kept tracing the pre-reassignment value through to the manifest open().

Split into two distinct names -- candidate_dir (tainted, used only to
call _trusted_plugin_dir) and trusted_dir (the only name used for any
downstream file access) -- so there's no reassigned variable for the
flow to walk through.
2026-07-12 09:53:09 -04:00
ChuckBuilds c696270182 fix(dev-server): use os.scandir for path-injection barrier, redact stack traces from render responses
CodeQL doesn't model Path.iterdir() as a taint-clearing enumeration the
way it does os.scandir() -- _trusted_plugin_dir's iterdir-based rebuild
still traced plugin_id through to the manifest.json open(). Switched to
scandir, matching the pattern already verified clean on PR #396.

Also stops surfacing raw exception text (update()/display() failures)
in the JSON render response -- logs full detail server-side via
exc_info instead, returning only the exception class name to the
client. And drops path values from three plugin_loader debug/error
logs that CodeQL flags as clear-text-logging of externally-influenced
data, keeping plugin_id (not flagged) for context.
2026-07-12 09:43:36 -04:00
ChuckandClaude Fable 5 a11c59698a fix(dev-server): derive plugin dir from trusted directory listings
CodeQL's barrier-guard recognition doesn't see a startswith check
inside an any() comprehension, so the normalize-and-prefix approach
still flagged. Break the taint outright instead: after lookup, re-derive
the directory by enumerating the search dirs (iterdir) and matching by
path equality — the Path used for all downstream file access is built
solely from trusted listings, never from request input.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam
2026-07-11 12:45:17 -04:00
ChuckandClaude Fable 5 149f1daa1e fix(dev-server): inline normpath containment barrier before render
CodeQL doesn't credit the sanitization inside find_plugin_dir along
this flow; apply its documented barrier (normpath + startswith against
the allowed roots) inline in _parse_render_request, on the exact path
that reaches the render/load sinks.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam
2026-07-11 12:41:35 -04:00
ChuckandClaude Fable 5 68d5540985 fix(dev-server): lexical containment check on resolved plugin dirs
CodeQL doesn't recognize the interprocedural allowlist as a
path-injection barrier; add the canonical one — normalize (without
following symlinks, since dev plugins are commonly symlinked into
plugins/) and require the result to stay inside the search dir.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam
2026-07-11 12:38:46 -04:00
ChuckandClaude Fable 5 72b443d541 fix(dev-server): allowlist plugin_id before any path lookup
CodeQL (py/path-injection): plugin_id arrives in request input and flows
into filesystem paths via find_plugin_dir. Gate it with the same
^[a-zA-Z0-9_-]{1,64}$ allowlist the web UI's pages_v3 uses, at the
single choke point every route resolves through.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam
2026-07-11 12:30:17 -04:00
ChuckandClaude Fable 5 c5f5e25150 fix: address CodeRabbit review on PR #393
- docs: scope the self.layout note to BasePlugin subclasses (others build
  a LayoutContext directly) and make explicit that adaptive layout is
  opt-in — classic rendering stays unless a plugin adopts the APIs.
- dev_server: broaden the render-request catch (a bad manifest.json now
  returns a clean 400 instead of an unhandled 500) and stop echoing raw
  exception text in the loader-failure responses — full tracebacks go to
  the dev server's console log instead.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam
2026-07-11 10:40:11 -04:00
ChuckandClaude Fable 5 278d757de0 docs: document scoreboard_regions' center-reserve and score-bleed params
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-11 09:00:37 -04:00
ChuckandClaude Fable 5 6b47599c4a fix(layout): scoreboard_regions reserves real center space at 2:1 aspect ratios
logo_slot = min(height, width // 2) has a blind spot: at exactly 2:1
aspect ratio (width == 2 * height -- a very common shape: two, four, or
more square modules stacked into a taller panel) width // 2 and height
are equal, so the two logo slots claim the ENTIRE width and leave zero
pixels for a center column, no matter how large the panel gets. Not a
'small panel' problem -- 96x48, 128x64, and 256x128 (all exactly 2:1) hit
it identically, while the 128x32 design baseline and panels like 192x48
or 256x32 never do, because height is already the tighter constraint
there.

Two new parameters fix it in the one shared helper every scoreboard-style
plugin composes through:

- min_center_fraction / min_center_design_px reserve at least
  max(width * fraction, design_px * ctx.scale) for the center column,
  capping logo_slot further when needed. The scaled design-px term
  matters on small panels where a flat fraction alone reserves too little
  absolute space.
- score_bleed_fraction extends the score's own fit box (not the logo
  slots themselves) a controlled amount into each side -- the same way
  real broadcast scoreboards let a big score number's edges cross into
  the team marks flanking it. Without this the reserve alone can still be
  too narrow for a short score to render without truncating.

score_area is now genuinely narrower than the full card width (previously
identical to status_band/detail_band, which still span the full width and
overlay the logos -- short text there was never the problem).

Verified against the full harness size spread: a real game score like
'17-21' never needs ellipsis at any tested 2:1-or-tighter aspect ratio
(test_score_never_needs_ellipsis_for_a_short_score), and wide panels
(128x32/192x48/256x32-style) are provably unaffected.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-11 09:00:37 -04:00
ChuckandClaude Fable 5 87ff97d006 feat(layout): fit_text_proportional gains an axis-specific scale override
self.scale (min(width_ratio, height_ratio)) is the right conservative
default for anything whose aspect ratio matters, but a caller whose
surrounding composition already scales along a single axis — e.g.
football-scoreboard's logo_slot = min(height, width // 2), which tracks
height alone — needs text sized the same way, or it reads as
under-scaled next to logos that grew on a panel that only got taller
(128x32 -> 128x64: self.scale stays 1.0 since width didn't grow, but
logos still double).

fit_text_proportional(..., scale=None) now accepts an explicit override;
None keeps the existing self.scale default.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-11 09:00:37 -04:00
ChuckandClaude Fable 5 4ab0c871c8 feat(layout): add fit_text_proportional — proportional sizing vs. always-maximize
fit_text always picks the largest ladder rung that fits its box. That's
right when an element owns dedicated space, but wrong when several
independently-fitted elements need to stay visually harmonious as the
panel grows: a score's box might have generous room while a neighboring
logo scales by a fixed geometry factor via px() — fit_text lets the score
balloon out of proportion (even overlapping the logo) even though its
individual pick is technically correct.

fit_text_proportional(text, box, base_size_px, ladder) instead targets
base_size_px * self.scale (the same scale factor px() already uses),
picking the nearest ladder rung at or below that target, still capped to
what fits the box, floored at the smallest rung when the target is below
every rung. Refactored the shared largest-that-fits/ellipsize walk into
_walk_ladder() so fit_text and fit_text_proportional don't duplicate it.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-11 09:00:37 -04:00
ChuckandClaude Fable 5 013a2663e5 feat(layout): add measure_font_crispness — verify a ladder rung isn't blurry
PIL antialiases TTF outlines by default; a 'pixel-style' font only
rasterizes without antialiasing at specific sizes (for PressStart2P:
exact multiples of its 8px design grid). A ladder rung at an unverified
size silently renders blurry on an LED panel — this exact bug shipped in
both text-display's and football-scoreboard's custom TTF ladders
(non-8-multiple PressStart2P sizes, and '5by7.regular'/'4x6-font' at
sizes that were never actually crisp).

measure_font_crispness(font, sample_text) renders the sample and reports
the fraction of ink-bbox pixels that are neither pure black nor pure
white. BDF fonts (real bitmaps) always score 0.0; TTF ladders should be
verified against this before shipping — see the new
TestFontFitting::test_ladder_arcade_is_crisp pattern.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-11 09:00:37 -04:00
ChuckandClaude Fable 5 af96c6ffd6 feat(plugins): adaptive-lib discoverability + advisory version compat warning
Discoverability: re-export the adaptive layout/image API from src.common
(the blessed-helpers package plugin authors already know) — canonical
paths stay src.adaptive_layout / src.adaptive_images so nothing breaks.
Document it in src/common/README.md and cross-link ADAPTIVE_LAYOUT.md
from the developer docs authors actually read (quick reference, API
reference, advanced dev, font manager, dev preview, plugin dev guide);
ADAPTIVE_LAYOUT.md gains adaptive-images, composite-layouts and
preserving-user-customization sections.

Compat: PluginLoader now logs one advisory warning (never raises) when a
plugin's manifest declares a min LEDMatrix version newer than the running
core, checking the min_ledmatrix_version / requires.* / versions[]
spellings found in the wild. Guarded against stale core version numbers.

src/__init__.py __version__ bumped 1.0.0 -> 3.1.0 to match the latest
release tag (v3.1.0) — it had never been updated and the compat check
needs a truthful number. NOTE: verify this matches the intended release
numbering before the next tag.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-11 09:00:37 -04:00
ChuckandClaude Fable 5 923611b836 feat(harness): scale-up fill check, config variants, multi-size dev gallery
Quality gates for adaptive layout:

- fill_metrics()/check_scale_up() in the safety harness: overflow catches
  content too big for a panel, but nothing caught content that stays tiny
  on panels >= 2x the plugin's declared design size. The check measures
  lit-content extents and warns (or fails, when a plugin opts into
  "fill_check": "strict" in test/harness.json) below 50% coverage on the
  doubled axis. Warn-only by default so no existing plugin breaks.

- harness.json "variants": extra runs with config overlays and their own
  golden dirs, so an opt-in mode (e.g. layout_mode: adaptive) is golden-
  tested beside the classic default. check_plugin.py loops base + variants
  and labels variant results mode@name.

- Dev preview server: GET /api/sizes (harness size sample), POST
  /api/render-matrix (render at up to 12 sizes in one call), size-preset
  dropdown, and an "All Sizes" side-by-side gallery in the preview UI.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-11 09:00:37 -04:00
ChuckandClaude Fable 5 dabe7f05bc feat(layout): adaptive image fitting + composite region helpers
Add src/adaptive_images.py — the image counterpart to fit_text:
- fit_image(img, box, mode=contain|cover|fill_height|stretch,
  crop_to_ink, anchor, resample, upscale) promoting the proven plugin
  patterns (football's crop-to-ink fill-height logos, masters' cover
  crop + NEAREST flags, static-image's letterbox). Upscales by default —
  thumbnail()'s downscale-only behavior is why imagery stays tiny on
  big panels.
- draw_fitted_image() pastes aligned within a Region with alpha mask.
- One central Pillow>=9.1 RESAMPLE shim replacing ~15 plugin copies.

LayoutContext.fit_image() caches results per (identity, box size,
options) with a 64-entry LRU; id()-keyed entries pin the source image.
BasePlugin.draw_image() is the one-liner adoption path beside draw_fit.

Composites in adaptive_layout.py: Region.offset() (user x/y-offset
passthrough), scoreboard_regions() (the two-logos-plus-score card math
duplicated across six sports plugins, logo_slot = min(H, W//2)), and
media_row() (art-left/text-right).

Fix LogoHelper's size-blind cache key (stale sizes on panel change);
deprecation note on dead image_utils.py.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-11 09:00:37 -04:00
ChuckandClaude Fable 5 21c0cfa22d feat(layout): adaptive layout & font scaling system for plugins
Add src/adaptive_layout.py — opt-in core helpers so plugins render
legibly on any panel size without hand-tuned per-display layouts:

- Region: integer rect algebra (bands/columns/weighted splits/centering)
  that partitions space so text bands can't overlap by construction
- Font ladders: ordered (family, size) steps known to render crisply
  (LADDER_GRID: X11 BDFs at native sizes; LADDER_ARCADE: PressStart2P at
  8px multiples) — fitting walks the ladder instead of scaling pixel
  fonts fractionally
- LayoutContext: breakpoint tiers, geometry scale vs. a declared design
  size, and cached fit_text/fit_lines/font_for_rows queries

Generalizes the three patterns proven in the field: f1-scoreboard's
scale factor, masters-tournament's tiers, baseball-scoreboard's font
fallback ladder.

Wiring: BasePlugin gains a lazy .layout property and draw_fit();
FontManager gains get_native_bdf_size() and a cache_generation counter;
manifest schema gains display.design_size and requires.display_size
max_width/max_height; 96x48 joins DEFAULT_TEST_SIZES; the bounds-check
harness records negative-coordinate draws; TextHelper's broken
measurement helpers are fixed.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-11 09:00:37 -04:00
16 changed files with 2487 additions and 681 deletions
+8 -54
View File
@@ -200,62 +200,16 @@ def _render_once(plugin_id, plugin_dir, manifest, config, mock_data, width, heig
skip_update):
"""Render one plugin at one size. Returns the /api/render response dict.
A fresh plugin instance per call, mirroring the safety harness, so sizes
never share state.
Thin wrapper over the shared render service (also used by the web UI's
config-page preview); a fresh plugin instance per call, mirroring the
safety harness, so sizes never share state.
"""
from src.plugin_system.testing import VisualTestDisplayManager, MockCacheManager, MockPluginManager
from src.plugin_system.plugin_loader import PluginLoader
from src.plugin_system.testing.render_service import render_plugin_once
display_manager = VisualTestDisplayManager(width=width, height=height)
cache_manager = MockCacheManager()
plugin_manager = MockPluginManager()
# Pre-populate cache with mock data
for key, value in mock_data.items():
cache_manager.set(key, value)
loader = PluginLoader()
errors = []
warnings = []
plugin_instance, _module = loader.load_plugin(
plugin_id=plugin_id,
manifest=manifest,
plugin_dir=plugin_dir,
config=config,
display_manager=display_manager,
cache_manager=cache_manager,
plugin_manager=plugin_manager,
install_deps=False,
)
start_time = time.time()
# Run update()
if not skip_update:
try:
plugin_instance.update()
except Exception as e:
logger.warning("update() raised for plugin %s", plugin_id, exc_info=True)
warnings.append(f"update() raised: {type(e).__name__} — see server log")
# Run display()
try:
plugin_instance.display(force_clear=True)
except Exception as e:
logger.warning("display() raised for plugin %s", plugin_id, exc_info=True)
errors.append(f"display() raised: {type(e).__name__} — see server log")
render_time_ms = round((time.time() - start_time) * 1000, 1)
return {
'image': f'data:image/png;base64,{display_manager.get_image_base64()}',
'width': width,
'height': height,
'render_time_ms': render_time_ms,
'errors': errors,
'warnings': warnings,
}
return render_plugin_once(
plugin_id, plugin_dir, manifest=manifest, config=config,
mock_data=mock_data, width=width, height=height,
skip_update=skip_update)
def _trusted_plugin_dir(plugin_dir: Path) -> Optional[Path]:
+9 -58
View File
@@ -502,10 +502,7 @@ class DisplayController:
# Run plugin updates inside the Vegas loop so the inter-iteration
# gap is <1 ms (nothing left for _tick_plugin_updates() to do).
# Use the Vegas-aware variant so plugins that got fresh data are
# hot-swapped into the scroll promptly instead of waiting for the
# next full cycle.
self.vegas_coordinator.set_update_callback(self._tick_plugin_updates_for_vegas)
self.vegas_coordinator.set_update_callback(self._tick_plugin_updates)
# Wire multi-display sync into Vegas render pipeline
follower_pos = self.config.get("sync", {}).get("follower_position", "left")
@@ -624,28 +621,18 @@ class DisplayController:
current_day = current_time.strftime('%A').lower() # e.g. 'monday'
current_time_only = current_time.time()
# Check if per-day schedule is configured
days_config = schedule_config.get('days')
# Determine which schedule to use. Respect an explicit 'mode' field
# (like the dim schedule does) so a stray/legacy 'days' dict left over
# from config migration or a prior per-day setup can't silently
# override a user's Global schedule selection.
mode = schedule_config.get('mode')
mode_normalized = mode.replace('_', '-') if mode else None
# Determine which schedule to use
use_per_day = False
if mode_normalized == 'global':
use_per_day = False
elif mode_normalized == 'per-day':
use_per_day = bool(days_config and current_day in days_config)
elif days_config:
# No explicit mode recorded (legacy config) - fall back to
# inferring from presence of a 'days' dict for the current day.
if current_day in days_config:
if days_config:
# Check if days dict is not empty and contains current day
if days_config and current_day in days_config:
use_per_day = True
else:
elif days_config:
# Days dict exists but doesn't have current day - fall back to global
logger.debug("Per-day schedule exists but %s not configured, using global schedule", current_day)
if use_per_day:
@@ -841,42 +828,6 @@ class DisplayController:
if hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
self.plugin_manager.health_tracker.record_failure(plugin_id, exc)
def _tick_plugin_updates_for_vegas(self) -> None:
"""Run scheduled plugin updates and tell Vegas mode which plugins
actually got fresh data, so it can hot-swap them into the scroll
without waiting for a full cycle to complete.
Used as the Vegas coordinator's update callback instead of the plain
_tick_plugin_updates() so that a live score change is reflected in
the ticker within a few seconds rather than at the next cycle
boundary (which, depending on min/max_cycle_duration, can be
minutes away). Restores wiring that PR #299 added and PR #330's
sync-mode refactor inadvertently dropped: coordinator.mark_plugin_updated()
has been unreachable dead code since.
Delegates the before/after plugin_last_update snapshot to
PluginManager.run_scheduled_updates_with_changes() so the snapshot,
update pass, and diff are lock-protected against this callback's own
background update-tick thread racing the main render loop.
"""
if not self.plugin_manager or not hasattr(self.plugin_manager, "run_scheduled_updates_with_changes"):
self._tick_plugin_updates()
return
updated = self.plugin_manager.run_scheduled_updates_with_changes()
vc = getattr(self, "vegas_coordinator", None)
if vc is None:
return
if updated:
logger.info("Vegas update tick: %d plugin(s) updated: %s", len(updated), updated)
for plugin_id in updated:
try:
vc.mark_plugin_updated(plugin_id)
except Exception: # pylint: disable=broad-except
logger.exception("Error marking plugin %s updated for Vegas", plugin_id)
def _tick_plugin_updates(self):
"""Run scheduled plugin updates if the plugin manager supports them."""
if not self.plugin_manager:
+552
View File
@@ -0,0 +1,552 @@
"""Universal per-element style resolution for plugin customization.
Plugins expose per-element user customization under ``config['customization']``:
"customization": {
"score_text": {"font": "PressStart2P-Regular.ttf", "font_size": 10,
"text_color": [255, 255, 255]},
"layout": {"score": {"x_offset": 2, "y_offset": 0}}
}
Before this module, every plugin re-implemented the same three pieces
a font loader, an x/y-offset reader, and (for adaptive layout mode) a
"did the user actually override this?" check. The loaders diverged four
ways across the sports plugins and music, the offset reader was copied
twice, and the override check is subtle enough that it shipped broken
twice: the web UI's save flow (schema_manager.merge_with_defaults) writes
the FULL schema default object into config.json on every save, and the
plugin manager merges defaults into ``config`` again before instantiation,
so a key being *present* never means the user set it. The only correct
test is "present AND different from the schema default", which requires
knowing the schema defaults previously a hand-maintained dict per plugin.
This module is that logic, once:
resolver = ElementStyleResolver(config, schema_defaults)
style = resolver.style('score_text', classic_font='PressStart2P-Regular.ttf',
classic_size=10)
style.font # loaded PIL font, ready for draw.text
style.user_forced # True only for a genuine user override
dx, dy = resolver.offset('score')
``BasePlugin.element_style()`` wires this up automatically (schema defaults
come from the plugin's own config_schema.json via the schema manager).
Standalone helper classes (e.g. a plugin's GameRenderer) should receive a
resolver from their owning plugin rather than build one themselves.
Deliberately pure PIL + stdlib: no imports from the plugin system or web
layer, so it is usable from any renderer and trivially testable.
"""
import json
import logging
import os
from dataclasses import dataclass
from typing import Any, Dict, Optional, Tuple, Union
from PIL import ImageFont
logger = logging.getLogger(__name__)
# Font-family aliases accepted in customization configs. Filenames pass
# through unchanged. (Supersedes the per-plugin copies in the baseball
# plugin; keep names in sync with the web UI's /fonts/catalog so the
# font-selector widget and this loader agree.)
FONT_ALIASES: Dict[str, str] = {
"press_start": "PressStart2P-Regular.ttf",
"four_by_six": "4x6-font.ttf",
"five_by_seven": "5x7.bdf",
}
DEFAULT_FONTS_DIR = os.path.join("assets", "fonts")
DEFAULT_FALLBACK_FONT = "PressStart2P-Regular.ttf"
PILFont = Union[ImageFont.FreeTypeFont, ImageFont.ImageFont]
def resolve_font_name(font_name: str) -> str:
"""Resolve a font family alias to its filename, leaving filenames as-is."""
return FONT_ALIASES.get(font_name, font_name)
def extract_schema_defaults(schema: Dict[str, Any]) -> Dict[str, Any]:
"""Nested defaults dict from a JSON Schema (mirrors
SchemaManager.extract_defaults_from_schema, kept here so this module
stays importable without the plugin system the parity test in
test_element_style.py guards against drift).
An object property carrying its own ``default`` short-circuits recursion,
and array-typed properties without one default to ``[]`` (or a
single-item list when the items schema declares a default), matching
the schema manager's behavior exactly.
"""
defaults: Dict[str, Any] = {}
for key, prop in (schema.get("properties") or {}).items():
if not isinstance(prop, dict):
continue
if "default" in prop:
defaults[key] = prop["default"]
elif prop.get("type") == "object" and "properties" in prop:
nested = extract_schema_defaults(prop)
if nested:
defaults[key] = nested
elif prop.get("type") == "array" and "items" in prop:
items = prop.get("items")
if isinstance(items, dict) and "default" in items and \
not (items.get("type") == "object" and "properties" in items):
defaults[key] = [items["default"]]
else:
defaults[key] = []
return defaults
def defaults_from_schema_file(schema_path: str) -> Dict[str, Any]:
"""Schema defaults straight from a plugin's own config_schema.json.
Plugins that hand a resolver to standalone helper classes should build
it with this, pointed at their own schema file it works identically
in production, the test harness, and the dev server, unlike the plugin
manager's schema manager (absent under mocks). x-style-elements
declarations are expanded first, so declared elements' defaults are
included exactly as the web UI's schema manager sees them. Returns {}
on any error.
"""
try:
with open(schema_path, "r", encoding="utf-8") as f:
schema = json.load(f)
if isinstance(schema, dict):
return extract_schema_defaults(expand_style_elements(schema))
except Exception as e:
logger.debug("Could not load schema defaults from %s: %s",
schema_path, e)
return {}
# ---------------------------------------------------------------------------
# x-style-elements schema expansion
# ---------------------------------------------------------------------------
#
# A plugin declares its styleable display elements ONCE, compactly, on its
# customization object instead of hand-copying ~50-line property blocks:
#
# "customization": {
# "type": "object",
# "x-style-elements": {
# "score_text": {
# "title": "Game Score",
# "font": {"default": "PressStart2P-Regular.ttf"},
# "size": {"default": 10, "min": 4, "max": 16},
# "color": true, # or {"default": [r,g,b]}
# "offsets": true
# }
# }
# }
#
# expand_style_elements() turns each declaration into full font/font_size/
# text_color/layout-offset property blocks (marked "x-style-managed": true)
# using widgets the web config form already renders. The declaration stays
# in the schema — it doubles as the element registry for tooling. Expansion
# is idempotent, and a hand-written property block for the same element
# always wins over the generated one.
#
# SchemaManager.load_schema() applies this at serve time (so the web form,
# save path, validation, and defaults generation all see the expanded
# shape), and defaults_from_schema_file() applies it when plugins read
# their own schema — one implementation, no drift.
def get_style_elements(schema: Dict[str, Any]) -> Dict[str, Any]:
"""The x-style-elements declaration from a schema ({} if none)."""
try:
decl = schema.get("properties", {}).get("customization", {}).get("x-style-elements")
return decl if isinstance(decl, dict) else {}
except AttributeError:
return {}
def expand_style_elements(schema: Dict[str, Any]) -> Dict[str, Any]:
"""Expand x-style-elements into full customization property blocks.
Returns the schema unchanged (same object) when there is nothing to
expand; otherwise returns an expanded DEEP COPY, leaving the input
untouched. Never raises on any error the original schema is returned
so a malformed declaration can't take a plugin down.
"""
import copy
try:
declarations = get_style_elements(schema)
if not declarations:
return schema
schema = copy.deepcopy(schema)
customization = schema["properties"]["customization"]
properties = customization.setdefault("properties", {})
order = customization.get("x-propertyOrder")
offset_elements = []
for element_key, declaration in declarations.items():
if not isinstance(declaration, dict):
continue
if declaration.get("offsets") is True:
offset_elements.append((element_key, declaration))
if element_key in properties:
# Hand-written (or previously expanded) block wins.
continue
properties[element_key] = _style_element_block(element_key, declaration)
if isinstance(order, list) and element_key not in order:
# Keep generated elements ahead of the layout section.
insert_at = order.index("layout") if "layout" in order else len(order)
order.insert(insert_at, element_key)
if offset_elements:
_expand_offset_blocks(properties, order, offset_elements)
return schema
except Exception as e:
logger.error("x-style-elements expansion failed: %s", e)
return schema
def _style_element_block(element_key: str, declaration: Dict[str, Any]) -> Dict[str, Any]:
"""One generated customization.<element> property block."""
title = declaration.get("title") or element_key.replace("_", " ").title()
font_decl = declaration.get("font") if isinstance(declaration.get("font"), dict) else {}
size_decl = declaration.get("size") if isinstance(declaration.get("size"), dict) else {}
block_properties: Dict[str, Any] = {
"font": {
"type": "string",
"title": "Font Family",
"description": "Select the font to use",
"x-widget": "font-selector",
"default": font_decl.get("default", DEFAULT_FALLBACK_FONT),
},
"font_size": {
"type": "integer",
"title": "Font Size",
"description": ("Font size in pixels (BDF fonts are fixed-size "
"and ignore this)"),
"minimum": size_decl.get("min", 4),
"maximum": size_decl.get("max", 32),
"default": size_decl.get("default", 8),
},
}
block_order = ["font", "font_size"]
color_decl = declaration.get("color")
if color_decl:
default_color = [255, 255, 255]
if isinstance(color_decl, dict) and isinstance(color_decl.get("default"), list):
default_color = color_decl["default"]
# The default doubles as the "untouched" sentinel: the resolver only
# honors a color that DIFFERS from it, so untouched saves (the web
# form always posts the RGB inputs) can't clobber a plugin's
# semantic/state-dependent colors.
block_properties["text_color"] = {
"type": "array",
"title": "Text Color",
"description": "RGB color as [red, green, blue] (0-255 each)",
"items": {"type": "integer", "minimum": 0, "maximum": 255},
"minItems": 3,
"maxItems": 3,
"x-widget": "color-picker",
"default": default_color,
}
block_order.append("text_color")
return {
"type": "object",
"title": title,
"description": f"Style settings for {title}",
"x-style-managed": True,
"properties": block_properties,
"x-propertyOrder": block_order,
"additionalProperties": False,
}
def _expand_offset_blocks(properties: Dict[str, Any], order,
offset_elements) -> None:
"""Generate customization.layout.<element> x/y offset blocks."""
layout = properties.get("layout")
if not isinstance(layout, dict):
layout = {
"type": "object",
"title": "Layout Positioning",
"description": ("Adjust X,Y coordinate offsets for elements. "
"Values are relative to default positions; "
"negative moves left/up, positive right/down."),
"x-style-managed": True,
"properties": {},
"additionalProperties": False,
}
properties["layout"] = layout
if isinstance(order, list) and "layout" not in order:
order.append("layout")
layout_properties = layout.setdefault("properties", {})
layout_order = layout.get("x-propertyOrder")
for element_key, declaration in offset_elements:
if element_key in layout_properties:
continue # hand-written layout entry wins
title = declaration.get("title") or element_key.replace("_", " ").title()
layout_properties[element_key] = {
"type": "object",
"title": title,
"x-style-managed": True,
"properties": {
"x_offset": {
"type": "integer",
"title": "X Offset",
"description": "Horizontal offset in pixels (default: 0)",
"default": 0,
},
"y_offset": {
"type": "integer",
"title": "Y Offset",
"description": "Vertical offset in pixels (default: 0)",
"default": 0,
},
},
"additionalProperties": False,
}
if isinstance(layout_order, list) and element_key not in layout_order:
layout_order.append(element_key)
def load_font(font_name: str, size: int, *,
fonts_dir: str = DEFAULT_FONTS_DIR,
fallback_font: str = DEFAULT_FALLBACK_FONT) -> PILFont:
"""Load a font by name at a pixel size, never raising.
Resolution order:
1. alias -> filename (``FONT_ALIASES``)
2. ``ImageFont.truetype`` handles .ttf/.otf, and .bdf too (FreeType
loads BDF strikes at their native size; a non-native size raises
"invalid pixel size" and falls through)
3. for .bdf: a pre-converted ``.pil`` sidecar via ``ImageFont.load``
4. ``fallback_font`` at the requested size
5. ``ImageFont.load_default()``
"""
font_name = resolve_font_name(font_name or "")
font_path = os.path.join(fonts_dir, font_name)
lower = font_name.lower()
if os.path.exists(font_path):
try:
return ImageFont.truetype(font_path, size)
except Exception as e:
logger.debug("truetype failed for %s@%s: %s", font_name, size, e)
if lower.endswith(".bdf"):
pil_path = font_path.rsplit(".", 1)[0] + ".pil"
if os.path.exists(pil_path):
try:
return ImageFont.load(pil_path)
except Exception as e:
logger.debug("PIL sidecar failed for %s: %s", pil_path, e)
logger.warning(
"BDF font %s could not be loaded at size %s (BDF fonts are "
"fixed-size; font_size must match the native size). Falling "
"back to %s.", font_name, size, fallback_font)
else:
logger.warning("Font file not found: %s, falling back to %s",
font_path, fallback_font)
fallback_path = os.path.join(fonts_dir, resolve_font_name(fallback_font))
try:
return ImageFont.truetype(fallback_path, size)
except Exception as e:
logger.warning("Fallback font %s failed (%s); using PIL default",
fallback_font, e)
return ImageFont.load_default()
@dataclass(frozen=True)
class ElementStyle:
"""Resolved style for one display element."""
font: PILFont
font_name: str
font_size: int
#: The user's color when they genuinely changed it, else the plugin's
#: classic color (which may be state-dependent — e.g. a score that turns
#: gold on a touchdown — so an untouched schema default must never
#: clobber it; the web form always posts the color inputs).
color: Optional[Tuple[int, int, int]]
#: Additive (dx, dy) translation from customization.layout offsets.
offset: Tuple[int, int]
#: True when the configured value genuinely differs from the schema
#: default (NOT merely present — saved configs always contain defaults).
user_forced_font: bool
user_forced_size: bool
user_forced_color: bool = False
@property
def user_forced(self) -> bool:
"""True when the user pinned this element's font or size; adaptive
layouts must use the font as-is instead of ladder-fitting. (Color is
deliberately excluded it never affects sizing.)"""
return self.user_forced_font or self.user_forced_size
def _as_int(value: Any, default: int) -> int:
"""Int coercion tolerant of floats and numeric strings from configs."""
if value is None:
return default
if isinstance(value, bool):
return default
if isinstance(value, (int, float)):
return int(value)
try:
return int(float(value))
except (TypeError, ValueError):
return default
def _as_color(value: Any) -> Optional[Tuple[int, int, int]]:
"""[r, g, b] list/tuple -> tuple; anything else -> None."""
if isinstance(value, (list, tuple)) and len(value) == 3:
try:
return tuple(max(0, min(255, int(c))) for c in value)
except (TypeError, ValueError):
return None
return None
class ElementStyleResolver:
"""Resolves per-element fonts, colors and offsets from a plugin config.
``schema_defaults`` is the nested defaults dict extracted from the
plugin's config_schema.json (SchemaManager.extract_defaults_from_schema).
It is the reference for the user-override check: a configured value
equal to its schema default is treated as untouched, because the save
flow persists all defaults. When ``schema_defaults`` is empty (older
cores, unit tests), the check degrades to comparing against the
``classic_*`` values the caller supplies.
"""
def __init__(self, config: Optional[Dict[str, Any]],
schema_defaults: Optional[Dict[str, Any]] = None, *,
fonts_dir: str = DEFAULT_FONTS_DIR,
fallback_font: str = DEFAULT_FALLBACK_FONT):
self._config = config if isinstance(config, dict) else {}
self._defaults = schema_defaults if isinstance(schema_defaults, dict) else {}
self._fonts_dir = fonts_dir
self._fallback_font = fallback_font
self._cache: Dict[Any, ElementStyle] = {}
# -- internals ----------------------------------------------------
def _element_config(self, element_key: str) -> Dict[str, Any]:
cust = self._config.get("customization")
if not isinstance(cust, dict):
return {}
element = cust.get(element_key)
return element if isinstance(element, dict) else {}
def _element_defaults(self, element_key: str) -> Dict[str, Any]:
cust = self._defaults.get("customization")
if not isinstance(cust, dict):
return {}
element = cust.get(element_key)
return element if isinstance(element, dict) else {}
# -- public API ---------------------------------------------------
def style(self, element_key: str, *, classic_font: str, classic_size: int,
classic_color: Optional[Tuple[int, int, int]] = None) -> ElementStyle:
"""Resolve the style for one element.
``classic_font``/``classic_size``/``classic_color`` are the plugin's
hardcoded defaults for this element used when the config has no
value, and as the override reference when schema defaults are
unavailable.
"""
cache_key = (element_key, classic_font, classic_size, classic_color)
cached = self._cache.get(cache_key)
if cached is not None:
return cached
element_cfg = self._element_config(element_key)
element_defaults = self._element_defaults(element_key)
configured_font = element_cfg.get("font")
configured_size = element_cfg.get("font_size")
# Reference for "did the user change it": schema default when known,
# else the plugin's classic default.
reference_font = element_defaults.get("font", classic_font)
reference_size = _as_int(element_defaults.get("font_size"), classic_size)
user_forced_font = (configured_font is not None
and configured_font != reference_font)
user_forced_size = (configured_size is not None
and _as_int(configured_size, reference_size) != reference_size)
font_name = configured_font if configured_font is not None else classic_font
font_size = _as_int(configured_size, classic_size)
font = load_font(font_name, font_size, fonts_dir=self._fonts_dir,
fallback_font=self._fallback_font)
# Color follows the same provenance rule as fonts: the web form
# always posts the RGB inputs, so a saved config carries the schema
# default whether or not the user touched it — only a value that
# DIFFERS from the schema default is a real override. Otherwise keep
# classic_color, which may be state-dependent (semantic colors like
# a gold touchdown score) and must not be clobbered by a default.
configured_color = _as_color(element_cfg.get("text_color"))
default_color = _as_color(element_defaults.get("text_color"))
if configured_color is None:
user_forced_color = False
elif default_color is None:
# no schema default to compare against — presence is intent
user_forced_color = True
else:
user_forced_color = configured_color != default_color
color = configured_color if user_forced_color else classic_color
resolved = ElementStyle(
font=font, font_name=font_name, font_size=font_size,
color=color, offset=self.offset(element_key),
user_forced_font=user_forced_font, user_forced_size=user_forced_size,
user_forced_color=user_forced_color,
)
self._cache[cache_key] = resolved
return resolved
def offset_value(self, element_key: str, axis: str, default: int = 0) -> int:
"""One offset axis for an element (e.g. 'x_offset', 'away_x_offset').
Reads ``customization.layout.<element>`` first (the deployed sports
convention), falling back to ``customization.<element>`` for plugins
that keep offsets on the element itself.
"""
cust = self._config.get("customization")
if not isinstance(cust, dict):
return default
layout = cust.get("layout")
if isinstance(layout, dict):
element = layout.get(element_key)
if isinstance(element, dict) and axis in element:
return _as_int(element.get(axis), default)
element = cust.get(element_key)
if isinstance(element, dict) and axis in element:
return _as_int(element.get(axis), default)
return default
def offset(self, element_key: str) -> Tuple[int, int]:
"""(dx, dy) additive translation for an element; (0, 0) when unset."""
return (self.offset_value(element_key, "x_offset"),
self.offset_value(element_key, "y_offset"))
def is_for(self, config: Optional[Dict[str, Any]]) -> bool:
"""True when this resolver was built over exactly this config object.
Callers that cache a resolver (plugins whose config dict gets
REPLACED on config change) use this to decide when to rebuild
preferred over reaching into the private ``_config``.
"""
return self._config is config
def clear_cache(self) -> None:
self._cache.clear()
+61
View File
@@ -270,6 +270,64 @@ class BasePlugin(ABC):
return (int(width), int(height))
return DEFAULT_DESIGN_SIZE
# -------------------------------------------------------------------------
# Element style resolution (per-element user customization)
# -------------------------------------------------------------------------
@property
def style_resolver(self) -> Any:
"""
ElementStyleResolver for this plugin's customization config.
Lazily built with the schema defaults from this plugin's own
config_schema.json (via the plugin manager's schema manager), so
"did the user override this font?" is answered correctly even though
saved configs always contain the schema defaults. Rebuilt when the
config changes (see on_config_change). Pass it into standalone helper
classes (game renderers etc.) instead of letting them build their own.
See src/element_style.py.
"""
from src.element_style import ElementStyleResolver
cached = getattr(self, "_style_resolver", None)
if cached is not None:
return cached
schema_defaults: Dict[str, Any] = {}
schema_manager = getattr(self.plugin_manager, "schema_manager", None)
if schema_manager is not None:
try:
schema = schema_manager.load_schema(self.plugin_id)
if schema:
schema_defaults = schema_manager.extract_defaults_from_schema(schema)
except Exception as e:
self.logger.debug("Schema defaults unavailable for %s: %s",
self.plugin_id, e)
resolver = ElementStyleResolver(self.config, schema_defaults)
self._style_resolver = resolver
return resolver
def element_style(self, element_key: str, *, classic_font: str,
classic_size: int,
classic_color: Optional[tuple] = None) -> Any:
"""
Resolved font/color/offset for one display element, honoring the
user's customization.<element> config. classic_* are this plugin's
hardcoded defaults for the element.
Example:
style = self.element_style('title_text',
classic_font='PressStart2P-Regular.ttf',
classic_size=8)
draw.text((x + style.offset[0], y + style.offset[1]),
title, font=style.font,
fill=style.color or (255, 255, 255))
"""
return self.style_resolver.style(element_key, classic_font=classic_font,
classic_size=classic_size,
classic_color=classic_color)
def get_display_duration(self) -> float:
"""
Get the display duration for this plugin instance.
@@ -732,6 +790,9 @@ class BasePlugin(ABC):
# Update simple flags
self.enabled = self.config.get("enabled", self.enabled)
# Invalidate the cached style resolver — it captured the old config
self._style_resolver = None
def get_info(self) -> Dict[str, Any]:
"""
Return plugin info for display in web UI.
+6 -43
View File
@@ -76,12 +76,6 @@ class PluginManager:
# concurrent mutation (background reconciliation) and reads (requests).
self._discovery_lock = threading.RLock()
# Lock protecting plugin_last_update from concurrent mutation/iteration.
# It's written from run_scheduled_updates()/update_all_plugins() (main
# loop) and read/diffed by run_scheduled_updates_with_changes(), which
# Vegas mode calls from its own background update-tick thread.
self._plugin_last_update_lock = threading.RLock()
# Active plugins
self.plugins: Dict[str, Any] = {}
self.plugin_manifests: Dict[str, Dict[str, Any]] = {}
@@ -323,8 +317,7 @@ class PluginManager:
# Store plugin instance
self.plugins[plugin_id] = plugin_instance
with self._plugin_last_update_lock:
self.plugin_last_update[plugin_id] = 0.0
self.plugin_last_update[plugin_id] = 0.0
# Invalidate cached interval so next tick re-derives it for this plugin
self._update_interval_cache.pop(plugin_id, None)
@@ -436,8 +429,7 @@ class PluginManager:
# Remove from active plugins
del self.plugins[plugin_id]
with self._plugin_last_update_lock:
self.plugin_last_update.pop(plugin_id, None)
self.plugin_last_update.pop(plugin_id, None)
self._update_interval_cache.pop(plugin_id, None)
# Remove main module from sys.modules if present
@@ -706,8 +698,7 @@ class PluginManager:
'recoverable': True,
}
self.logger.warning("Plugin %s update() failed; will retry after interval", plugin_id)
with self._plugin_last_update_lock:
self.plugin_last_update[plugin_id] = failure_time
self.plugin_last_update[plugin_id] = failure_time
self.state_manager.set_state_with_error(plugin_id, PluginState.ENABLED, error_info, error=err)
if self.health_tracker:
self.health_tracker.record_failure(plugin_id, err)
@@ -740,8 +731,7 @@ class PluginManager:
if interval is None:
continue
with self._plugin_last_update_lock:
last_update = self.plugin_last_update.get(plugin_id, 0.0)
last_update = self.plugin_last_update.get(plugin_id, 0.0)
if last_update == 0.0 or (current_time - last_update) >= interval:
# Update state to RUNNING
@@ -772,8 +762,7 @@ class PluginManager:
success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
if success:
with self._plugin_last_update_lock:
self.plugin_last_update[plugin_id] = current_time
self.plugin_last_update[plugin_id] = current_time
self.state_manager.record_update(plugin_id)
# Update state back to ENABLED
self.state_manager.set_state(plugin_id, PluginState.ENABLED)
@@ -786,31 +775,6 @@ class PluginManager:
self.logger.exception("Error updating plugin %s: %s", plugin_id, exc)
self._record_update_failure(plugin_id, exc=exc)
def run_scheduled_updates_with_changes(self, current_time: Optional[float] = None) -> List[str]:
"""
Like run_scheduled_updates(), but also returns the plugin_ids whose
plugin_last_update timestamp actually advanced during this call.
The before/after snapshots and the update pass itself are each
individually lock-protected against concurrent plugin_last_update
mutation (Vegas mode calls this from its own background
update-tick thread, racing the main render loop's plugin updates),
so callers get an atomic "who got fresh data" answer without
reaching into plugin_last_update themselves. The lock is not held
across the update pass so slow/blocking plugin update() calls don't
serialize against other plugin_last_update readers.
"""
with self._plugin_last_update_lock:
old_times = dict(self.plugin_last_update)
self.run_scheduled_updates(current_time)
with self._plugin_last_update_lock:
return [
plugin_id for plugin_id, new_time in self.plugin_last_update.items()
if new_time > old_times.get(plugin_id, 0.0)
]
def update_all_plugins(self) -> None:
"""
Update all enabled plugins.
@@ -833,8 +797,7 @@ class PluginManager:
try:
success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
if success:
with self._plugin_last_update_lock:
self.plugin_last_update[plugin_id] = time.time()
self.plugin_last_update[plugin_id] = time.time()
self.state_manager.record_update(plugin_id)
self.state_manager.set_state(plugin_id, PluginState.ENABLED)
else:
+39 -3
View File
@@ -110,12 +110,17 @@ class SchemaManager:
try:
with open(schema_path, 'r', encoding='utf-8') as f:
schema = json.load(f)
# Validate schema structure (basic check)
if not isinstance(schema, dict):
self.logger.error(f"Invalid schema format for {plugin_id}: not a dictionary")
return None
# Expand x-style-elements declarations BEFORE caching, so every
# consumer (config form GET, save path, validation, defaults
# generation) sees the identical expanded shape.
schema = self._expand_style_elements(schema)
# Cache the schema
self._schema_cache[plugin_id] = schema
@@ -132,10 +137,41 @@ class SchemaManager:
self.logger.error(f"Error loading schema for {plugin_id}: {e}")
return None
# ------------------------------------------------------------------
# x-style-elements expansion
# ------------------------------------------------------------------
# Plugins declare styleable display elements compactly via an
# "x-style-elements" object on their customization schema; load_schema
# expands each declaration into full font/font_size/text_color/offset
# property blocks before caching, so the config form, save path,
# validation, and defaults generation all see the same expanded shape.
# The single implementation lives in src.element_style (pure, also used
# by plugins reading their own schema file) — see that module for the
# declaration format.
@staticmethod
def get_style_elements(schema: Dict[str, Any]) -> Dict[str, Any]:
"""The x-style-elements declaration from a schema ({} if none)."""
from src.element_style import get_style_elements
return get_style_elements(schema)
def _expand_style_elements(self, schema: Dict[str, Any]) -> Dict[str, Any]:
"""Expand x-style-elements declarations (no-op without any).
Never raises; on any failure the original schema is returned so a
malformed declaration can't take a plugin down.
"""
try:
from src.element_style import expand_style_elements
return expand_style_elements(schema)
except Exception as e:
self.logger.error(f"x-style-elements expansion failed: {e}")
return schema
def invalidate_cache(self, plugin_id: Optional[str] = None) -> None:
"""
Invalidate schema cache for a plugin or all plugins.
Args:
plugin_id: Plugin identifier to invalidate, or None to clear all
"""
+121
View File
@@ -0,0 +1,121 @@
"""Headless single-render service for plugins.
Renders one plugin instance at one panel size to an in-memory PIL image
no hardware, no singletons, no pip (install_deps is always False). Shared
by the dev server's /api/render endpoints and the production web UI's
config-page live preview.
A fresh plugin instance is created per call (mirroring the safety
harness), so repeated renders never share instance state. The plugin's
module does stay imported in the process module-level globals persist
across calls, which is fine for previewing but worth knowing.
"""
import json
import logging
import time
from pathlib import Path
from typing import Any, Dict, Optional
logger = logging.getLogger(__name__)
def render_plugin_once(plugin_id: str, plugin_dir: Path,
manifest: Optional[Dict[str, Any]] = None,
config: Optional[Dict[str, Any]] = None,
mock_data: Optional[Dict[str, Any]] = None,
width: int = 128, height: int = 32,
skip_update: bool = True) -> Dict[str, Any]:
"""Render one plugin at one size. Returns a response-shaped dict:
{'image': 'data:image/png;base64,...', 'width', 'height',
'render_time_ms', 'errors': [...], 'warnings': [...]}
``skip_update`` defaults to True: update() may block on live network
(sports APIs, Spotify) callers that want real data should prime
``mock_data`` (e.g. from the plugin's test/harness.json fixture, see
``load_harness_spec``) or explicitly pass skip_update=False.
Raises on plugin load failure; update()/display() exceptions are
captured into warnings/errors instead so a broken render still shows
whatever was drawn.
"""
from src.plugin_system.plugin_loader import PluginLoader
from src.plugin_system.testing import (
MockCacheManager, MockPluginManager, VisualTestDisplayManager)
plugin_dir = Path(plugin_dir)
if manifest is None:
with open(plugin_dir / 'manifest.json', 'r', encoding='utf-8') as f:
manifest = json.load(f)
config = config or {'enabled': True}
mock_data = mock_data or {}
display_manager = VisualTestDisplayManager(width=width, height=height)
cache_manager = MockCacheManager()
plugin_manager = MockPluginManager()
# Pre-populate cache with mock data
for key, value in mock_data.items():
cache_manager.set(key, value)
loader = PluginLoader()
errors = []
warnings = []
plugin_instance, _module = loader.load_plugin(
plugin_id=plugin_id,
manifest=manifest,
plugin_dir=plugin_dir,
config=config,
display_manager=display_manager,
cache_manager=cache_manager,
plugin_manager=plugin_manager,
install_deps=False,
)
start_time = time.time()
# Exception detail policy (matches the dev server's convention): full
# tracebacks go to the server log; clients get only the exception class
# name — CodeQL flags raw exception text in responses as stack-trace
# exposure, and the log is where developers look anyway.
try:
if not skip_update:
try:
plugin_instance.update()
except Exception as e:
logger.warning("update() raised for plugin %s", plugin_id,
exc_info=True)
warnings.append(f"update() raised: {type(e).__name__} — see server log")
try:
plugin_instance.display(force_clear=True)
except Exception as e:
logger.warning("display() raised for plugin %s", plugin_id,
exc_info=True)
errors.append(f"display() raised: {type(e).__name__} — see server log")
render_time_ms = round((time.time() - start_time) * 1000, 1)
# Capture BEFORE cleanup — a plugin's cleanup may clear the canvas
image_b64 = display_manager.get_image_base64()
finally:
# The instance is throwaway, but its __init__ may have opened
# sessions or started threads (music's clients, sports API
# sessions). In a long-running web process, previews without
# cleanup would leak those per request.
try:
plugin_instance.cleanup()
except Exception as e:
logger.warning("cleanup() raised for plugin %s", plugin_id,
exc_info=True)
warnings.append(f"cleanup() raised: {type(e).__name__} — see server log")
return {
'image': f'data:image/png;base64,{image_b64}',
'width': width,
'height': height,
'render_time_ms': render_time_ms,
'errors': errors,
'warnings': warnings,
}
-8
View File
@@ -297,8 +297,6 @@ class RenderPipeline:
Returns True when:
- Cycle is complete and we should start fresh
- Staging buffer has new content
- A plugin currently visible in the scroll has pending updated data
(e.g. a live score changed) standalone (non-sync) mode only
"""
if self._cycle_complete:
return True
@@ -316,12 +314,6 @@ class RenderPipeline:
if buffer_status['staging_count'] > 0:
return True
# Trigger recompose when pending updates affect visible segments, so
# live score/status changes reach the display within a few seconds
# instead of waiting for the next full cycle.
if self.stream_manager.has_pending_updates_for_visible_segments():
return True
return False
def hot_swap_content(self) -> bool:
@@ -1,88 +0,0 @@
"""
Regression tests for DisplayController._tick_plugin_updates_for_vegas().
PR #299 added logic to detect which plugins actually got fresh data on a
scheduled-update tick and notify Vegas mode via
vegas_coordinator.mark_plugin_updated(), so a live score change reaches the
scroll within seconds instead of waiting for a full cycle. PR #330's
multi-display sync refactor deleted this method (folding the callback back
to the plain _tick_plugin_updates(), which reports nothing), silently
orphaning VegasModeCoordinator.mark_plugin_updated() -- it has had zero
callers since.
"""
from typing import Dict, List, Optional
from unittest.mock import MagicMock
from src.display_controller import DisplayController
def _make_controller(updated: Optional[List[str]] = None, vegas_coordinator: Optional[MagicMock] = None) -> DisplayController:
dc = object.__new__(DisplayController)
dc.plugin_manager = MagicMock()
dc.plugin_manager.run_scheduled_updates_with_changes.return_value = list(updated or [])
dc.vegas_coordinator = vegas_coordinator
return dc
class TestTickPluginUpdatesForVegas:
def test_marks_only_plugins_whose_timestamp_advanced(self):
vc = MagicMock()
dc = _make_controller(updated=["stock-news"], vegas_coordinator=vc)
dc._tick_plugin_updates_for_vegas()
vc.mark_plugin_updated.assert_called_once_with("stock-news")
def test_no_advance_marks_nothing(self):
vc = MagicMock()
dc = _make_controller(updated=[], vegas_coordinator=vc)
dc._tick_plugin_updates_for_vegas()
vc.mark_plugin_updated.assert_not_called()
def test_no_vegas_coordinator_does_not_raise(self):
dc = _make_controller(updated=["stock-news"], vegas_coordinator=None)
dc._tick_plugin_updates_for_vegas() # must not raise
def test_mark_plugin_updated_exception_does_not_propagate(self):
"""One plugin's mark_plugin_updated failing must not stop the tick
or crash the update loop it runs in."""
vc = MagicMock()
vc.mark_plugin_updated.side_effect = [RuntimeError("boom"), None]
dc = _make_controller(updated=["a", "b"], vegas_coordinator=vc)
dc._tick_plugin_updates_for_vegas() # must not raise
assert vc.mark_plugin_updated.call_count == 2
class TestVegasCoordinatorCallbackWiring:
def test_initialize_wires_vegas_aware_tick_as_update_callback(self):
"""The Vegas coordinator must be given the Vegas-aware
_tick_plugin_updates_for_vegas as its update callback, not the plain
_tick_plugin_updates() -- that's the exact wiring PR #330 dropped."""
dc = object.__new__(DisplayController)
dc.config = {"display": {"vegas_scroll": {"enabled": True}}, "sync": {}}
dc.display_manager = MagicMock()
dc.plugin_manager = MagicMock()
dc.sync_manager = MagicMock()
dc._check_live_priority = MagicMock()
dc._check_vegas_interrupt = MagicMock(return_value=False)
fake_coordinator = MagicMock()
import src.display_controller as dc_module
original_imported = dc_module._vegas_mode_imported
original_class = dc_module.VegasModeCoordinator
try:
dc_module._vegas_mode_imported = True
dc_module.VegasModeCoordinator = MagicMock(return_value=fake_coordinator)
dc._initialize_vegas_mode()
finally:
dc_module._vegas_mode_imported = original_imported
dc_module.VegasModeCoordinator = original_class
fake_coordinator.set_update_callback.assert_called_once_with(dc._tick_plugin_updates_for_vegas)
+519
View File
@@ -0,0 +1,519 @@
"""Tests for src/element_style.py — universal per-element style resolution.
The load-bearing behavior is the user-override check: saved configs ALWAYS
contain the schema defaults (merge_with_defaults runs at save time and again
before plugin instantiation), so "key present" must never be read as "user
set it". Only "present and different from the schema default" counts.
"""
import json
import os
import sys
import pytest
from PIL import ImageFont
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from src.element_style import ( # noqa: E402
ElementStyle,
ElementStyleResolver,
FONT_ALIASES,
defaults_from_schema_file,
extract_schema_defaults,
load_font,
resolve_font_name,
)
PRESS_START = "PressStart2P-Regular.ttf"
FOUR_BY_SIX = "4x6-font.ttf"
FIVE_BY_SEVEN_BDF = "5x7.bdf"
# ---------------------------------------------------------------------------
# load_font
# ---------------------------------------------------------------------------
class TestLoadFont:
def test_ttf(self):
font = load_font(PRESS_START, 8)
assert isinstance(font, ImageFont.FreeTypeFont)
assert font.size == 8
def test_bdf_at_native_size(self):
"""FreeType loads BDF strikes directly at their native size."""
font = load_font(FIVE_BY_SEVEN_BDF, 7)
assert isinstance(font, ImageFont.FreeTypeFont)
def test_bdf_at_wrong_size_falls_back(self):
"""BDF fonts are fixed-size; a non-native size falls back to the
fallback font at the requested size rather than raising."""
font = load_font(FIVE_BY_SEVEN_BDF, 14)
assert isinstance(font, ImageFont.FreeTypeFont)
assert font.size == 14 # fallback font honored the requested size
def test_alias_resolves(self):
assert resolve_font_name("press_start") == PRESS_START
font = load_font("press_start", 16)
assert isinstance(font, ImageFont.FreeTypeFont)
assert font.size == 16
def test_filename_passes_through_alias(self):
assert resolve_font_name(FOUR_BY_SIX) == FOUR_BY_SIX
def test_missing_file_falls_back(self):
font = load_font("no-such-font.ttf", 10)
assert isinstance(font, ImageFont.FreeTypeFont)
assert font.size == 10
@pytest.mark.parametrize("garbage", ["", None, "../../etc/passwd", "x" * 300])
def test_garbage_never_raises(self, garbage):
font = load_font(garbage, 8)
assert font is not None
def test_everything_missing_uses_pil_default(self):
font = load_font("nope.ttf", 8, fonts_dir="/nonexistent",
fallback_font="also-nope.ttf")
assert font is not None # ImageFont.load_default()
def test_aliases_cover_the_baseball_set(self):
"""The centralized map must be a superset of the per-plugin copies
it replaces (baseball game_renderer.py + sports.py)."""
assert FONT_ALIASES["press_start"] == PRESS_START
assert FONT_ALIASES["four_by_six"] == FOUR_BY_SIX
assert FONT_ALIASES["five_by_seven"] == FIVE_BY_SEVEN_BDF
# ---------------------------------------------------------------------------
# user_forced provenance
# ---------------------------------------------------------------------------
SCHEMA_DEFAULTS = {
"customization": {
"score_text": {"font": PRESS_START, "font_size": 10},
}
}
def _style(config, defaults=SCHEMA_DEFAULTS):
return ElementStyleResolver(config, defaults).style(
"score_text", classic_font=PRESS_START, classic_size=10)
class TestUserForced:
def test_absent_is_not_forced(self):
style = _style({})
assert not style.user_forced
assert style.font_name == PRESS_START
assert style.font_size == 10
def test_schema_default_present_is_not_forced(self):
"""THE bug this module exists to fix: the save flow writes schema
defaults into every saved config, so their presence means nothing."""
config = {"customization": {"score_text": {
"font": PRESS_START, "font_size": 10}}}
style = _style(config)
assert not style.user_forced
def test_different_font_is_forced(self):
config = {"customization": {"score_text": {
"font": FOUR_BY_SIX, "font_size": 10}}}
style = _style(config)
assert style.user_forced_font
assert not style.user_forced_size
assert style.user_forced
def test_different_size_is_forced(self):
config = {"customization": {"score_text": {
"font": PRESS_START, "font_size": 14}}}
style = _style(config)
assert style.user_forced_size
assert not style.user_forced_font
assert style.font_size == 14
def test_string_size_equal_to_default_is_not_forced(self):
config = {"customization": {"score_text": {"font_size": "10"}}}
assert not _style(config).user_forced
def test_without_schema_defaults_compares_against_classic(self):
"""Degraded mode (old cores, tests): classic_* is the reference."""
config = {"customization": {"score_text": {
"font": PRESS_START, "font_size": 10}}}
style = _style(config, defaults={})
assert not style.user_forced
forced = _style({"customization": {"score_text": {"font_size": 12}}},
defaults={})
assert forced.user_forced
def test_schema_default_differing_from_classic_wins_as_reference(self):
"""When the schema declares a different default than the classic_*
args, the schema is the reference a config equal to the schema
default is untouched."""
defaults = {"customization": {"score_text": {
"font": FOUR_BY_SIX, "font_size": 6}}}
config = {"customization": {"score_text": {
"font": FOUR_BY_SIX, "font_size": 6}}}
style = ElementStyleResolver(config, defaults).style(
"score_text", classic_font=PRESS_START, classic_size=10)
assert not style.user_forced
def test_unknown_element_uses_classic(self):
style = ElementStyleResolver({}, SCHEMA_DEFAULTS).style(
"no_such_element", classic_font=FOUR_BY_SIX, classic_size=6)
assert not style.user_forced
assert style.font_name == FOUR_BY_SIX
assert style.font_size == 6
def test_malformed_customization_is_tolerated(self):
for bad in [{"customization": "oops"},
{"customization": {"score_text": "oops"}},
{"customization": {"score_text": {"font_size": "huge"}}},
None]:
style = _style(bad)
assert not style.user_forced
assert style.font_size == 10
# ---------------------------------------------------------------------------
# color
# ---------------------------------------------------------------------------
class TestColor:
"""Color provenance mirrors fonts: the web form ALWAYS posts the RGB
inputs, so a saved config carries the schema-default color whether or
not the user touched it. Only a value differing from the schema default
is an override; otherwise the plugin's classic color survives — critical
for state-dependent colors (a score that turns gold on a touchdown)."""
COLOR_DEFAULTS = {"customization": {"score_text": {
"font": PRESS_START, "font_size": 10, "text_color": [255, 255, 255]}}}
def _color_style(self, config, defaults=None, classic_color=(255, 215, 0)):
return ElementStyleResolver(config, defaults or self.COLOR_DEFAULTS).style(
"score_text", classic_font=PRESS_START, classic_size=10,
classic_color=classic_color)
def test_absent_returns_classic_color(self):
style = self._color_style({})
assert style.color == (255, 215, 0)
assert not style.user_forced_color
def test_absent_with_no_classic_is_none(self):
assert _style({}).color is None
def test_schema_default_present_keeps_classic_color(self):
"""A saved config always contains the default — it must not clobber
the plugin's (possibly semantic) classic color."""
config = {"customization": {"score_text": {"text_color": [255, 255, 255]}}}
style = self._color_style(config)
assert style.color == (255, 215, 0)
assert not style.user_forced_color
def test_changed_color_is_an_override(self):
config = {"customization": {"score_text": {"text_color": [0, 128, 255]}}}
style = self._color_style(config)
assert style.color == (0, 128, 255)
assert style.user_forced_color
def test_present_without_schema_default_is_an_override(self):
"""Hand-written schemas without a text_color default: presence is
intent (there is nothing to compare against)."""
config = {"customization": {"score_text": {"text_color": [0, 128, 255]}}}
style = self._color_style(config, defaults=SCHEMA_DEFAULTS)
assert style.color == (0, 128, 255)
assert style.user_forced_color
def test_values_clamped(self):
config = {"customization": {"score_text": {"text_color": [300, -5, 128]}}}
assert self._color_style(config).color == (255, 0, 128)
def test_color_never_affects_user_forced_sizing(self):
config = {"customization": {"score_text": {"text_color": [0, 128, 255]}}}
style = self._color_style(config)
assert style.user_forced_color
assert not style.user_forced
@pytest.mark.parametrize("bad", [[1, 2], [1, 2, 3, 4], "red",
["a", "b", "c"], 255, None])
def test_malformed_falls_back_to_classic(self, bad):
config = {"customization": {"score_text": {"text_color": bad}}}
style = self._color_style(config, classic_color=(1, 2, 3))
assert style.color == (1, 2, 3)
assert not style.user_forced_color
# ---------------------------------------------------------------------------
# offsets
# ---------------------------------------------------------------------------
class TestOffsets:
def test_unset_is_zero(self):
assert ElementStyleResolver({}).offset("score") == (0, 0)
def test_layout_section(self):
"""The deployed sports convention: customization.layout.<element>."""
config = {"customization": {"layout": {"score": {
"x_offset": 3, "y_offset": -2}}}}
assert ElementStyleResolver(config).offset("score") == (3, -2)
def test_element_section_fallback(self):
config = {"customization": {"score": {"x_offset": 5}}}
assert ElementStyleResolver(config).offset("score") == (5, 0)
def test_layout_section_wins_over_element_section(self):
config = {"customization": {
"layout": {"score": {"x_offset": 1}},
"score": {"x_offset": 9, "y_offset": 9},
}}
resolver = ElementStyleResolver(config)
assert resolver.offset_value("score", "x_offset") == 1
# y_offset absent from layout section -> element section supplies it
assert resolver.offset_value("score", "y_offset") == 9
@pytest.mark.parametrize("raw,expected", [
(2, 2), (2.7, 2), ("3", 3), ("2.0", 2), ("-4", -4),
(None, 0), ("junk", 0), ([], 0), (True, 0),
])
def test_coercion_matches_sports_helper(self, raw, expected):
"""Same tolerance as the sports.py/_get_layout_offset copies this
replaces: int/float/numeric-string pass, anything else -> default."""
config = {"customization": {"layout": {"e": {"x_offset": raw}}}}
assert ElementStyleResolver(config).offset_value("e", "x_offset") == expected
def test_custom_axis_names(self):
"""Football's records use away_x_offset/home_x_offset."""
config = {"customization": {"layout": {"records": {
"away_x_offset": 4, "home_x_offset": -4}}}}
resolver = ElementStyleResolver(config)
assert resolver.offset_value("records", "away_x_offset") == 4
assert resolver.offset_value("records", "home_x_offset") == -4
def test_style_carries_offset(self):
config = {"customization": {"layout": {"score_text": {
"x_offset": 2, "y_offset": 1}}}}
assert _style(config).offset == (2, 1)
# ---------------------------------------------------------------------------
# caching
# ---------------------------------------------------------------------------
class TestCaching:
def test_same_call_is_cached(self):
resolver = ElementStyleResolver({}, SCHEMA_DEFAULTS)
a = resolver.style("score_text", classic_font=PRESS_START, classic_size=10)
b = resolver.style("score_text", classic_font=PRESS_START, classic_size=10)
assert a is b
def test_clear_cache(self):
resolver = ElementStyleResolver({}, SCHEMA_DEFAULTS)
a = resolver.style("score_text", classic_font=PRESS_START, classic_size=10)
resolver.clear_cache()
b = resolver.style("score_text", classic_font=PRESS_START, classic_size=10)
assert a is not b
# PIL fonts compare by identity; compare the value fields
assert (a.font_name, a.font_size, a.color, a.offset, a.user_forced) == \
(b.font_name, b.font_size, b.color, b.offset, b.user_forced)
# ---------------------------------------------------------------------------
# schema default extraction
# ---------------------------------------------------------------------------
class TestSchemaDefaults:
def test_matches_schema_manager_extraction(self):
"""The pure helper must agree with SchemaManager.extract_defaults_from_schema
on a real plugin-style schema it exists so plugins get the same
answer in harness contexts where the schema manager is absent."""
from src.plugin_system.schema_manager import SchemaManager
schema = {
"type": "object",
"properties": {
"enabled": {"type": "boolean", "default": True},
"customization": {
"type": "object",
"properties": {
"score_text": {
"type": "object",
"properties": {
"font": {"type": "string", "default": PRESS_START},
"font_size": {"type": "integer", "default": 10},
"y_percent": {"type": "number"},
},
},
"layout": {
"type": "object",
"properties": {
"score": {
"type": "object",
"properties": {
"x_offset": {"type": "integer", "default": 0},
},
},
},
},
},
},
"opaque_with_default": {"type": "object", "default": {},
"properties": {"x": {"default": 1}}},
# array shapes the schema manager special-cases — parity
# must hold for these too (found divergent in review)
"plain_array": {"type": "array", "items": {"type": "string"}},
"obj_array": {"type": "array", "items": {
"type": "object", "properties": {"x": {"default": 1}}}},
"item_default_array": {"type": "array",
"items": {"type": "string", "default": "a"}},
},
}
pure = extract_schema_defaults(schema)
managed = SchemaManager().extract_defaults_from_schema(schema)
assert pure == managed
assert pure["customization"]["score_text"]["font"] == PRESS_START
# object-level default short-circuits recursion (both must agree)
assert pure["opaque_with_default"] == {}
assert pure["plain_array"] == []
assert pure["obj_array"] == []
assert pure["item_default_array"] == ["a"]
def test_defaults_from_schema_file(self, tmp_path):
schema_path = tmp_path / "config_schema.json"
schema_path.write_text(json.dumps({
"type": "object",
"properties": {
"customization": {
"type": "object",
"properties": {
"title_text": {
"type": "object",
"properties": {
"font": {"type": "string", "default": PRESS_START},
},
},
},
},
},
}))
defaults = defaults_from_schema_file(str(schema_path))
assert defaults["customization"]["title_text"]["font"] == PRESS_START
def test_defaults_from_missing_or_bad_file(self, tmp_path):
assert defaults_from_schema_file("/nonexistent/schema.json") == {}
bad = tmp_path / "bad.json"
bad.write_text("{not json")
assert defaults_from_schema_file(str(bad)) == {}
# ---------------------------------------------------------------------------
# BasePlugin integration
# ---------------------------------------------------------------------------
class _StubSchemaManager:
"""Schema manager double exposing the two methods the resolver path uses."""
def __init__(self, schema):
self._schema = schema
def load_schema(self, plugin_id, use_cache=True):
return self._schema
def extract_defaults_from_schema(self, schema, prefix=""):
# Mirror the real nested-dict extraction for this simple shape
def walk(props):
out = {}
for key, spec in props.get("properties", {}).items():
if "default" in spec:
out[key] = spec["default"]
elif spec.get("type") == "object" and "properties" in spec:
nested = walk(spec)
if nested:
out[key] = nested
return out
return walk(schema)
def _make_plugin(config, schema=None):
from src.plugin_system.base_plugin import BasePlugin
from src.plugin_system.testing.mocks import (
MockCacheManager, MockPluginManager)
from src.plugin_system.testing.visual_display_manager import (
VisualTestDisplayManager)
class _Plugin(BasePlugin):
def update(self):
return True
def display(self, force_clear=False):
return None
plugin_manager = MockPluginManager()
if schema is not None:
plugin_manager.schema_manager = _StubSchemaManager(schema)
return _Plugin("test-plugin", config,
VisualTestDisplayManager(64, 32),
MockCacheManager(), plugin_manager)
TEST_SCHEMA = {
"type": "object",
"properties": {
"customization": {
"type": "object",
"properties": {
"score_text": {
"type": "object",
"properties": {
"font": {"type": "string", "default": PRESS_START},
"font_size": {"type": "integer", "default": 10},
},
},
},
},
},
}
class TestBasePluginIntegration:
def test_element_style_with_schema_defaults(self):
"""The full path: saved config carries schema defaults, plugin's
element_style still reports not-forced."""
config = {"enabled": True, "customization": {"score_text": {
"font": PRESS_START, "font_size": 10}}}
plugin = _make_plugin(config, schema=TEST_SCHEMA)
style = plugin.element_style("score_text", classic_font=PRESS_START,
classic_size=10)
assert not style.user_forced
def test_element_style_detects_real_override(self):
config = {"enabled": True, "customization": {"score_text": {
"font": PRESS_START, "font_size": 14}}}
plugin = _make_plugin(config, schema=TEST_SCHEMA)
style = plugin.element_style("score_text", classic_font=PRESS_START,
classic_size=10)
assert style.user_forced_size
assert style.font_size == 14
def test_works_without_schema_manager(self):
"""MockPluginManager has no schema_manager attribute by default —
the resolver degrades to classic-default comparison, no crash."""
config = {"enabled": True, "customization": {"score_text": {
"font_size": 12}}}
plugin = _make_plugin(config, schema=None)
style = plugin.element_style("score_text", classic_font=PRESS_START,
classic_size=10)
assert style.user_forced_size # 12 != classic 10
def test_resolver_is_cached_and_invalidated_on_config_change(self):
plugin = _make_plugin({"enabled": True}, schema=TEST_SCHEMA)
first = plugin.style_resolver
assert plugin.style_resolver is first
plugin.on_config_change({"enabled": True, "customization": {
"score_text": {"font_size": 14}}})
second = plugin.style_resolver
assert second is not first
style = plugin.element_style("score_text", classic_font=PRESS_START,
classic_size=10)
assert style.user_forced_size
if __name__ == "__main__":
sys.exit(pytest.main([__file__, "-v"]))
+273
View File
@@ -0,0 +1,273 @@
"""Tests for x-style-elements schema expansion.
A plugin declares styleable elements once, compactly; expansion generates
the full customization property blocks at schema-load time. The invariants
that matter:
- idempotent (expand(expand(s)) == expand(s)) and the input is never mutated
- both load paths (cached GET, uncached save) see the identical shape
- generated defaults flow into generate_default_config, and saving twice is
round-trip stable (merge_with_defaults produces no churn)
- hand-written property blocks for the same element always win
- defaults_from_schema_file (what plugins use to build resolvers from their
RAW schema file) agrees exactly with the schema manager's expanded view
"""
import copy
import json
import os
import sys
import jsonschema
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from src.element_style import ( # noqa: E402
defaults_from_schema_file,
expand_style_elements,
extract_schema_defaults,
get_style_elements,
)
from src.plugin_system.schema_manager import SchemaManager # noqa: E402
PRESS_START = "PressStart2P-Regular.ttf"
def _declared_schema():
return {
"$schema": "http://json-schema.org/draft-07/schema#",
"type": "object",
"properties": {
"enabled": {"type": "boolean", "default": True},
"customization": {
"type": "object",
"title": "Display Customization",
"x-style-elements": {
"score_text": {
"title": "Game Score",
"font": {"default": PRESS_START},
"size": {"default": 10, "min": 4, "max": 16},
"color": True,
"offsets": True,
},
"detail_text": {
"font": {"default": "4x6-font.ttf"},
"size": {"default": 6},
},
},
"properties": {},
"additionalProperties": False,
},
},
}
class TestExpansionShape:
def test_generates_element_blocks(self):
expanded = expand_style_elements(_declared_schema())
cust = expanded["properties"]["customization"]["properties"]
score = cust["score_text"]
assert score["x-style-managed"] is True
assert score["title"] == "Game Score"
assert score["properties"]["font"]["default"] == PRESS_START
assert score["properties"]["font"]["x-widget"] == "font-selector"
assert score["properties"]["font_size"]["default"] == 10
assert score["properties"]["font_size"]["minimum"] == 4
assert score["properties"]["font_size"]["maximum"] == 16
assert score["properties"]["text_color"]["x-widget"] == "color-picker"
assert score["properties"]["text_color"]["default"] == [255, 255, 255]
assert score["additionalProperties"] is False
def test_color_and_offsets_are_optional(self):
expanded = expand_style_elements(_declared_schema())
cust = expanded["properties"]["customization"]["properties"]
detail = cust["detail_text"]
assert "text_color" not in detail["properties"]
assert detail["title"] == "Detail Text" # prettified from the key
layout = cust["layout"]["properties"]
assert "score_text" in layout
assert "detail_text" not in layout
def test_offsets_block_shape(self):
expanded = expand_style_elements(_declared_schema())
layout = expanded["properties"]["customization"]["properties"]["layout"]
assert layout["x-style-managed"] is True
entry = layout["properties"]["score_text"]
assert entry["properties"]["x_offset"]["default"] == 0
assert entry["properties"]["y_offset"]["default"] == 0
def test_declared_color_default(self):
schema = _declared_schema()
decl = schema["properties"]["customization"]["x-style-elements"]
decl["score_text"]["color"] = {"default": [255, 215, 0]}
expanded = expand_style_elements(schema)
color = (expanded["properties"]["customization"]["properties"]
["score_text"]["properties"]["text_color"])
assert color["default"] == [255, 215, 0]
def test_declaration_survives_expansion(self):
"""The declaration is the element registry for tooling — it must
remain readable from the expanded schema."""
expanded = expand_style_elements(_declared_schema())
assert set(get_style_elements(expanded)) == {"score_text", "detail_text"}
assert set(SchemaManager.get_style_elements(expanded)) == {
"score_text", "detail_text"}
def test_no_declaration_returns_same_object(self):
schema = {"type": "object", "properties": {"enabled": {"default": True}}}
assert expand_style_elements(schema) is schema
def test_valid_draft7(self):
jsonschema.Draft7Validator.check_schema(
expand_style_elements(_declared_schema()))
def test_property_order_updated_when_present(self):
schema = _declared_schema()
schema["properties"]["customization"]["x-propertyOrder"] = []
expanded = expand_style_elements(schema)
order = expanded["properties"]["customization"]["x-propertyOrder"]
# generated elements before layout (the template only renders keys
# in x-propertyOrder when one exists)
assert set(order) == {"score_text", "detail_text", "layout"}
assert order.index("score_text") < order.index("layout")
def test_malformed_declaration_is_harmless(self):
schema = _declared_schema()
schema["properties"]["customization"]["x-style-elements"] = {
"bad": "not a dict", "score_text": {"size": {"default": 10}}}
expanded = expand_style_elements(schema)
cust = expanded["properties"]["customization"]["properties"]
assert "bad" not in cust
assert "score_text" in cust
class TestExpansionInvariants:
def test_idempotent(self):
once = expand_style_elements(_declared_schema())
twice = expand_style_elements(once)
assert once == twice
def test_input_never_mutated(self):
schema = _declared_schema()
snapshot = copy.deepcopy(schema)
expand_style_elements(schema)
assert schema == snapshot
def test_hand_written_block_wins(self):
schema = _declared_schema()
hand_written = {
"type": "object",
"properties": {"font": {"type": "string", "default": "custom.ttf"}},
}
schema["properties"]["customization"]["properties"]["score_text"] = \
copy.deepcopy(hand_written)
expanded = expand_style_elements(schema)
assert (expanded["properties"]["customization"]["properties"]["score_text"]
== hand_written)
def test_hand_written_layout_entry_wins(self):
schema = _declared_schema()
schema["properties"]["customization"]["properties"]["layout"] = {
"type": "object",
"properties": {"score_text": {"type": "object", "properties": {
"x_offset": {"type": "integer", "default": 5}}}},
}
expanded = expand_style_elements(schema)
layout = expanded["properties"]["customization"]["properties"]["layout"]
assert layout["properties"]["score_text"]["properties"]["x_offset"]["default"] == 5
class TestSchemaManagerIntegration:
def _manager_with_schema(self, tmp_path, schema):
plugin_dir = tmp_path / "test-plugin"
plugin_dir.mkdir()
(plugin_dir / "config_schema.json").write_text(json.dumps(schema))
return SchemaManager(plugins_dir=tmp_path)
def test_load_schema_expands(self, tmp_path):
mgr = self._manager_with_schema(tmp_path, _declared_schema())
loaded = mgr.load_schema("test-plugin")
assert "score_text" in loaded["properties"]["customization"]["properties"]
def test_cached_and_uncached_loads_agree(self, tmp_path):
"""The save path uses use_cache=False while the form GET uses the
cache they must see the identical expanded shape."""
mgr = self._manager_with_schema(tmp_path, _declared_schema())
cached = mgr.load_schema("test-plugin", use_cache=True)
again = mgr.load_schema("test-plugin", use_cache=True)
uncached = mgr.load_schema("test-plugin", use_cache=False)
assert cached == uncached == again
def test_disk_file_untouched(self, tmp_path):
schema = _declared_schema()
mgr = self._manager_with_schema(tmp_path, schema)
mgr.load_schema("test-plugin")
on_disk = json.loads(
(tmp_path / "test-plugin" / "config_schema.json").read_text())
assert on_disk == schema
assert "score_text" not in on_disk["properties"]["customization"]["properties"]
def test_defaults_include_generated_elements(self, tmp_path):
mgr = self._manager_with_schema(tmp_path, _declared_schema())
defaults = mgr.generate_default_config("test-plugin")
assert defaults["customization"]["score_text"]["font"] == PRESS_START
assert defaults["customization"]["score_text"]["font_size"] == 10
assert defaults["customization"]["score_text"]["text_color"] == [255, 255, 255]
assert defaults["customization"]["layout"]["score_text"]["x_offset"] == 0
def test_save_twice_is_round_trip_stable(self, tmp_path):
"""merge_with_defaults(merged, defaults) must be a fixed point —
saving a config twice can't keep growing/altering it."""
mgr = self._manager_with_schema(tmp_path, _declared_schema())
defaults = mgr.generate_default_config("test-plugin")
user_config = {"enabled": True, "customization": {
"score_text": {"font_size": 14}}}
merged_once = mgr.merge_with_defaults(user_config, defaults)
merged_twice = mgr.merge_with_defaults(merged_once, defaults)
assert merged_once == merged_twice
assert merged_once["customization"]["score_text"]["font_size"] == 14
class TestResolverParity:
def test_defaults_from_schema_file_matches_manager_view(self, tmp_path):
"""Plugins build resolvers from their RAW schema file; the web UI
merges defaults from the EXPANDED schema. Both must produce the
same defaults or override detection diverges between contexts."""
schema = _declared_schema()
plugin_dir = tmp_path / "test-plugin"
plugin_dir.mkdir()
schema_path = plugin_dir / "config_schema.json"
schema_path.write_text(json.dumps(schema))
mgr = SchemaManager(plugins_dir=tmp_path)
manager_defaults = mgr.extract_defaults_from_schema(
mgr.load_schema("test-plugin"))
raw_file_defaults = defaults_from_schema_file(str(schema_path))
assert raw_file_defaults == manager_defaults
def test_resolver_treats_generated_defaults_as_untouched(self, tmp_path):
"""End to end: a config saved through the web UI (all generated
defaults baked in) must not read as a user override, and the
schema-default color must not clobber a classic color."""
from src.element_style import ElementStyleResolver
schema_path = tmp_path / "config_schema.json"
schema_path.write_text(json.dumps(_declared_schema()))
defaults = defaults_from_schema_file(str(schema_path))
saved_config = {"enabled": True, "customization": {
"score_text": {"font": PRESS_START, "font_size": 10,
"text_color": [255, 255, 255]},
"layout": {"score_text": {"x_offset": 0, "y_offset": 0}},
}}
resolver = ElementStyleResolver(saved_config, defaults)
style = resolver.style("score_text", classic_font=PRESS_START,
classic_size=10, classic_color=(255, 215, 0))
assert not style.user_forced
assert not style.user_forced_color
assert style.color == (255, 215, 0) # semantic classic color survives
assert style.offset == (0, 0)
if __name__ == "__main__":
sys.exit(pytest.main([__file__, "-v"]))
@@ -1,72 +0,0 @@
"""
Regression tests for RenderPipeline.should_recompose()'s pending-updates check.
PR #299 added a check so a plugin's live score/status change (a "pending
update" in StreamManager) triggers a hot-swap within a few seconds instead
of waiting for a full scroll cycle to complete. PR #330 (multi-display sync)
refactored should_recompose() and dropped that check entirely -- not just
gated behind the new sync-mode deferral it added, but removed outright, so
even standalone (non-sync) installations silently lost live-refresh and fell
back to waiting for full cycle boundaries (which, depending on
min/max_cycle_duration, can be minutes).
"""
from unittest.mock import MagicMock
from src.vegas_mode.config import VegasModeConfig
from src.vegas_mode.render_pipeline import RenderPipeline
class FakeDisplayManager:
width = 64
height = 32
def _make_pipeline(sync_manager=None):
stream_manager = MagicMock()
stream_manager.get_buffer_status.return_value = {'staging_count': 0}
pipeline = RenderPipeline(VegasModeConfig(), FakeDisplayManager(), stream_manager)
pipeline.sync_manager = sync_manager
return pipeline, stream_manager
class TestShouldRecompose:
def test_cycle_complete_always_recomposes(self):
pipeline, stream_manager = _make_pipeline()
pipeline._cycle_complete = True
stream_manager.has_pending_updates_for_visible_segments.return_value = False
assert pipeline.should_recompose() is True
def test_no_pending_updates_no_staging_does_not_recompose(self):
pipeline, stream_manager = _make_pipeline()
stream_manager.has_pending_updates_for_visible_segments.return_value = False
assert pipeline.should_recompose() is False
def test_pending_updates_on_visible_segment_triggers_recompose(self):
"""The actual regression: a live-updated plugin currently in view
must trigger a recompose instead of waiting for cycle end."""
pipeline, stream_manager = _make_pipeline()
stream_manager.has_pending_updates_for_visible_segments.return_value = True
assert pipeline.should_recompose() is True
def test_staging_buffer_content_triggers_recompose(self):
pipeline, stream_manager = _make_pipeline()
stream_manager.get_buffer_status.return_value = {'staging_count': 1}
stream_manager.has_pending_updates_for_visible_segments.return_value = False
assert pipeline.should_recompose() is True
def test_sync_active_defers_pending_updates_to_cycle_boundary(self):
"""Sync-mode deferral (PR #330's actual intent) must still hold:
pending updates alone must NOT trigger a mid-cycle hot-swap when a
follower display is attached, since that causes a visible
freeze+jump on the follower. This must keep working after
restoring the non-sync pending-updates check above."""
pipeline, stream_manager = _make_pipeline(sync_manager=MagicMock())
stream_manager.has_pending_updates_for_visible_segments.return_value = True
assert pipeline.should_recompose() is False
def test_sync_active_still_recomposes_on_cycle_complete(self):
pipeline, stream_manager = _make_pipeline(sync_manager=MagicMock())
pipeline._cycle_complete = True
stream_manager.has_pending_updates_for_visible_segments.return_value = True
assert pipeline.should_recompose() is True
+252
View File
@@ -0,0 +1,252 @@
"""Tests for POST /api/v3/plugins/preview — the config-page live preview.
The endpoint renders a plugin headlessly (pure PIL, no hardware, no pip)
with a CANDIDATE config: either the current form state (parsed by the same
parse_plugin_config_form used by save, so preview and save can never
disagree) or a JSON config body.
"""
import base64
import io
import json
import sys
from pathlib import Path
from unittest.mock import MagicMock
import pytest
from flask import Flask
from PIL import Image
project_root = Path(__file__).parent.parent.parent
sys.path.insert(0, str(project_root))
from web_interface.blueprints import api_v3 as api_v3_module # noqa: E402
from web_interface.blueprints.api_v3 import api_v3 # noqa: E402
PLUGIN_ID = "preview-test-plugin"
MANAGER_PY = '''
from PIL import ImageFont
from src.plugin_system.base_plugin import BasePlugin
class PreviewTestPlugin(BasePlugin):
def update(self):
return True
def display(self, force_clear=False):
if force_clear:
self.display_manager.clear()
text = self.config.get("message", "hello")
self.display_manager.draw.text((1, 1), text, fill=(255, 255, 255))
self.display_manager.update_display()
'''
MANIFEST = {
"id": PLUGIN_ID,
"name": "Preview Test Plugin",
"version": "1.0.0",
"class_name": "PreviewTestPlugin",
"entry_point": "manager.py",
"display_modes": ["preview_test"],
}
SCHEMA = {
"type": "object",
"properties": {
"enabled": {"type": "boolean", "default": True},
"message": {"type": "string", "default": "hello"},
},
}
@pytest.fixture
def plugin_dir(tmp_path):
plugin = tmp_path / PLUGIN_ID
plugin.mkdir()
(plugin / "manager.py").write_text(MANAGER_PY)
(plugin / "manifest.json").write_text(json.dumps(MANIFEST))
(plugin / "config_schema.json").write_text(json.dumps(SCHEMA))
return plugin
@pytest.fixture
def client(plugin_dir, tmp_path):
from src.plugin_system.schema_manager import SchemaManager
test_app = Flask(__name__)
test_app.register_blueprint(api_v3, url_prefix="/api/v3")
config_manager = MagicMock()
config_manager.load_config.return_value = {
"display": {"hardware": {"cols": 64, "chain_length": 2,
"rows": 32, "parallel": 1}},
PLUGIN_ID: {"enabled": False, "message": "saved"},
}
plugin_manager = MagicMock()
plugin_manager.plugins_dir = str(tmp_path)
old = (getattr(api_v3_module.api_v3, "config_manager", None),
getattr(api_v3_module.api_v3, "plugin_manager", None),
getattr(api_v3_module.api_v3, "schema_manager", None))
api_v3_module.api_v3.config_manager = config_manager
api_v3_module.api_v3.plugin_manager = plugin_manager
api_v3_module.api_v3.schema_manager = SchemaManager(plugins_dir=tmp_path)
with test_app.test_client() as c:
yield c
(api_v3_module.api_v3.config_manager,
api_v3_module.api_v3.plugin_manager,
api_v3_module.api_v3.schema_manager) = old
def _decode_image(data_url):
assert data_url.startswith("data:image/png;base64,")
raw = base64.b64decode(data_url.split(",", 1)[1])
return Image.open(io.BytesIO(raw))
class TestPreviewEndpoint:
def test_json_body_renders_at_default_panel_size(self, client):
"""No width/height -> the user's real panel (64*2 x 32*1)."""
resp = client.post(f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}",
json={"config": {"message": "hi"}})
assert resp.status_code == 200
data = resp.get_json()["data"]
img = _decode_image(data["image"])
assert img.size == (128, 32)
assert data["errors"] == []
def test_explicit_size(self, client):
resp = client.post(
f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}&width=64&height=64",
json={"config": {}})
assert resp.status_code == 200
img = _decode_image(resp.get_json()["data"]["image"])
assert img.size == (64, 64)
def test_form_encoding_matches_json(self, client):
"""The form path (what HTMX posts) and the JSON path must render
the same candidate config identically."""
via_json = client.post(
f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}&width=128&height=32",
json={"config": {"message": "same"}})
via_form = client.post(
f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}&width=128&height=32",
data={"message": "same"})
a = _decode_image(via_json.get_json()["data"]["image"])
b = _decode_image(via_form.get_json()["data"]["image"])
assert list(a.getdata()) == list(b.getdata())
def test_candidate_config_wins_over_saved(self, client):
"""The preview must show the UNSAVED form state, not the saved
config ('saved' vs 'candidate' render differently)."""
saved = client.post(
f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}&width=128&height=32",
json={"config": {}})
candidate = client.post(
f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}&width=128&height=32",
json={"config": {"message": "candidate"}})
a = _decode_image(saved.get_json()["data"]["image"])
b = _decode_image(candidate.get_json()["data"]["image"])
assert list(a.getdata()) != list(b.getdata())
def test_disabled_plugin_still_previews(self, client):
"""Saved config has enabled: False — preview forces enabled."""
resp = client.post(f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}",
json={"config": {}})
assert resp.status_code == 200
assert resp.get_json()["data"]["errors"] == []
def test_preview_size_form_field(self, client):
"""The UI size selector posts __preview_size=WxH via hx-vals (htmx
caches hx-post's path, so it can't ride the query string). It must
set the render size and must NOT leak into the candidate config."""
resp = client.post(f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}",
data={"message": "hi", "__preview_size": "64x64"})
assert resp.status_code == 200
data = resp.get_json()["data"]
img = _decode_image(data["image"])
assert img.size == (64, 64)
assert data["errors"] == []
def test_query_args_beat_preview_size_field(self, client):
resp = client.post(
f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}&width=128&height=32",
data={"message": "hi", "__preview_size": "64x64"})
img = _decode_image(resp.get_json()["data"]["image"])
assert img.size == (128, 32)
def test_malformed_preview_size_falls_back_to_panel(self, client):
resp = client.post(f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}",
data={"message": "hi", "__preview_size": "bogus x"})
img = _decode_image(resp.get_json()["data"]["image"])
assert img.size == (128, 32) # cols*chain x rows*parallel
def test_json_candidate_deep_merges_onto_saved_config(self, client):
"""A partial JSON candidate must not wipe saved sibling values in
the same nested section (form path and save both deep-merge)."""
# Saved config has message "saved"; posting an unrelated nested key
# must not discard it — render must still differ from a candidate
# that explicitly changes message.
keep_saved = client.post(
f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}&width=128&height=32",
json={"config": {}})
explicit = client.post(
f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}&width=128&height=32",
json={"config": {"message": "saved"}})
a = _decode_image(keep_saved.get_json()["data"]["image"])
b = _decode_image(explicit.get_json()["data"]["image"])
assert list(a.getdata()) == list(b.getdata())
def test_hanging_plugin_times_out(self, client, plugin_dir, monkeypatch):
"""A plugin whose display() hangs must not pin the web worker.
Uses its own plugin id: the loader caches the module per id, so
reusing PLUGIN_ID would run the already-imported (non-hanging) code
when this test follows others in the suite.
"""
from web_interface.blueprints import api_v3 as api_v3_module
monkeypatch.setattr(api_v3_module, "PREVIEW_RENDER_TIMEOUT_SEC", 1)
hang_id = "preview-hang-plugin"
hang_dir = plugin_dir.parent / hang_id
hang_dir.mkdir()
(hang_dir / "manager.py").write_text(MANAGER_PY.replace(
"self.display_manager.update_display()",
"import time; time.sleep(10); self.display_manager.update_display()"))
manifest = dict(MANIFEST, id=hang_id, name="Hang Plugin")
(hang_dir / "manifest.json").write_text(json.dumps(manifest))
(hang_dir / "config_schema.json").write_text(json.dumps(SCHEMA))
resp = client.post(f"/api/v3/plugins/preview?plugin_id={hang_id}",
json={"config": {}})
assert resp.status_code == 504
def test_htmx_gets_html_fragment(self, client):
resp = client.post(
f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}&width=64&height=32",
data={"message": "hi"}, headers={"HX-Request": "true"})
assert resp.status_code == 200
assert resp.mimetype == "text/html"
body = resp.get_data(as_text=True)
assert "<img" in body and "data:image/png;base64," in body
def test_unknown_plugin_404(self, client):
resp = client.post("/api/v3/plugins/preview?plugin_id=nope",
json={"config": {}})
assert resp.status_code == 404
def test_missing_plugin_id_400(self, client):
resp = client.post("/api/v3/plugins/preview", json={"config": {}})
assert resp.status_code == 400
def test_absurd_size_rejected(self, client):
resp = client.post(
f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}&width=99999&height=32",
json={"config": {}})
assert resp.status_code == 400
if __name__ == "__main__":
sys.exit(pytest.main([__file__, "-v"]))
+597 -354
View File
@@ -329,7 +329,6 @@ def save_schedule_config():
}
mode = data.get('mode', 'global')
schedule_config['mode'] = mode
if mode == 'global':
# Simple global schedule
@@ -4283,6 +4282,373 @@ def _filter_config_by_schema(config, schema, prefix=''):
return filtered
def parse_plugin_config_form(form, schema, existing_config):
"""Convert an HTMX config-form submission into a nested plugin config.
form is the werkzeug form MultiDict; existing_config is the saved
config to merge updates onto (mutated and returned). Handles dotted
field names, bracket/indexed array fields (color pickers), schema-
driven type coercion, and unchecked-checkbox fixup.
Shared by save_plugin_config and the plugin preview endpoint so the
two interpretations of the form can never drift apart."""
plugin_config = existing_config
# Convert form data to config dict
# Form fields can use dot notation for nested values (e.g., "transition.type")
form_data = form.to_dict()
# First pass: handle bracket notation array fields (e.g., "field_name[]" from checkbox-group)
# These fields use getlist() to preserve all values, then replace in form_data
# Sentinel empty value ("") allows clearing array to [] when all checkboxes unchecked
bracket_array_fields = {} # Maps base field path to list of values
for key in form.keys():
# Check if key ends with "[]" (bracket notation for array fields)
if key.endswith('[]'):
base_path = key[:-2] # Remove "[]" suffix
values = form.getlist(key)
# Filter out sentinel empty string - if only sentinel present, array should be []
# If sentinel + values present, use the actual values
filtered_values = [v for v in values if v and v.strip()]
# If no non-empty values but key exists, it means all checkboxes unchecked (empty array)
bracket_array_fields[base_path] = filtered_values
# Remove the bracket notation key from form_data if present
if key in form_data:
del form_data[key]
# Process bracket notation fields and set directly in plugin_config
# Use JSON encoding instead of comma-join to handle values containing commas
import json
for base_path, values in bracket_array_fields.items():
# Get schema property to verify it's an array
base_prop = _get_schema_property(schema, base_path)
if base_prop and base_prop.get('type') == 'array':
# Filter out empty values and sentinel empty strings
filtered_values = [v for v in values if v and v.strip()]
# Set directly in plugin_config (values are already strings, no need to parse)
# Empty array (all unchecked) is represented as []
_set_nested_value(plugin_config, base_path, filtered_values)
logger.debug(f"Processed bracket notation array field {base_path}: {values} -> {filtered_values}")
# Remove from form_data to avoid double processing
if base_path in form_data:
del form_data[base_path]
# Second pass: detect and combine array index fields (e.g., "text_color.0", "text_color.1" -> "text_color" as array)
# This handles cases where forms send array fields as indexed inputs
array_fields = {} # Maps base field path to list of (index, value) tuples
processed_keys = set()
indexed_base_paths = set() # Track which base paths have indexed fields
for key, value in form_data.items():
# Check if this looks like an array index field (ends with .0, .1, .2, etc.)
if '.' in key:
parts = key.rsplit('.', 1) # Split on last dot
if len(parts) == 2:
base_path, last_part = parts
# Check if last part is a numeric string (array index)
if last_part.isdigit():
# Get schema property for the base path to verify it's an array
base_prop = _get_schema_property(schema, base_path)
if base_prop and base_prop.get('type') == 'array':
# This is an array index field
index = int(last_part)
if base_path not in array_fields:
array_fields[base_path] = []
array_fields[base_path].append((index, value))
processed_keys.add(key)
indexed_base_paths.add(base_path)
continue
# Process combined array fields
for base_path, index_values in array_fields.items():
# Sort by index and extract values
index_values.sort(key=lambda x: x[0])
values = [v for _, v in index_values]
# Combine values into comma-separated string for parsing
combined_value = ', '.join(str(v) for v in values)
# Parse as array using schema
parsed_value = _parse_form_value_with_schema(combined_value, base_path, schema)
# Debug logging
logger.debug(f"Combined indexed array field {base_path}: {values} -> {combined_value} -> {parsed_value}")
# Only set if not skipped
if parsed_value is not _SKIP_FIELD:
_set_nested_value(plugin_config, base_path, parsed_value)
# Process remaining (non-indexed) fields
# Skip any base paths that were processed as indexed arrays
for key, value in form_data.items():
if key not in processed_keys:
# Skip if this key is a base path that was processed as indexed array
# (to avoid overwriting the combined array with a single value)
if key not in indexed_base_paths:
# Parse value using schema to determine correct type
parsed_value = _parse_form_value_with_schema(value, key, schema)
# Debug logging for array fields
if schema:
prop = _get_schema_property(schema, key)
if prop and prop.get('type') == 'array':
logger.debug(f"Array field {key}: form value='{value}' -> parsed={parsed_value}")
# Use helper to set nested values correctly (skips if _SKIP_FIELD)
if parsed_value is not _SKIP_FIELD:
_set_nested_value(plugin_config, key, parsed_value)
# Post-process: Fix array fields that might have been incorrectly structured
# This handles cases where array fields are stored as dicts (e.g., from indexed form fields)
def fix_array_structures(config_dict, schema_props, prefix=''):
"""Recursively fix array structures (convert dicts with numeric keys to arrays, fix length issues)"""
for prop_key, prop_schema in schema_props.items():
prop_type = prop_schema.get('type')
if prop_type == 'array':
# Navigate to the field location
if prefix:
parent_parts = prefix.split('.')
parent = config_dict
for part in parent_parts:
if isinstance(parent, dict) and part in parent:
parent = parent[part]
else:
parent = None
break
if parent is not None and isinstance(parent, dict) and prop_key in parent:
current_value = parent[prop_key]
# If it's a dict with numeric string keys, convert to array
if isinstance(current_value, dict) and not isinstance(current_value, list):
try:
# Check if all keys are numeric strings (array indices)
keys = [k for k in current_value.keys()]
if all(k.isdigit() for k in keys):
# Convert to sorted array by index
sorted_keys = sorted(keys, key=int)
array_value = [current_value[k] for k in sorted_keys]
# Convert array elements to correct types based on schema
items_schema = prop_schema.get('items', {})
item_type = items_schema.get('type')
if item_type in ('number', 'integer'):
converted_array = []
for v in array_value:
if isinstance(v, str):
try:
if item_type == 'integer':
converted_array.append(int(v))
else:
converted_array.append(float(v))
except (ValueError, TypeError):
converted_array.append(v)
else:
converted_array.append(v)
array_value = converted_array
parent[prop_key] = array_value
current_value = array_value # Update for length check below
except (ValueError, KeyError, TypeError):
# Conversion failed, check if we should use default
pass
# If it's an array, ensure correct types and check minItems
if isinstance(current_value, list):
# First, ensure array elements are correct types
items_schema = prop_schema.get('items', {})
item_type = items_schema.get('type')
if item_type in ('number', 'integer'):
converted_array = []
for v in current_value:
if isinstance(v, str):
try:
if item_type == 'integer':
converted_array.append(int(v))
else:
converted_array.append(float(v))
except (ValueError, TypeError):
converted_array.append(v)
else:
converted_array.append(v)
parent[prop_key] = converted_array
current_value = converted_array
# Then check minItems
min_items = prop_schema.get('minItems')
if min_items is not None and len(current_value) < min_items:
# Use default if available, otherwise keep as-is (validation will catch it)
default = prop_schema.get('default')
if default and isinstance(default, list) and len(default) >= min_items:
parent[prop_key] = default
else:
# Top-level field
if prop_key in config_dict:
current_value = config_dict[prop_key]
# If it's a dict with numeric string keys, convert to array
if isinstance(current_value, dict) and not isinstance(current_value, list):
try:
keys = list(current_value.keys())
if keys and all(str(k).isdigit() for k in keys):
sorted_keys = sorted(keys, key=lambda x: int(str(x)))
array_value = [current_value[k] for k in sorted_keys]
# Convert array elements to correct types based on schema
items_schema = prop_schema.get('items', {})
item_type = items_schema.get('type')
if item_type in ('number', 'integer'):
converted_array = []
for v in array_value:
if isinstance(v, str):
try:
if item_type == 'integer':
converted_array.append(int(v))
else:
converted_array.append(float(v))
except (ValueError, TypeError):
converted_array.append(v)
else:
converted_array.append(v)
array_value = converted_array
config_dict[prop_key] = array_value
current_value = array_value # Update for length check below
except (ValueError, KeyError, TypeError) as e:
logger.debug(f"Failed to convert {prop_key} to array: {e}")
# If it's an array, ensure correct types and check minItems
if isinstance(current_value, list):
# First, ensure array elements are correct types
items_schema = prop_schema.get('items', {})
item_type = items_schema.get('type')
if item_type in ('number', 'integer'):
converted_array = []
for v in current_value:
if isinstance(v, str):
try:
if item_type == 'integer':
converted_array.append(int(v))
else:
converted_array.append(float(v))
except (ValueError, TypeError):
converted_array.append(v)
else:
converted_array.append(v)
config_dict[prop_key] = converted_array
current_value = converted_array
# Then check minItems
min_items = prop_schema.get('minItems')
if min_items is not None and len(current_value) < min_items:
default = prop_schema.get('default')
if default and isinstance(default, list) and len(default) >= min_items:
config_dict[prop_key] = default
# Recurse into nested objects
elif prop_type == 'object' and 'properties' in prop_schema:
nested_prefix = f"{prefix}.{prop_key}" if prefix else prop_key
if prefix:
parent_parts = prefix.split('.')
parent = config_dict
for part in parent_parts:
if isinstance(parent, dict) and part in parent:
parent = parent[part]
else:
parent = None
break
nested_dict = parent.get(prop_key) if parent is not None and isinstance(parent, dict) else None
else:
nested_dict = config_dict.get(prop_key)
if isinstance(nested_dict, dict):
# Pass no prefix: config_dict is already the navigated sub-dict,
# so path segments from the parent would mis-navigate it.
fix_array_structures(nested_dict, prop_schema['properties'])
# Also ensure array fields that are None get converted to empty arrays
def ensure_array_defaults(config_dict, schema_props, prefix=''):
"""Recursively ensure array fields have defaults if None"""
for prop_key, prop_schema in schema_props.items():
prop_type = prop_schema.get('type')
if prop_type == 'array':
if prefix:
parent_parts = prefix.split('.')
parent = config_dict
for part in parent_parts:
if isinstance(parent, dict) and part in parent:
parent = parent[part]
else:
parent = None
break
if parent is not None and isinstance(parent, dict):
if prop_key not in parent or parent[prop_key] is None:
default = prop_schema.get('default', [])
parent[prop_key] = default if default else []
else:
if prop_key not in config_dict or config_dict[prop_key] is None:
default = prop_schema.get('default', [])
config_dict[prop_key] = default if default else []
elif prop_type == 'object' and 'properties' in prop_schema:
nested_prefix = f"{prefix}.{prop_key}" if prefix else prop_key
if prefix:
parent_parts = prefix.split('.')
parent = config_dict
for part in parent_parts:
if isinstance(parent, dict) and part in parent:
parent = parent[part]
else:
parent = None
break
nested_dict = parent.get(prop_key) if parent is not None and isinstance(parent, dict) else None
else:
nested_dict = config_dict.get(prop_key)
if nested_dict is None:
if prefix:
parent_parts = prefix.split('.')
parent = config_dict
for part in parent_parts:
if part not in parent:
parent[part] = {}
parent = parent[part]
if prop_key not in parent:
parent[prop_key] = {}
nested_dict = parent[prop_key]
else:
if prop_key not in config_dict:
config_dict[prop_key] = {}
nested_dict = config_dict[prop_key]
if isinstance(nested_dict, dict):
# Pass no prefix: config_dict is already navigated.
ensure_array_defaults(nested_dict, prop_schema['properties'])
if schema and 'properties' in schema:
# First, fix any dict structures that should be arrays
# This must be called BEFORE validation to convert dicts with numeric keys to arrays
fix_array_structures(plugin_config, schema['properties'])
# Then, ensure None arrays get defaults
ensure_array_defaults(plugin_config, schema['properties'])
# Debug: Log the structure after fixing
if 'feeds' in plugin_config and 'custom_feeds' in plugin_config.get('feeds', {}):
custom_feeds = plugin_config['feeds']['custom_feeds']
logger.debug(f"After fix_array_structures: custom_feeds type={type(custom_feeds)}, value={custom_feeds}")
# Force fix for feeds.custom_feeds if it's still a dict (fallback)
if 'feeds' in plugin_config:
feeds_config = plugin_config.get('feeds') or {}
if feeds_config and 'custom_feeds' in feeds_config and isinstance(feeds_config['custom_feeds'], dict):
custom_feeds_dict = feeds_config['custom_feeds']
# Check if all keys are numeric
keys = list(custom_feeds_dict.keys())
if keys and all(str(k).isdigit() for k in keys):
# Convert to array
sorted_keys = sorted(keys, key=lambda x: int(str(x)))
feeds_config['custom_feeds'] = [custom_feeds_dict[k] for k in sorted_keys]
logger.info(f"Force-converted feeds.custom_feeds from dict to array: {len(feeds_config['custom_feeds'])} items")
# Fix unchecked boolean checkboxes: HTML checkboxes don't submit values
# when unchecked, so the existing config value (potentially True) persists.
# Walk the schema and set any boolean fields missing from form data to False.
if schema and 'properties' in schema:
form_keys = set(form.keys())
_set_missing_booleans_to_false(plugin_config, schema['properties'], form_keys)
return plugin_config
@api_v3.route('/plugins/config', methods=['POST'])
def save_plugin_config():
"""Save plugin configuration, separating secrets from regular config"""
@@ -4336,359 +4702,9 @@ def save_plugin_config():
# Start with existing config and apply form updates
plugin_config = existing_config
# Convert form data to config dict
# Form fields can use dot notation for nested values (e.g., "transition.type")
form_data = request.form.to_dict()
# First pass: handle bracket notation array fields (e.g., "field_name[]" from checkbox-group)
# These fields use getlist() to preserve all values, then replace in form_data
# Sentinel empty value ("") allows clearing array to [] when all checkboxes unchecked
bracket_array_fields = {} # Maps base field path to list of values
for key in request.form.keys():
# Check if key ends with "[]" (bracket notation for array fields)
if key.endswith('[]'):
base_path = key[:-2] # Remove "[]" suffix
values = request.form.getlist(key)
# Filter out sentinel empty string - if only sentinel present, array should be []
# If sentinel + values present, use the actual values
filtered_values = [v for v in values if v and v.strip()]
# If no non-empty values but key exists, it means all checkboxes unchecked (empty array)
bracket_array_fields[base_path] = filtered_values
# Remove the bracket notation key from form_data if present
if key in form_data:
del form_data[key]
# Process bracket notation fields and set directly in plugin_config
# Use JSON encoding instead of comma-join to handle values containing commas
import json
for base_path, values in bracket_array_fields.items():
# Get schema property to verify it's an array
base_prop = _get_schema_property(schema, base_path)
if base_prop and base_prop.get('type') == 'array':
# Filter out empty values and sentinel empty strings
filtered_values = [v for v in values if v and v.strip()]
# Set directly in plugin_config (values are already strings, no need to parse)
# Empty array (all unchecked) is represented as []
_set_nested_value(plugin_config, base_path, filtered_values)
logger.debug(f"Processed bracket notation array field {base_path}: {values} -> {filtered_values}")
# Remove from form_data to avoid double processing
if base_path in form_data:
del form_data[base_path]
# Second pass: detect and combine array index fields (e.g., "text_color.0", "text_color.1" -> "text_color" as array)
# This handles cases where forms send array fields as indexed inputs
array_fields = {} # Maps base field path to list of (index, value) tuples
processed_keys = set()
indexed_base_paths = set() # Track which base paths have indexed fields
for key, value in form_data.items():
# Check if this looks like an array index field (ends with .0, .1, .2, etc.)
if '.' in key:
parts = key.rsplit('.', 1) # Split on last dot
if len(parts) == 2:
base_path, last_part = parts
# Check if last part is a numeric string (array index)
if last_part.isdigit():
# Get schema property for the base path to verify it's an array
base_prop = _get_schema_property(schema, base_path)
if base_prop and base_prop.get('type') == 'array':
# This is an array index field
index = int(last_part)
if base_path not in array_fields:
array_fields[base_path] = []
array_fields[base_path].append((index, value))
processed_keys.add(key)
indexed_base_paths.add(base_path)
continue
# Process combined array fields
for base_path, index_values in array_fields.items():
# Sort by index and extract values
index_values.sort(key=lambda x: x[0])
values = [v for _, v in index_values]
# Combine values into comma-separated string for parsing
combined_value = ', '.join(str(v) for v in values)
# Parse as array using schema
parsed_value = _parse_form_value_with_schema(combined_value, base_path, schema)
# Debug logging
logger.debug(f"Combined indexed array field {base_path}: {values} -> {combined_value} -> {parsed_value}")
# Only set if not skipped
if parsed_value is not _SKIP_FIELD:
_set_nested_value(plugin_config, base_path, parsed_value)
# Process remaining (non-indexed) fields
# Skip any base paths that were processed as indexed arrays
for key, value in form_data.items():
if key not in processed_keys:
# Skip if this key is a base path that was processed as indexed array
# (to avoid overwriting the combined array with a single value)
if key not in indexed_base_paths:
# Parse value using schema to determine correct type
parsed_value = _parse_form_value_with_schema(value, key, schema)
# Debug logging for array fields
if schema:
prop = _get_schema_property(schema, key)
if prop and prop.get('type') == 'array':
logger.debug(f"Array field {key}: form value='{value}' -> parsed={parsed_value}")
# Use helper to set nested values correctly (skips if _SKIP_FIELD)
if parsed_value is not _SKIP_FIELD:
_set_nested_value(plugin_config, key, parsed_value)
# Post-process: Fix array fields that might have been incorrectly structured
# This handles cases where array fields are stored as dicts (e.g., from indexed form fields)
def fix_array_structures(config_dict, schema_props, prefix=''):
"""Recursively fix array structures (convert dicts with numeric keys to arrays, fix length issues)"""
for prop_key, prop_schema in schema_props.items():
prop_type = prop_schema.get('type')
if prop_type == 'array':
# Navigate to the field location
if prefix:
parent_parts = prefix.split('.')
parent = config_dict
for part in parent_parts:
if isinstance(parent, dict) and part in parent:
parent = parent[part]
else:
parent = None
break
if parent is not None and isinstance(parent, dict) and prop_key in parent:
current_value = parent[prop_key]
# If it's a dict with numeric string keys, convert to array
if isinstance(current_value, dict) and not isinstance(current_value, list):
try:
# Check if all keys are numeric strings (array indices)
keys = [k for k in current_value.keys()]
if all(k.isdigit() for k in keys):
# Convert to sorted array by index
sorted_keys = sorted(keys, key=int)
array_value = [current_value[k] for k in sorted_keys]
# Convert array elements to correct types based on schema
items_schema = prop_schema.get('items', {})
item_type = items_schema.get('type')
if item_type in ('number', 'integer'):
converted_array = []
for v in array_value:
if isinstance(v, str):
try:
if item_type == 'integer':
converted_array.append(int(v))
else:
converted_array.append(float(v))
except (ValueError, TypeError):
converted_array.append(v)
else:
converted_array.append(v)
array_value = converted_array
parent[prop_key] = array_value
current_value = array_value # Update for length check below
except (ValueError, KeyError, TypeError):
# Conversion failed, check if we should use default
pass
# If it's an array, ensure correct types and check minItems
if isinstance(current_value, list):
# First, ensure array elements are correct types
items_schema = prop_schema.get('items', {})
item_type = items_schema.get('type')
if item_type in ('number', 'integer'):
converted_array = []
for v in current_value:
if isinstance(v, str):
try:
if item_type == 'integer':
converted_array.append(int(v))
else:
converted_array.append(float(v))
except (ValueError, TypeError):
converted_array.append(v)
else:
converted_array.append(v)
parent[prop_key] = converted_array
current_value = converted_array
# Then check minItems
min_items = prop_schema.get('minItems')
if min_items is not None and len(current_value) < min_items:
# Use default if available, otherwise keep as-is (validation will catch it)
default = prop_schema.get('default')
if default and isinstance(default, list) and len(default) >= min_items:
parent[prop_key] = default
else:
# Top-level field
if prop_key in config_dict:
current_value = config_dict[prop_key]
# If it's a dict with numeric string keys, convert to array
if isinstance(current_value, dict) and not isinstance(current_value, list):
try:
keys = list(current_value.keys())
if keys and all(str(k).isdigit() for k in keys):
sorted_keys = sorted(keys, key=lambda x: int(str(x)))
array_value = [current_value[k] for k in sorted_keys]
# Convert array elements to correct types based on schema
items_schema = prop_schema.get('items', {})
item_type = items_schema.get('type')
if item_type in ('number', 'integer'):
converted_array = []
for v in array_value:
if isinstance(v, str):
try:
if item_type == 'integer':
converted_array.append(int(v))
else:
converted_array.append(float(v))
except (ValueError, TypeError):
converted_array.append(v)
else:
converted_array.append(v)
array_value = converted_array
config_dict[prop_key] = array_value
current_value = array_value # Update for length check below
except (ValueError, KeyError, TypeError) as e:
logger.debug(f"Failed to convert {prop_key} to array: {e}")
# If it's an array, ensure correct types and check minItems
if isinstance(current_value, list):
# First, ensure array elements are correct types
items_schema = prop_schema.get('items', {})
item_type = items_schema.get('type')
if item_type in ('number', 'integer'):
converted_array = []
for v in current_value:
if isinstance(v, str):
try:
if item_type == 'integer':
converted_array.append(int(v))
else:
converted_array.append(float(v))
except (ValueError, TypeError):
converted_array.append(v)
else:
converted_array.append(v)
config_dict[prop_key] = converted_array
current_value = converted_array
# Then check minItems
min_items = prop_schema.get('minItems')
if min_items is not None and len(current_value) < min_items:
default = prop_schema.get('default')
if default and isinstance(default, list) and len(default) >= min_items:
config_dict[prop_key] = default
# Recurse into nested objects
elif prop_type == 'object' and 'properties' in prop_schema:
nested_prefix = f"{prefix}.{prop_key}" if prefix else prop_key
if prefix:
parent_parts = prefix.split('.')
parent = config_dict
for part in parent_parts:
if isinstance(parent, dict) and part in parent:
parent = parent[part]
else:
parent = None
break
nested_dict = parent.get(prop_key) if parent is not None and isinstance(parent, dict) else None
else:
nested_dict = config_dict.get(prop_key)
if isinstance(nested_dict, dict):
# Pass no prefix: config_dict is already the navigated sub-dict,
# so path segments from the parent would mis-navigate it.
fix_array_structures(nested_dict, prop_schema['properties'])
# Also ensure array fields that are None get converted to empty arrays
def ensure_array_defaults(config_dict, schema_props, prefix=''):
"""Recursively ensure array fields have defaults if None"""
for prop_key, prop_schema in schema_props.items():
prop_type = prop_schema.get('type')
if prop_type == 'array':
if prefix:
parent_parts = prefix.split('.')
parent = config_dict
for part in parent_parts:
if isinstance(parent, dict) and part in parent:
parent = parent[part]
else:
parent = None
break
if parent is not None and isinstance(parent, dict):
if prop_key not in parent or parent[prop_key] is None:
default = prop_schema.get('default', [])
parent[prop_key] = default if default else []
else:
if prop_key not in config_dict or config_dict[prop_key] is None:
default = prop_schema.get('default', [])
config_dict[prop_key] = default if default else []
elif prop_type == 'object' and 'properties' in prop_schema:
nested_prefix = f"{prefix}.{prop_key}" if prefix else prop_key
if prefix:
parent_parts = prefix.split('.')
parent = config_dict
for part in parent_parts:
if isinstance(parent, dict) and part in parent:
parent = parent[part]
else:
parent = None
break
nested_dict = parent.get(prop_key) if parent is not None and isinstance(parent, dict) else None
else:
nested_dict = config_dict.get(prop_key)
if nested_dict is None:
if prefix:
parent_parts = prefix.split('.')
parent = config_dict
for part in parent_parts:
if part not in parent:
parent[part] = {}
parent = parent[part]
if prop_key not in parent:
parent[prop_key] = {}
nested_dict = parent[prop_key]
else:
if prop_key not in config_dict:
config_dict[prop_key] = {}
nested_dict = config_dict[prop_key]
if isinstance(nested_dict, dict):
# Pass no prefix: config_dict is already navigated.
ensure_array_defaults(nested_dict, prop_schema['properties'])
if schema and 'properties' in schema:
# First, fix any dict structures that should be arrays
# This must be called BEFORE validation to convert dicts with numeric keys to arrays
fix_array_structures(plugin_config, schema['properties'])
# Then, ensure None arrays get defaults
ensure_array_defaults(plugin_config, schema['properties'])
# Debug: Log the structure after fixing
if 'feeds' in plugin_config and 'custom_feeds' in plugin_config.get('feeds', {}):
custom_feeds = plugin_config['feeds']['custom_feeds']
logger.debug(f"After fix_array_structures: custom_feeds type={type(custom_feeds)}, value={custom_feeds}")
# Force fix for feeds.custom_feeds if it's still a dict (fallback)
if 'feeds' in plugin_config:
feeds_config = plugin_config.get('feeds') or {}
if feeds_config and 'custom_feeds' in feeds_config and isinstance(feeds_config['custom_feeds'], dict):
custom_feeds_dict = feeds_config['custom_feeds']
# Check if all keys are numeric
keys = list(custom_feeds_dict.keys())
if keys and all(str(k).isdigit() for k in keys):
# Convert to array
sorted_keys = sorted(keys, key=lambda x: int(str(x)))
feeds_config['custom_feeds'] = [custom_feeds_dict[k] for k in sorted_keys]
logger.info(f"Force-converted feeds.custom_feeds from dict to array: {len(feeds_config['custom_feeds'])} items")
# Fix unchecked boolean checkboxes: HTML checkboxes don't submit values
# when unchecked, so the existing config value (potentially True) persists.
# Walk the schema and set any boolean fields missing from form data to False.
if schema and 'properties' in schema:
form_keys = set(request.form.keys())
_set_missing_booleans_to_false(plugin_config, schema['properties'], form_keys)
# Convert form data to config dict (shared with the preview
# endpoint — see parse_plugin_config_form)
plugin_config = parse_plugin_config_form(request.form, schema, plugin_config)
# Get schema manager instance (for JSON requests)
schema_mgr = api_v3.schema_manager
@@ -5290,6 +5306,233 @@ def get_plugin_schema():
logger.error('Error in get_plugin_schema', exc_info=True)
return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details'}), 500
# plugin_id arrives in request input and is used to build filesystem paths —
# allowlist it (same pattern pages_v3 uses)
_SAFE_PREVIEW_PLUGIN_ID_RE = re.compile(r'^[a-zA-Z0-9_-]{1,64}$')
def _find_plugin_dir_for_preview(plugin_id: str) -> 'Path | None':
"""Locate an installed plugin's directory (store dir, then dev dirs) —
same search order the schema manager uses. Rejects any id that could
name a path outside the plugin directories."""
if not isinstance(plugin_id, str) or not _SAFE_PREVIEW_PLUGIN_ID_RE.match(plugin_id):
return None
candidates = []
active_pm = getattr(api_v3, 'plugin_manager', None)
if active_pm and getattr(active_pm, 'plugins_dir', None):
candidates.append(Path(active_pm.plugins_dir))
else:
_cm = getattr(api_v3, 'config_manager', None)
_cfg = _cm.load_config() if _cm else {}
_dir_name = _cfg.get('plugin_system', {}).get('plugins_directory', 'plugin-repos')
candidates.append(Path(_dir_name) if os.path.isabs(_dir_name)
else PROJECT_ROOT / _dir_name)
candidates.append(PROJECT_ROOT / 'plugins')
candidates.append(PROJECT_ROOT / 'plugin-repos')
for base in candidates:
plugin_dir = base / plugin_id
if (plugin_dir / 'manifest.json').exists():
return plugin_dir
return None
PREVIEW_MIN_SIZE, PREVIEW_MAX_W, PREVIEW_MAX_H = 8, 1024, 512
PREVIEW_RENDER_TIMEOUT_SEC = 15
@api_v3.route('/plugins/preview', methods=['POST'])
def preview_plugin_render():
"""Render a plugin headlessly with a CANDIDATE (unsaved) config.
Powers the config page's live preview: the browser posts the current
form state (same encoding as save parsed by the same
parse_plugin_config_form, so preview and save can never disagree) or a
JSON body {"config": {...}}, and gets back a base64 PNG of what the
panel would show.
Entirely hardware-free: renders through VisualTestDisplayManager (pure
PIL) with install_deps=False. update() is skipped by default so the
request never blocks on live APIs plugins with a test/harness.json
get their mock-data fixture primed into the cache instead, and
?skip_update=0 opts into a real update() for plugins that need it.
Query params: plugin_id (required); width/height (defaults: the real
panel size from display.hardware); skip_update (default 1).
"""
try:
plugin_id = request.args.get('plugin_id')
if not plugin_id:
return error_response(ErrorCode.INVALID_INPUT,
'plugin_id required in query string',
status_code=400)
plugin_dir = _find_plugin_dir_for_preview(plugin_id)
if not plugin_dir:
return error_response(ErrorCode.PLUGIN_NOT_FOUND,
f'Plugin not found: {plugin_id}',
status_code=404)
schema_mgr = api_v3.schema_manager
if not schema_mgr:
return error_response(ErrorCode.SYSTEM_ERROR,
'Schema manager not initialized',
status_code=500)
# ---- panel size: explicit query args, else the real panel ----
main_config = {}
if api_v3.config_manager:
try:
main_config = api_v3.config_manager.load_config() or {}
except Exception:
main_config = {}
hardware = main_config.get('display', {}).get('hardware', {})
default_width = int(hardware.get('cols', 64)) * int(hardware.get('chain_length', 2))
default_height = int(hardware.get('rows', 32)) * int(hardware.get('parallel', 1))
# The UI's size selector posts "__preview_size=WxH" via the button's
# hx-vals (evaluated at request time — htmx caches hx-post's path at
# process time, so a dynamically updated query string doesn't work).
# Explicit query args still take precedence for API callers.
preview_size = request.values.get('__preview_size', '')
if preview_size and 'x' in preview_size and 'width' not in request.args:
size_w, _, size_h = preview_size.partition('x')
try:
default_width, default_height = int(size_w), int(size_h)
except (TypeError, ValueError):
pass # malformed selector value — fall back to panel size
try:
width = int(request.args.get('width', default_width))
height = int(request.args.get('height', default_height))
except (TypeError, ValueError):
return error_response(ErrorCode.INVALID_INPUT,
'width and height must be integers',
status_code=400)
if not (PREVIEW_MIN_SIZE <= width <= PREVIEW_MAX_W
and PREVIEW_MIN_SIZE <= height <= PREVIEW_MAX_H):
return error_response(
ErrorCode.INVALID_INPUT,
f'size must be within {PREVIEW_MIN_SIZE}x{PREVIEW_MIN_SIZE} '
f'and {PREVIEW_MAX_W}x{PREVIEW_MAX_H}',
status_code=400)
# ---- candidate config: saved config + submitted changes + defaults ----
schema = schema_mgr.load_schema(plugin_id, use_cache=False)
existing_config = (main_config.get(plugin_id) or {}).copy()
content_type = request.content_type or ''
if 'application/json' in content_type:
import copy
data = request.get_json(silent=True) or {}
# Deep-merge the candidate onto the saved config, matching how
# the form path (and save) treat partial updates — a shallow
# update() would silently drop the user's saved values in any
# nested section the candidate touches (e.g. posting one
# element's color would discard the saved font of its sibling).
def _deep_merge(base, overlay):
for key, value in overlay.items():
if (isinstance(value, dict)
and isinstance(base.get(key), dict)):
_deep_merge(base[key], value)
else:
base[key] = value
plugin_config = copy.deepcopy(existing_config)
_deep_merge(plugin_config, data.get('config', {}))
else:
# Strip the preview-only control field so it never lands in the
# candidate config the plugin sees
form = request.form.copy()
form.poplist('__preview_size')
plugin_config = parse_plugin_config_form(form, schema,
existing_config)
if schema:
defaults = schema_mgr.generate_default_config(plugin_id, use_cache=True)
plugin_config = schema_mgr.merge_with_defaults(plugin_config, defaults)
# Preview regardless of the enabled toggle
plugin_config['enabled'] = True
# ---- deterministic data: the plugin's own harness fixture ----
mock_data = {}
try:
from src.plugin_system.testing.loading import load_harness_spec
spec = load_harness_spec(plugin_dir)
mock_data = spec.get('mock_data_contents', {}) or {}
harness_config = spec.get('config') or {}
if harness_config:
# harness settings under the candidate config: user's
# in-form values always win
merged = dict(harness_config)
merged.update(plugin_config)
plugin_config = merged
except Exception as e:
logger.debug(f'No usable harness spec for {plugin_id}: {e}')
skip_update = request.args.get('skip_update', '1') not in ('0', 'false')
# Bounded render: a plugin whose update()/display() hangs must not
# pin a web worker forever. The runaway thread can't be killed, but
# the request returns and the thread is daemonized so it can't block
# shutdown either.
from src.plugin_system.testing.render_service import render_plugin_once
import threading
render_out: dict = {}
def _do_render():
try:
render_out['result'] = render_plugin_once(
plugin_id, plugin_dir, config=plugin_config,
mock_data=mock_data, width=width, height=height,
skip_update=skip_update)
except Exception as e: # surfaced below
render_out['error'] = e
render_thread = threading.Thread(target=_do_render, daemon=True,
name=f'preview-{plugin_id}')
render_thread.start()
render_thread.join(timeout=PREVIEW_RENDER_TIMEOUT_SEC)
if render_thread.is_alive():
logger.warning('preview render timed out for %s after %ss',
plugin_id, PREVIEW_RENDER_TIMEOUT_SEC)
if request.headers.get('HX-Request'):
return Response('<p class="text-xs text-red-600">Preview timed '
'out &mdash; the plugin took too long to render.</p>',
mimetype='text/html')
return error_response(ErrorCode.SYSTEM_ERROR,
'Preview render timed out', status_code=504)
if 'error' in render_out:
raise render_out['error']
result = render_out['result']
# HTMX callers get a ready-to-swap fragment; API callers get JSON
if request.headers.get('HX-Request'):
import html as _html
meta = f"{result['width']}&times;{result['height']} &middot; {result['render_time_ms']} ms"
errors_html = ''
if result['errors'] or result['warnings']:
notes = _html.escape('; '.join(result['errors'] + result['warnings']))
errors_html = (f'<p class="text-xs text-red-600 mt-1">'
f'{notes}</p>')
return Response(
f'<img src="{result["image"]}" alt="Plugin preview" '
f'class="preview-pixelated" '
f'style="image-rendering: pixelated; width: 100%; max-width: '
f'{result["width"] * 4}px; border: 1px solid #333; '
f'border-radius: 4px; background: #000;">'
f'<p class="text-xs text-gray-500 mt-1">{meta}</p>'
f'{errors_html}',
mimetype='text/html')
return jsonify({'status': 'success', 'data': result})
except Exception:
logger.error('Error in preview_plugin_render', exc_info=True)
if request.headers.get('HX-Request'):
return Response('<p class="text-xs text-red-600">Preview failed — '
'see logs for details.</p>', mimetype='text/html')
return error_response(ErrorCode.SYSTEM_ERROR,
'Preview failed; see logs for details',
status_code=500)
@api_v3.route('/plugins/config/reset', methods=['POST'])
def reset_plugin_config():
"""Reset plugin configuration to schema defaults"""
+6
View File
@@ -706,6 +706,12 @@ def _load_plugin_config_partial(plugin_id):
try:
with open(schema_path, 'r', encoding='utf-8') as f:
schema = json.load(f)
# Expand x-style-elements declarations into full property
# blocks — the same expansion SchemaManager.load_schema
# applies on the API paths. The form must render the exact
# shape the save path parses and validates against.
from src.element_style import expand_style_elements
schema = expand_style_elements(schema)
except Exception as e:
logger.warning("Could not load schema for plugin: %s", e)
@@ -994,8 +994,51 @@
<p class="text-xs text-amber-600">Plugin is disabled, but on-demand will temporarily enable it.</p>
{% endif %}
</div>
{# Live Preview — renders the plugin headlessly with the CURRENT
(unsaved) form values, at the real panel size or a chosen one #}
<div class="mt-4 pt-4 border-t border-gray-200 space-y-3">
<div class="flex items-center gap-2">
<i class="fas fa-eye text-blue-500"></i>
<span class="text-sm font-semibold text-gray-900">Live Preview</span>
</div>
<div class="flex flex-wrap items-center gap-2">
{# No name attribute: must not be submitted with the config
save. The size travels via the button's hx-vals (read at
request time — htmx caches hx-post's path at process
time, so mutating the attribute onchange doesn't work). #}
<select id="preview-size-{{ plugin.id }}"
aria-label="Preview panel size"
class="text-sm rounded-md border-gray-300 py-1.5">
<option value="">My panel size</option>
<option value="64x32">64 &times; 32</option>
<option value="128x32">128 &times; 32</option>
<option value="128x64">128 &times; 64</option>
<option value="192x48">192 &times; 48</option>
<option value="256x128">256 &times; 128</option>
</select>
<button type="button"
id="preview-btn-{{ plugin.id }}"
hx-post="/api/v3/plugins/preview?plugin_id={{ plugin.id }}"
hx-include="closest form"
hx-vals='js:{"__preview_size": document.getElementById("preview-size-{{ plugin.id }}").value}'
hx-target="#plugin-preview-{{ plugin.id }}"
hx-swap="innerHTML"
hx-indicator="#preview-indicator-{{ plugin.id }}"
class="px-3 py-2 text-sm bg-blue-600 hover:bg-blue-700 text-white rounded-md flex items-center gap-2 transition-colors">
<i class="fas fa-tv"></i>
<span>Preview</span>
</button>
<span id="preview-indicator-{{ plugin.id }}" class="htmx-indicator text-xs text-gray-500">
<i class="fas fa-spinner fa-spin"></i> rendering&hellip;
</span>
</div>
<div id="plugin-preview-{{ plugin.id }}">
<p class="text-xs text-gray-400">Renders this plugin with the current (unsaved) settings &mdash; try it before you save.</p>
</div>
</div>
</div>
{# Configuration Form Panel #}
<div class="bg-gray-50 rounded-lg p-4">
<h3 class="text-md font-medium text-gray-900 mb-3">Configuration</h3>