mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-04 06:15:09 +00:00
* fix(plugin-system): unload/update race, failed-load module cleanup, limits validation, schema lookup, install rollback, op-queue dedupe - unload_plugin takes the per-plugin lock (5s bounded) before cleanup(), and an update() that finishes after its plugin was unloaded no longer sets the state back to ENABLED. - A load that fails after import drops plugin_<id> and its submodules and forgets its manager fonts, so a fixed plugin reloads new code. - Resource limits are validated as non-negative numbers: 400 at POST /plugins/limits, bad cached records ignored with one warning. Route docstrings note health/metrics reset and limits only change the web process's view. - SchemaManager.get_schema_path resolves each search dir via resolve_plugin_dir (manifest id, ledmatrix-<id>) before the literal paths; plugins/ still before plugin-repos/. Misses cached 30s and logged once at DEBUG. - install_from_url sets an existing copy aside and restores it if the move fails, under the per-plugin reinstall lock. - Operation queue refuses a second pending op for a plugin and trims _operations with history. - get_vegas_render_width reads display_manager.width first. - get_logger in store/schema/health/resource/saved_repositories; UTF-8 reads in store_manager and state_manager. - Docs: update_interval precedence (manifest over config) stated where users are told to set it in config. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(web): build the limits 400 message from the field name, not an exception CodeQL flagged str(e) flowing into the response. invalid_limit_field() returns the offending field without raising, and limits_from_dict uses it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
836 lines
36 KiB
Python
836 lines
36 KiB
Python
"""
|
|
Schema Manager
|
|
|
|
Manages plugin configuration schemas with caching, validation, and reliable path resolution.
|
|
Provides utilities for extracting defaults, validating configurations, and managing schema lifecycle.
|
|
"""
|
|
|
|
import copy
|
|
import json
|
|
import logging
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
import jsonschema
|
|
from jsonschema import Draft7Validator, ValidationError
|
|
|
|
from src.core_config_keys import CORE_CONFIG_KEYS
|
|
from src.element_style import expand_style_elements
|
|
from src.logging_config import get_logger
|
|
from src.plugin_system.plugin_dirs import resolve_plugin_dir
|
|
|
|
|
|
def _renders_as_object(prop: Dict[str, Any]) -> bool:
|
|
"""``field_type == 'object'`` as ``plugin_config.html`` computes it.
|
|
|
|
The template takes a type list's *first* entry, so ``["object", "null"]``
|
|
is an object and ``["boolean", "object"]`` is a checkbox.
|
|
"""
|
|
field_type = prop.get('type')
|
|
if isinstance(field_type, list):
|
|
field_type = field_type[0] if field_type else None
|
|
return field_type == 'object'
|
|
|
|
|
|
def legacy_bool_as_object(value: Any, prop: Any) -> Any:
|
|
"""Read a boolean stored where the schema now has an ``{enabled, ...}`` object.
|
|
|
|
Plugins turn an on/off switch into a settings object (news'
|
|
``global.dynamic_duration: true`` became ``{enabled, min_duration_seconds,
|
|
...}``), but config.json keeps the boolean until the user next saves that
|
|
plugin's form. The boolean was the switch the object's ``enabled`` now
|
|
holds, so it becomes ``{"enabled": value}`` and schema defaults fill the rest.
|
|
|
|
Returns ``value`` itself when the rule does not apply: anything that is not
|
|
a real ``bool`` (``1``, ``"true"``, ``None``) or a schema property that is not
|
|
an object with an ``enabled`` child. Those stay as they are, so validation
|
|
still reports a genuine mismatch.
|
|
|
|
This is the rule ``render_nested_section`` in
|
|
``web_interface/templates/v3/partials/plugin_config.html`` applies when it
|
|
draws the form. ``test/test_legacy_boolean_config.py`` renders that macro
|
|
against this function, so change both together.
|
|
"""
|
|
if not isinstance(value, bool) or not isinstance(prop, dict):
|
|
return value
|
|
properties = prop.get('properties')
|
|
if (_renders_as_object(prop) and isinstance(properties, dict)
|
|
and 'enabled' in properties):
|
|
return {'enabled': value}
|
|
return value
|
|
|
|
|
|
def normalize_legacy_booleans(config: Any, schema: Any,
|
|
changed_paths: Optional[List[str]] = None,
|
|
_prefix: str = '') -> Any:
|
|
"""Apply :func:`legacy_bool_as_object` at every depth of a plugin config.
|
|
|
|
Walks the config along the schema's ``properties`` the way the settings
|
|
form does: into nested objects, not into array items (the form hands arrays
|
|
to widgets and never applies the rule there).
|
|
|
|
Never mutates ``config``. Returns the same object when nothing changed, and
|
|
otherwise copies only the dicts on the path to each upgraded value. When
|
|
``changed_paths`` is given, the dotted path of each upgraded value is
|
|
appended to it.
|
|
"""
|
|
if not isinstance(config, dict) or not isinstance(schema, dict):
|
|
return config
|
|
properties = schema.get('properties')
|
|
if not isinstance(properties, dict):
|
|
return config
|
|
|
|
result = config
|
|
for key, value in config.items():
|
|
prop = properties.get(key)
|
|
if not isinstance(prop, dict) or not _renders_as_object(prop):
|
|
continue
|
|
path = f"{_prefix}.{key}" if _prefix else key
|
|
new_value = legacy_bool_as_object(value, prop)
|
|
if new_value is not value:
|
|
if changed_paths is not None:
|
|
changed_paths.append(path)
|
|
elif isinstance(value, dict):
|
|
new_value = normalize_legacy_booleans(value, prop, changed_paths, path)
|
|
if new_value is not value:
|
|
if result is config:
|
|
result = dict(config)
|
|
result[key] = new_value
|
|
return result
|
|
|
|
|
|
#: Per-plugin settings the **core** owns: it reads them out of each plugin's
|
|
#: config section, so they are allowed in every plugin's config whether or not
|
|
#: the plugin's schema declares them. The one list for validation, for the web
|
|
#: save filter and for the load-time checks -- a private copy is how JSON saves
|
|
#: came to drop the ``vegas_*`` keys while the validator accepted them.
|
|
#:
|
|
#: Values are the schema used when the plugin does not declare the property.
|
|
CORE_PLUGIN_PROPERTIES: Dict[str, Dict[str, Any]] = {
|
|
# Defaults match BasePlugin behavior: enabled=True, display_duration=15,
|
|
# live_priority=False.
|
|
"enabled": {
|
|
"type": "boolean",
|
|
"default": True,
|
|
"description": "Enable or disable this plugin"
|
|
},
|
|
"display_duration": {
|
|
"type": "number",
|
|
"default": 15,
|
|
"minimum": 1,
|
|
"maximum": 300,
|
|
"description": "How long to display this plugin in seconds"
|
|
},
|
|
"live_priority": {
|
|
"type": "boolean",
|
|
"default": False,
|
|
"description": "Enable live priority takeover when plugin has live content"
|
|
},
|
|
# Vegas tuning read by vegas_mode/plugin_adapter.py and base_plugin.py.
|
|
# Left untyped: the adapter validates them itself and ignores a bad
|
|
# value with a log line, so a stored one must never block a save.
|
|
"vegas_width_pct": {
|
|
"description": "Vegas mode: width of this plugin's card, as a percentage of the panel"
|
|
},
|
|
"vegas_overflow": {
|
|
"description": "Vegas mode: 'rotate' or 'truncate' when this plugin's content overflows"
|
|
},
|
|
"vegas_max_width_screens": {
|
|
"description": "Vegas mode: widest this plugin's card may be, in screens"
|
|
},
|
|
}
|
|
|
|
#: The keys of CORE_PLUGIN_PROPERTIES that are Vegas tuning rather than plugin
|
|
#: state. PluginManager strips these before its soft validation (see
|
|
#: PluginManager.CORE_OWNED_CONFIG_KEYS).
|
|
CORE_VEGAS_TUNING_KEYS = frozenset({
|
|
'vegas_width_pct', 'vegas_overflow', 'vegas_max_width_screens',
|
|
})
|
|
|
|
|
|
#: Per-plugin keys the core used to own and no longer reads. ``skin`` and
|
|
#: ``skin_options`` belonged to the skin system, which was removed; a
|
|
#: config.json written before then can still carry them in any plugin section,
|
|
#: and most plugin schemas set ``additionalProperties: false``. They are
|
|
#: dropped wherever a section is prepared (prepare_plugin_config) or validated,
|
|
#: and the web saves drop them from the stored section, so an old config loads
|
|
#: and saves without a validation error and loses them on its next save.
|
|
RETIRED_PLUGIN_KEYS = frozenset({'skin', 'skin_options'})
|
|
|
|
|
|
def drop_retired_plugin_keys(config: Any, schema: Any) -> Any:
|
|
"""``config`` without the RETIRED_PLUGIN_KEYS its plugin's schema leaves undeclared.
|
|
|
|
A plugin whose schema declares one of these names owns it and keeps it;
|
|
without a schema nothing is dropped. Never mutates ``config``, and returns
|
|
it unchanged when there is nothing to drop.
|
|
"""
|
|
if not isinstance(config, dict) or not isinstance(schema, dict) \
|
|
or RETIRED_PLUGIN_KEYS.isdisjoint(config):
|
|
return config
|
|
declared = schema.get('properties')
|
|
declared = declared if isinstance(declared, dict) else {}
|
|
return {key: value for key, value in config.items()
|
|
if key not in RETIRED_PLUGIN_KEYS or key in declared}
|
|
|
|
|
|
def with_core_plugin_properties(schema: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""A deep copy of a plugin schema with CORE_PLUGIN_PROPERTIES allowed.
|
|
|
|
Properties the plugin declares itself are left as declared. Core
|
|
properties are removed from ``required``: they are system-managed.
|
|
"""
|
|
enhanced = copy.deepcopy(schema) if isinstance(schema, dict) else {}
|
|
properties = enhanced.setdefault("properties", {})
|
|
for name, definition in CORE_PLUGIN_PROPERTIES.items():
|
|
if name not in properties:
|
|
properties[name] = copy.deepcopy(definition)
|
|
if "required" in enhanced:
|
|
enhanced["required"] = [field for field in enhanced["required"]
|
|
if field not in CORE_PLUGIN_PROPERTIES]
|
|
return enhanced
|
|
|
|
|
|
def extract_schema_defaults(schema: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Default values of a JSON Schema's properties, recursively.
|
|
|
|
A property's own ``default`` wins; otherwise a nested object contributes
|
|
its children's defaults, and an array contributes ``[]`` (or a one-item
|
|
list of its ``items`` default). This is what a device runs with, so the
|
|
dev tools use it too (src/plugin_system/testing/loading.py).
|
|
"""
|
|
defaults: Dict[str, Any] = {}
|
|
properties = schema.get('properties', {}) if isinstance(schema, dict) else {}
|
|
if not isinstance(properties, dict):
|
|
return defaults
|
|
|
|
for key, prop_schema in properties.items():
|
|
if not isinstance(prop_schema, dict):
|
|
continue
|
|
# If property has a default, use it
|
|
if 'default' in prop_schema:
|
|
defaults[key] = prop_schema['default']
|
|
continue
|
|
|
|
# Handle nested objects
|
|
if prop_schema.get('type') == 'object' and 'properties' in prop_schema:
|
|
nested_defaults = extract_schema_defaults(prop_schema)
|
|
if nested_defaults:
|
|
defaults[key] = nested_defaults
|
|
|
|
# Handle arrays with object items
|
|
elif prop_schema.get('type') == 'array' and 'items' in prop_schema:
|
|
items_schema = prop_schema['items']
|
|
if items_schema.get('type') == 'object' and 'properties' in items_schema:
|
|
# For arrays of objects, use empty array as default
|
|
# Individual objects will use their defaults when created
|
|
defaults[key] = []
|
|
elif 'default' in items_schema:
|
|
# Array with default item value
|
|
defaults[key] = [items_schema['default']]
|
|
else:
|
|
# Empty array as default
|
|
defaults[key] = []
|
|
|
|
# For other types without defaults, don't add to defaults dict
|
|
# This allows plugins to handle missing values as needed
|
|
|
|
return defaults
|
|
|
|
|
|
def plugin_config_defaults(schema: Optional[Dict[str, Any]]) -> Dict[str, Any]:
|
|
"""Every default a plugin's config gets: the schema's plus the core ones.
|
|
|
|
A plugin with no schema gets the minimal ``enabled: False,
|
|
display_duration: 15``. Device location is not applied here; that needs a
|
|
config manager (SchemaManager.generate_default_config).
|
|
"""
|
|
if not schema:
|
|
return {
|
|
'enabled': False,
|
|
'display_duration': 15
|
|
}
|
|
|
|
defaults = extract_schema_defaults(schema)
|
|
|
|
# Ensure core properties have defaults (they may not be in the schema)
|
|
# These match BasePlugin behavior
|
|
for name in ('enabled', 'display_duration', 'live_priority'):
|
|
if name not in defaults:
|
|
defaults[name] = CORE_PLUGIN_PROPERTIES[name]['default']
|
|
return defaults
|
|
|
|
|
|
def merge_config_defaults(config: Dict[str, Any], defaults: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Merge configuration with defaults, preserving user values.
|
|
|
|
Also replaces None values with defaults so a config never starts with
|
|
None where a default exists. Neither argument is mutated.
|
|
"""
|
|
merged = copy.deepcopy(defaults)
|
|
|
|
def deep_merge(target: Dict[str, Any], source: Dict[str, Any], default_dict: Dict[str, Any]) -> None:
|
|
"""Recursively merge source into target, replacing None with defaults."""
|
|
for key, value in source.items():
|
|
default_value = default_dict.get(key)
|
|
|
|
if key in target and isinstance(target[key], dict) and isinstance(value, dict):
|
|
# Both are dicts, recursively merge
|
|
if isinstance(default_value, dict):
|
|
deep_merge(target[key], value, default_value)
|
|
else:
|
|
deep_merge(target[key], value, {})
|
|
elif value is None and default_value is not None:
|
|
# Value is None and we have a default, use the default
|
|
target[key] = copy.deepcopy(default_value) if isinstance(default_value, (dict, list)) else default_value
|
|
else:
|
|
# Normal merge: user value takes precedence (copy if dict/list)
|
|
if isinstance(value, (dict, list)):
|
|
target[key] = copy.deepcopy(value)
|
|
else:
|
|
target[key] = value
|
|
|
|
deep_merge(merged, config, defaults)
|
|
|
|
# Final pass: replace any remaining None values at any level with defaults
|
|
def replace_none_with_defaults(target: Dict[str, Any], default_dict: Dict[str, Any]) -> None:
|
|
"""Recursively replace None values with defaults."""
|
|
for key in list(target.keys()):
|
|
value = target[key]
|
|
default_value = default_dict.get(key)
|
|
|
|
if value is None and default_value is not None:
|
|
# Replace None with default
|
|
target[key] = copy.deepcopy(default_value) if isinstance(default_value, (dict, list)) else default_value
|
|
elif isinstance(value, dict) and isinstance(default_value, dict):
|
|
# Recursively process nested dicts
|
|
replace_none_with_defaults(value, default_value)
|
|
|
|
replace_none_with_defaults(merged, defaults)
|
|
return merged
|
|
|
|
|
|
def prepare_plugin_config(config: Any, schema: Optional[Dict[str, Any]],
|
|
defaults: Dict[str, Any],
|
|
changed_paths: Optional[List[str]] = None) -> Dict[str, Any]:
|
|
"""The config a plugin runs with, from its stored (or submitted) section.
|
|
|
|
Retired core keys are dropped (drop_retired_plugin_keys), legacy booleans
|
|
are read as ``{"enabled": ...}`` objects (normalize_legacy_booleans), then
|
|
schema defaults fill in whatever is missing. Loading a plugin, both config
|
|
saves, GET /plugins/config, hot reload and the dev tools all go through
|
|
this, so a plugin sees the same shape however its config reached it.
|
|
"""
|
|
config = config if isinstance(config, dict) else {}
|
|
if schema:
|
|
config = drop_retired_plugin_keys(config, schema)
|
|
config = normalize_legacy_booleans(config, schema, changed_paths)
|
|
return merge_config_defaults(config, defaults)
|
|
|
|
|
|
class SchemaManager:
|
|
"""
|
|
Manages plugin configuration schemas with caching and validation.
|
|
|
|
Features:
|
|
- Schema loading and caching
|
|
- Default value extraction from schemas
|
|
- Configuration validation against schemas
|
|
- Reliable path resolution for schema files
|
|
- Cache invalidation on plugin changes
|
|
"""
|
|
|
|
# Plugin config keys that mean "where this device is". A plugin declaring
|
|
# any of these in its schema gets the device-wide ``location`` block from
|
|
# config.json as the *default* for that field, instead of whatever city the
|
|
# plugin author happened to ship. A value the user set on the plugin itself
|
|
# always wins -- this only ever replaces the schema default, so an explicit
|
|
# per-plugin location is still honoured.
|
|
#
|
|
# Only these fully-namespaced keys are substituted. A bare ``state`` or
|
|
# ``city`` key is deliberately left alone: plugins use those for unrelated
|
|
# things (ledmatrix-elections' ``state`` is a two-letter code, not a place
|
|
# name), and silently rewriting them would break those plugins.
|
|
DEVICE_LOCATION_KEYS: Dict[str, str] = {
|
|
'location_city': 'city',
|
|
'location_state': 'state',
|
|
'location_country': 'country',
|
|
}
|
|
|
|
def __init__(self, plugins_dir: Optional[Path] = None, project_root: Optional[Path] = None,
|
|
logger: Optional[logging.Logger] = None, config_manager: Optional[Any] = None):
|
|
"""
|
|
Initialize the Schema Manager.
|
|
|
|
Args:
|
|
plugins_dir: Base plugins directory path
|
|
project_root: Project root directory path
|
|
logger: Optional logger instance
|
|
config_manager: Optional config manager, used to resolve the
|
|
device-wide ``location`` that seeds plugin location defaults.
|
|
Omitting it simply leaves schema defaults untouched.
|
|
"""
|
|
self.logger = logger or get_logger(__name__)
|
|
self.plugins_dir = plugins_dir
|
|
self.project_root = project_root or Path.cwd()
|
|
self.config_manager = config_manager
|
|
|
|
# Schema cache: plugin_id -> schema dict
|
|
self._schema_cache: Dict[str, Dict[str, Any]] = {}
|
|
|
|
# Default config cache: plugin_id -> default config dict
|
|
self._defaults_cache: Dict[str, Dict[str, Any]] = {}
|
|
|
|
# Schema-path misses: plugin_id -> monotonic time of the miss. A
|
|
# lookup now scans each search directory's manifests, and plugins
|
|
# without a schema are asked about on every page render.
|
|
self._schema_path_misses: Dict[str, float] = {}
|
|
self._schema_miss_logged: set = set()
|
|
|
|
#: How long a "no schema" answer is reused before the directories are
|
|
#: searched again -- short, so a plugin installed by a path that doesn't
|
|
#: call invalidate_cache() (a dev symlink, a manual copy) still shows up.
|
|
SCHEMA_MISS_TTL = 30.0
|
|
|
|
def get_schema_path(self, plugin_id: str) -> Optional[Path]:
|
|
"""
|
|
Get the path to a plugin's config_schema.json file.
|
|
|
|
Each search directory -- plugins_dir, then PROJECT_ROOT/plugins, then
|
|
PROJECT_ROOT/plugin-repos -- is first resolved the way the plugin
|
|
loader resolves it (``plugin_dirs.resolve_plugin_dir``: the directory
|
|
whose manifest declares the id, else ``<id>`` / ``ledmatrix-<id>``,
|
|
case-insensitively). Only if none of those holds a schema are the
|
|
literal locations tried:
|
|
1. plugins_dir / plugin_id / config_schema.json
|
|
2. PROJECT_ROOT / plugins / plugin_id / config_schema.json
|
|
3. PROJECT_ROOT / plugin-repos / plugin_id / config_schema.json
|
|
4. a case-insensitive match of plugin_id in plugins/ and plugin-repos/
|
|
|
|
A miss is remembered for SCHEMA_MISS_TTL seconds (or until
|
|
invalidate_cache()) and logged once, at DEBUG: PluginManager already
|
|
warns at load time about a plugin that ships no schema.
|
|
|
|
Args:
|
|
plugin_id: Plugin identifier
|
|
|
|
Returns:
|
|
Path to schema file or None if not found
|
|
"""
|
|
missed_at = self._schema_path_misses.get(plugin_id)
|
|
if missed_at is not None and time.monotonic() - missed_at < self.SCHEMA_MISS_TTL:
|
|
return None
|
|
|
|
search_dirs = []
|
|
if self.plugins_dir:
|
|
search_dirs.append(Path(self.plugins_dir))
|
|
search_dirs.extend([self.project_root / 'plugins',
|
|
self.project_root / 'plugin-repos'])
|
|
|
|
# Resolved the way the loader does, so a plugin installed as
|
|
# ``ledmatrix-<id>`` or under a directory named differently from its
|
|
# manifest id still gets its schema. One directory at a time keeps
|
|
# the documented plugins/-before-plugin-repos/ order.
|
|
possible_paths = []
|
|
for search_dir in search_dirs:
|
|
try:
|
|
resolved = resolve_plugin_dir(
|
|
plugin_id, [search_dir], prefix=True, case_insensitive=True)
|
|
except Exception as e: # pragma: no cover - defensive
|
|
self.logger.debug(f"Could not resolve {plugin_id} in {search_dir}: {e}")
|
|
resolved = None
|
|
if resolved is not None:
|
|
possible_paths.append(resolved / 'config_schema.json')
|
|
|
|
# Try plugins_dir if set
|
|
if self.plugins_dir:
|
|
possible_paths.append(self.plugins_dir / plugin_id / 'config_schema.json')
|
|
|
|
# Try standard locations relative to project root
|
|
possible_paths.extend([
|
|
self.project_root / 'plugins' / plugin_id / 'config_schema.json',
|
|
self.project_root / 'plugin-repos' / plugin_id / 'config_schema.json',
|
|
])
|
|
|
|
# Try case-insensitive directory matching
|
|
for base_dir in [self.project_root / 'plugins', self.project_root / 'plugin-repos']:
|
|
if base_dir.exists():
|
|
for item in base_dir.iterdir():
|
|
if item.is_dir() and item.name.lower() == plugin_id.lower():
|
|
possible_paths.append(item / 'config_schema.json')
|
|
|
|
# Try each path
|
|
for path in possible_paths:
|
|
if path.exists():
|
|
self.logger.debug(f"Found schema for {plugin_id} at {path}")
|
|
self._schema_path_misses.pop(plugin_id, None)
|
|
self._schema_miss_logged.discard(plugin_id)
|
|
return path
|
|
|
|
self._schema_path_misses[plugin_id] = time.monotonic()
|
|
if plugin_id not in self._schema_miss_logged:
|
|
self._schema_miss_logged.add(plugin_id)
|
|
self.logger.debug(f"Schema file not found for plugin {plugin_id}")
|
|
return None
|
|
|
|
def load_schema(self, plugin_id: str, use_cache: bool = True) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Load a plugin's configuration schema.
|
|
|
|
Args:
|
|
plugin_id: Plugin identifier
|
|
use_cache: If True, return cached schema if available
|
|
|
|
Returns:
|
|
Schema dictionary or None if not found
|
|
"""
|
|
# Check cache first
|
|
if use_cache and plugin_id in self._schema_cache:
|
|
return self._schema_cache[plugin_id]
|
|
|
|
schema_path = self.get_schema_path(plugin_id)
|
|
if not schema_path:
|
|
return None
|
|
|
|
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 any customization.x-style-elements declaration into the
|
|
# full per-element style blocks (font/size/color + layout
|
|
# offsets) the web-UI config form renders. No-op for schemas
|
|
# without the declaration; never raises.
|
|
schema = expand_style_elements(schema)
|
|
|
|
# Cache the schema
|
|
self._schema_cache[plugin_id] = schema
|
|
|
|
# Invalidate defaults cache when schema changes
|
|
if plugin_id in self._defaults_cache:
|
|
del self._defaults_cache[plugin_id]
|
|
|
|
return schema
|
|
|
|
except json.JSONDecodeError as e:
|
|
self.logger.error(f"Invalid JSON in schema file for {plugin_id}: {e}")
|
|
return None
|
|
except Exception as e:
|
|
self.logger.error(f"Error loading schema for {plugin_id}: {e}")
|
|
return None
|
|
|
|
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
|
|
"""
|
|
if plugin_id:
|
|
self._schema_cache.pop(plugin_id, None)
|
|
self._defaults_cache.pop(plugin_id, None)
|
|
self._schema_path_misses.pop(plugin_id, None)
|
|
self.logger.debug(f"Invalidated cache for plugin {plugin_id}")
|
|
else:
|
|
self._schema_cache.clear()
|
|
self._defaults_cache.clear()
|
|
self._schema_path_misses.clear()
|
|
self.logger.debug("Invalidated all schema caches")
|
|
|
|
def extract_defaults_from_schema(self, schema: Dict[str, Any], prefix: str = '') -> Dict[str, Any]:
|
|
"""
|
|
Recursively extract default values from a JSON Schema.
|
|
|
|
See :func:`extract_schema_defaults`; ``prefix`` is accepted for
|
|
compatibility and unused.
|
|
"""
|
|
return extract_schema_defaults(schema)
|
|
|
|
def get_device_location(self) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Return the device-wide ``location`` block from config.json, or None.
|
|
|
|
This is the City/State/Country the user sets once under General
|
|
settings. Returns None when there is no config manager wired, the
|
|
config can't be read, or no location has been configured.
|
|
"""
|
|
if self.config_manager is None:
|
|
return None
|
|
try:
|
|
config = self.config_manager.load_config()
|
|
except Exception as e:
|
|
# A config that can't be read must never stop defaults being
|
|
# generated -- the plugin's own schema defaults still apply.
|
|
self.logger.debug(f"Could not read device location from config: {e}")
|
|
return None
|
|
if not isinstance(config, dict):
|
|
return None
|
|
location = config.get('location')
|
|
return location if isinstance(location, dict) else None
|
|
|
|
def apply_device_location(self, defaults: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Replace location-shaped schema defaults with the device's own location.
|
|
|
|
Without this, a plugin that ships ``"location_city": "Dallas"`` as its
|
|
schema default silently reports Dallas weather (and centres its radar
|
|
there) for every user who never opened that plugin's config form --
|
|
even though they set their real city under General settings. The
|
|
substituted value is still only a *default*: ``merge_with_defaults``
|
|
lets any per-plugin value the user saved win over it.
|
|
|
|
Mutates and returns ``defaults`` for convenience.
|
|
"""
|
|
if not defaults:
|
|
return defaults
|
|
if not any(key in defaults for key in self.DEVICE_LOCATION_KEYS):
|
|
return defaults
|
|
|
|
location = self.get_device_location()
|
|
if not location:
|
|
return defaults
|
|
|
|
for key, field in self.DEVICE_LOCATION_KEYS.items():
|
|
if key not in defaults:
|
|
continue
|
|
value = location.get(field)
|
|
# Only a non-empty string is a real answer; a blank or missing
|
|
# field means "not configured", which leaves the schema default.
|
|
if isinstance(value, str) and value.strip():
|
|
defaults[key] = value.strip()
|
|
|
|
return defaults
|
|
|
|
def generate_default_config(self, plugin_id: str, use_cache: bool = True) -> Dict[str, Any]:
|
|
"""
|
|
Generate default configuration for a plugin from its schema.
|
|
|
|
Location fields (see ``DEVICE_LOCATION_KEYS``) default to the device's
|
|
configured location rather than the plugin author's. That substitution
|
|
is applied on the way out rather than being cached, so changing the
|
|
device location takes effect without invalidating the defaults cache.
|
|
|
|
Args:
|
|
plugin_id: Plugin identifier
|
|
use_cache: If True, return cached defaults if available
|
|
|
|
Returns:
|
|
Dictionary of default configuration values
|
|
"""
|
|
# Check cache first
|
|
if use_cache and plugin_id in self._defaults_cache:
|
|
return self.apply_device_location(self._defaults_cache[plugin_id].copy())
|
|
|
|
schema = self.load_schema(plugin_id, use_cache=use_cache)
|
|
if not schema:
|
|
# Return minimal defaults if no schema
|
|
return plugin_config_defaults(None)
|
|
|
|
# Schema defaults plus the core properties' (they may not be in the
|
|
# schema)
|
|
defaults = plugin_config_defaults(schema)
|
|
|
|
# Cache the defaults *before* the device location is layered on, so a
|
|
# later change to the device location is picked up by the next call.
|
|
self._defaults_cache[plugin_id] = defaults.copy()
|
|
|
|
return self.apply_device_location(defaults)
|
|
|
|
def prepare_plugin_config(self, plugin_id: str, config: Any,
|
|
schema: Optional[Dict[str, Any]] = None,
|
|
changed_paths: Optional[List[str]] = None) -> Dict[str, Any]:
|
|
"""
|
|
The config a plugin runs with: see :func:`prepare_plugin_config`.
|
|
|
|
Args:
|
|
plugin_id: Plugin identifier
|
|
config: The plugin's stored or submitted config section
|
|
schema: The plugin's schema, when the caller already has it
|
|
changed_paths: Receives the dotted path of each legacy boolean
|
|
read as an object
|
|
|
|
Returns:
|
|
A new dict; ``config`` is not mutated
|
|
"""
|
|
if schema is None:
|
|
schema = self.load_schema(plugin_id, use_cache=True)
|
|
defaults = self.generate_default_config(plugin_id, use_cache=True)
|
|
return prepare_plugin_config(config, schema, defaults, changed_paths)
|
|
|
|
def validate_config_against_schema(self, config: Dict[str, Any], schema: Dict[str, Any],
|
|
plugin_id: Optional[str] = None) -> Tuple[bool, List[str]]:
|
|
"""
|
|
Validate configuration against a JSON Schema.
|
|
|
|
Uses jsonschema library for comprehensive validation.
|
|
Automatically injects core plugin properties (enabled, display_duration, etc.)
|
|
into the schema before validation to ensure they're always allowed.
|
|
|
|
Args:
|
|
config: Configuration dictionary to validate
|
|
schema: JSON Schema dictionary
|
|
plugin_id: Optional plugin ID for error messages
|
|
|
|
Returns:
|
|
Tuple of (is_valid, list_of_error_messages)
|
|
"""
|
|
errors = []
|
|
|
|
try:
|
|
# Core plugin properties (CORE_PLUGIN_PROPERTIES) are handled by
|
|
# the base plugin system and should not cause validation failures:
|
|
# they are allowed even when the plugin's schema doesn't declare
|
|
# them, and never required. Retired ones are ignored.
|
|
config = drop_retired_plugin_keys(config, schema)
|
|
enhanced_schema = with_core_plugin_properties(schema)
|
|
if plugin_id:
|
|
declared = schema.get("properties", {}) if isinstance(schema, dict) else {}
|
|
self.logger.debug(
|
|
"Injected core properties into schema for %s: %s", plugin_id,
|
|
[name for name in CORE_PLUGIN_PROPERTIES if name not in declared]
|
|
)
|
|
|
|
# iter_errors reports every violation, including one ``required``
|
|
# error per missing field at every depth.
|
|
validator = Draft7Validator(enhanced_schema)
|
|
for error in validator.iter_errors(config):
|
|
errors.append(self._format_validation_error(error, plugin_id))
|
|
|
|
if errors:
|
|
return False, errors
|
|
|
|
return True, []
|
|
|
|
except jsonschema.SchemaError as e:
|
|
error_msg = f"Schema error{' for ' + plugin_id if plugin_id else ''}: {str(e)}"
|
|
self.logger.error(error_msg)
|
|
return False, [error_msg]
|
|
|
|
except Exception as e:
|
|
error_msg = f"Validation error{' for ' + plugin_id if plugin_id else ''}: {str(e)}"
|
|
self.logger.error(error_msg)
|
|
return False, [error_msg]
|
|
|
|
def _format_validation_error(self, error: ValidationError, plugin_id: Optional[str] = None) -> str:
|
|
"""
|
|
Format a validation error into a readable message.
|
|
|
|
Args:
|
|
error: ValidationError from jsonschema
|
|
plugin_id: Optional plugin ID for context
|
|
|
|
Returns:
|
|
Formatted error message
|
|
"""
|
|
path = '.'.join(str(p) for p in error.path)
|
|
field_path = f"'{path}'" if path else "root"
|
|
|
|
if error.validator == 'required':
|
|
# validator_value is the schema's whole ``required`` list; the
|
|
# error itself is about one field, which jsonschema names only in
|
|
# its message ("'api_key' is a required property").
|
|
missing = next(
|
|
(name for name in error.validator_value
|
|
if error.message.startswith(f"{name!r} ")),
|
|
None)
|
|
if missing is None:
|
|
return f"Field {field_path}: {error.message}"
|
|
return f"Field {field_path}: Missing required property '{missing}'"
|
|
elif error.validator == 'type':
|
|
expected = error.validator_value
|
|
actual = type(error.instance).__name__
|
|
return f"Field {field_path}: Expected type {expected}, got {actual}"
|
|
elif error.validator == 'enum':
|
|
allowed = error.validator_value
|
|
return f"Field {field_path}: Value '{error.instance}' not in allowed values {allowed}"
|
|
elif error.validator in ['minimum', 'maximum']:
|
|
limit = error.validator_value
|
|
return f"Field {field_path}: Value {error.instance} violates {error.validator} constraint ({limit})"
|
|
elif error.validator in ['minLength', 'maxLength']:
|
|
limit = error.validator_value
|
|
return f"Field {field_path}: Length {len(error.instance)} violates {error.validator} constraint ({limit})"
|
|
elif error.validator in ['minItems', 'maxItems']:
|
|
limit = error.validator_value
|
|
return f"Field {field_path}: Array length {len(error.instance)} violates {error.validator} constraint ({limit})"
|
|
else:
|
|
return f"Field {field_path}: {error.message}"
|
|
|
|
def merge_with_defaults(self, config: Dict[str, Any], defaults: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Merge configuration with defaults, preserving user values.
|
|
Also replaces None values with defaults to ensure config never has None from the start.
|
|
|
|
Args:
|
|
config: User configuration
|
|
defaults: Default values from schema
|
|
|
|
Returns:
|
|
Merged configuration with defaults applied where missing or None
|
|
"""
|
|
return merge_config_defaults(config, defaults)
|
|
|
|
def detect_config_key_collisions(
|
|
self,
|
|
plugin_ids: List[str]
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Detect config key collisions between plugins.
|
|
|
|
Checks for:
|
|
1. Plugin IDs that collide with reserved system config keys
|
|
2. Plugin IDs that might cause confusion or conflicts
|
|
|
|
Args:
|
|
plugin_ids: List of plugin identifiers to check
|
|
|
|
Returns:
|
|
List of collision warnings, each containing:
|
|
- type: 'reserved_key_collision' or 'case_collision'
|
|
- plugin_id: The plugin ID involved
|
|
- message: Human-readable warning message
|
|
"""
|
|
collisions = []
|
|
|
|
# Reserved top-level config keys that plugins should not use as IDs:
|
|
# every core section (src/core_config_keys.py), plus a few names that
|
|
# read as core even though no current section uses them.
|
|
reserved_keys = set(CORE_CONFIG_KEYS) | {
|
|
'display_modes', 'hardware', 'debug',
|
|
'log_level', 'emulator', 'web_interface'
|
|
}
|
|
|
|
# Track plugin IDs for case collision detection
|
|
lowercase_ids: Dict[str, str] = {}
|
|
|
|
for plugin_id in plugin_ids:
|
|
# Check reserved key collision
|
|
if plugin_id.lower() in {k.lower() for k in reserved_keys}:
|
|
collisions.append({
|
|
"type": "reserved_key_collision",
|
|
"plugin_id": plugin_id,
|
|
"message": f"Plugin ID '{plugin_id}' conflicts with reserved config key. "
|
|
f"This may cause configuration issues."
|
|
})
|
|
|
|
# Check for case-insensitive collisions between plugins
|
|
lower_id = plugin_id.lower()
|
|
if lower_id in lowercase_ids:
|
|
existing_id = lowercase_ids[lower_id]
|
|
if existing_id != plugin_id:
|
|
collisions.append({
|
|
"type": "case_collision",
|
|
"plugin_id": plugin_id,
|
|
"conflicting_id": existing_id,
|
|
"message": f"Plugin ID '{plugin_id}' may conflict with '{existing_id}' "
|
|
f"on case-insensitive file systems."
|
|
})
|
|
else:
|
|
lowercase_ids[lower_id] = plugin_id
|
|
|
|
return collisions
|
|
|