diff --git a/scripts/dev_server.py b/scripts/dev_server.py index 60976ae80..4743ba8ca 100644 --- a/scripts/dev_server.py +++ b/scripts/dev_server.py @@ -200,60 +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 - - display_manager = VisualTestDisplayManager(width=width, height=height) - cache_manager = MockCacheManager() - plugin_manager = MockPluginManager() + from src.plugin_system.testing.render_service import render_plugin_once - # 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: - warnings.append(f"update() raised: {e}") - - # Run display() - try: - plugin_instance.display(force_clear=True) - except Exception as e: - errors.append(f"display() raised: {e}") - - 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]: diff --git a/src/element_style.py b/src/element_style.py new file mode 100644 index 000000000..da5b336fa --- /dev/null +++ b/src/element_style.py @@ -0,0 +1,533 @@ +"""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). + + An object property carrying its own ``default`` short-circuits recursion, + matching the schema manager's behavior. + """ + 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 + 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. 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. 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.`` first (the deployed sports + convention), falling back to ``customization.`` 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 clear_cache(self) -> None: + self._cache.clear() diff --git a/src/plugin_system/base_plugin.py b/src/plugin_system/base_plugin.py index a990ecd93..b8285d688 100644 --- a/src/plugin_system/base_plugin.py +++ b/src/plugin_system/base_plugin.py @@ -270,6 +270,64 @@ def _get_design_size(self) -> tuple: 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. 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 @@ def on_config_change(self, new_config: Dict[str, Any]) -> None: # 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. diff --git a/src/plugin_system/schema_manager.py b/src/plugin_system/schema_manager.py index 385b53fd7..779338772 100644 --- a/src/plugin_system/schema_manager.py +++ b/src/plugin_system/schema_manager.py @@ -110,12 +110,17 @@ def load_schema(self, plugin_id: str, use_cache: bool = True) -> Optional[Dict[s 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 @@ def load_schema(self, plugin_id: str, use_cache: bool = True) -> Optional[Dict[s 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 """ diff --git a/src/plugin_system/testing/render_service.py b/src/plugin_system/testing/render_service.py new file mode 100644 index 000000000..5a1b8c7dd --- /dev/null +++ b/src/plugin_system/testing/render_service.py @@ -0,0 +1,96 @@ +"""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 time +from pathlib import Path +from typing import Any, Dict, Optional + + +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() + + if not skip_update: + try: + plugin_instance.update() + except Exception as e: + warnings.append(f"update() raised: {e}") + + try: + plugin_instance.display(force_clear=True) + except Exception as e: + errors.append(f"display() raised: {e}") + + 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, + } diff --git a/test/test_element_style.py b/test/test_element_style.py new file mode 100644 index 000000000..a42912302 --- /dev/null +++ b/test/test_element_style.py @@ -0,0 +1,509 @@ +"""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..""" + 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}}}, + }, + } + 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"] == {} + + 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"])) diff --git a/test/test_schema_style_expansion.py b/test/test_schema_style_expansion.py new file mode 100644 index 000000000..233365d46 --- /dev/null +++ b/test/test_schema_style_expansion.py @@ -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"])) diff --git a/test/web_interface/test_plugin_preview.py b/test/web_interface/test_plugin_preview.py new file mode 100644 index 000000000..5c27fc1cb --- /dev/null +++ b/test/web_interface/test_plugin_preview.py @@ -0,0 +1,214 @@ +"""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_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 " {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""" @@ -4335,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 @@ -5289,6 +5306,189 @@ 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 + + +@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: + data = request.get_json(silent=True) or {} + plugin_config = existing_config + plugin_config.update(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') + + from src.plugin_system.testing.render_service import render_plugin_once + result = render_plugin_once( + plugin_id, plugin_dir, config=plugin_config, + mock_data=mock_data, width=width, height=height, + skip_update=skip_update) + + # 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']}×{result['height']} · {result['render_time_ms']} ms" + errors_html = '' + if result['errors'] or result['warnings']: + notes = _html.escape('; '.join(result['errors'] + result['warnings'])) + errors_html = (f'

' + f'{notes}

') + return Response( + f'Plugin preview' + f'

{meta}

' + 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('

Preview failed — ' + 'see logs for details.

', 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""" diff --git a/web_interface/blueprints/pages_v3.py b/web_interface/blueprints/pages_v3.py index 8c3c1bbc6..c839c5dc9 100644 --- a/web_interface/blueprints/pages_v3.py +++ b/web_interface/blueprints/pages_v3.py @@ -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) diff --git a/web_interface/templates/v3/partials/plugin_config.html b/web_interface/templates/v3/partials/plugin_config.html index e427918f1..fbb366481 100644 --- a/web_interface/templates/v3/partials/plugin_config.html +++ b/web_interface/templates/v3/partials/plugin_config.html @@ -994,8 +994,51 @@

Plugin Information

Plugin is disabled, but on-demand will temporarily enable it.

{% endif %} + + {# Live Preview — renders the plugin headlessly with the CURRENT + (unsaved) form values, at the real panel size or a chosen one #} +
+
+ + Live Preview +
+
+ {# 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). #} + + + + rendering… + +
+
+

Renders this plugin with the current (unsaved) settings — try it before you save.

+
+
- + {# Configuration Form Panel #}

Configuration