mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-13 02:26:37 +00:00
ChuckandClaude Opus 5
a0e1ce6c6c
StartupValidator warned "Plugin 'auto_update' is enabled but not found" on every display start with auto-update or a dim schedule on; the reserved plugin-id check missed auto_update, sync, location and the rest; and ConfigManager's (uncalled) orphan cleanup would have deleted display, schedule and auto_update. All three now read src/core_config_keys.py, which also gains CORE_SECRETS_KEYS for the github/youtube secrets sections. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
741 lines
32 KiB
Python
741 lines
32 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
|
|
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
|
|
|
|
|
|
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
|
|
|
|
|
|
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 logging.getLogger(__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]] = {}
|
|
|
|
def get_schema_path(self, plugin_id: str) -> Optional[Path]:
|
|
"""
|
|
Get the path to a plugin's config_schema.json file.
|
|
|
|
Tries multiple locations in order:
|
|
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
|
|
|
|
Args:
|
|
plugin_id: Plugin identifier
|
|
|
|
Returns:
|
|
Path to schema file or None if not found
|
|
"""
|
|
possible_paths = []
|
|
|
|
# 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}")
|
|
return path
|
|
|
|
self.logger.warning(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.
|
|
try:
|
|
from src.element_style import expand_style_elements
|
|
schema = expand_style_elements(schema)
|
|
except ImportError:
|
|
pass
|
|
|
|
# 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.logger.debug(f"Invalidated cache for plugin {plugin_id}")
|
|
else:
|
|
self._schema_cache.clear()
|
|
self._defaults_cache.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.
|
|
|
|
Handles nested objects, arrays, and all schema types.
|
|
|
|
Args:
|
|
schema: JSON Schema dictionary
|
|
prefix: Optional prefix for logging/debugging
|
|
|
|
Returns:
|
|
Dictionary of default values
|
|
"""
|
|
defaults = {}
|
|
|
|
# Handle schema with properties
|
|
properties = schema.get('properties', {})
|
|
if not properties:
|
|
return defaults
|
|
|
|
for key, prop_schema in properties.items():
|
|
field_path = f"{prefix}.{key}" if prefix else key
|
|
|
|
# If property has a default, use it
|
|
if 'default' in prop_schema:
|
|
defaults[key] = prop_schema['default']
|
|
self.logger.debug(f"Found default for {field_path}: {prop_schema['default']}")
|
|
continue
|
|
|
|
# Handle nested objects
|
|
if prop_schema.get('type') == 'object' and 'properties' in prop_schema:
|
|
nested_defaults = self.extract_defaults_from_schema(prop_schema, field_path)
|
|
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 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 {
|
|
'enabled': False,
|
|
'display_duration': 15
|
|
}
|
|
|
|
# Extract defaults from schema
|
|
defaults = self.extract_defaults_from_schema(schema)
|
|
|
|
# Ensure core properties have defaults (they may not be in the schema)
|
|
# These match BasePlugin behavior
|
|
if 'enabled' not in defaults:
|
|
defaults['enabled'] = schema.get('properties', {}).get('enabled', {}).get('default', True)
|
|
|
|
if 'display_duration' not in defaults:
|
|
defaults['display_duration'] = schema.get('properties', {}).get('display_duration', {}).get('default', 15)
|
|
|
|
if 'live_priority' not in defaults:
|
|
defaults['live_priority'] = schema.get('properties', {}).get('live_priority', {}).get('default', False)
|
|
|
|
# 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 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 that should always be allowed
|
|
# These are handled by the base plugin system and should not cause validation failures
|
|
# Defaults match BasePlugin behavior: enabled=True, display_duration=15, live_priority=False
|
|
core_properties = {
|
|
"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"
|
|
},
|
|
# Skin selection (docs/SKIN_SYSTEM.md). Deliberately NOT an
|
|
# enum here: validation must keep passing when a configured
|
|
# skin gets uninstalled (rendering falls back to built-in).
|
|
# The install-dependent enum is injected only at serve time
|
|
# (inject_skin_selector) for the web UI dropdown.
|
|
"skin": {
|
|
"type": ["string", "object", "null"],
|
|
"description": "Visual skin id, or a per-mode mapping like {\"live\": \"my-skin\"}"
|
|
},
|
|
"skin_options": {
|
|
"type": "object",
|
|
"description": "Options passed through to the selected skin"
|
|
}
|
|
}
|
|
|
|
# Create a deep copy of the schema to modify (to avoid mutating the original)
|
|
enhanced_schema = copy.deepcopy(schema)
|
|
if "properties" not in enhanced_schema:
|
|
enhanced_schema["properties"] = {}
|
|
|
|
# Inject core properties if they're not already defined in the schema
|
|
# This ensures core properties are always allowed even if not in the plugin's schema
|
|
properties_added = []
|
|
for prop_name, prop_def in core_properties.items():
|
|
if prop_name not in enhanced_schema["properties"]:
|
|
enhanced_schema["properties"][prop_name] = copy.deepcopy(prop_def)
|
|
properties_added.append(prop_name)
|
|
|
|
# Log if we added any core properties (for debugging)
|
|
if properties_added and plugin_id:
|
|
self.logger.debug(
|
|
f"Injected core properties into schema for {plugin_id}: {properties_added}"
|
|
)
|
|
|
|
# Remove core properties from required array (they're system-managed)
|
|
# Core properties should be allowed but not required for validation
|
|
if "required" in enhanced_schema:
|
|
core_prop_names = list(core_properties.keys())
|
|
removed_from_required = [
|
|
field for field in enhanced_schema["required"]
|
|
if field in core_prop_names
|
|
]
|
|
enhanced_schema["required"] = [
|
|
field for field in enhanced_schema["required"]
|
|
if field not in core_prop_names
|
|
]
|
|
|
|
# Log if we removed any core properties from required (for debugging)
|
|
if removed_from_required and plugin_id:
|
|
self.logger.debug(
|
|
f"Removed core properties from required array for {plugin_id}: {removed_from_required}"
|
|
)
|
|
|
|
# Create validator with enhanced schema
|
|
validator = Draft7Validator(enhanced_schema)
|
|
|
|
# Collect all validation errors
|
|
for error in validator.iter_errors(config):
|
|
error_msg = self._format_validation_error(error, plugin_id)
|
|
errors.append(error_msg)
|
|
|
|
# Check required fields
|
|
required_fields = enhanced_schema.get('required', [])
|
|
for field in required_fields:
|
|
if field not in config:
|
|
errors.append(f"Missing required field: '{field}'")
|
|
|
|
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 inject_skin_selector(self, schema: Dict[str, Any], plugin_id: str,
|
|
current_value: Any = None) -> Dict[str, Any]:
|
|
"""Return a copy of a plugin's schema with a "skin" dropdown added
|
|
when installed skins target this plugin (docs/SKIN_SYSTEM.md).
|
|
|
|
Serve-time only — validation never sees this enum, so a config
|
|
referencing an uninstalled skin stays valid (rendering falls back
|
|
to the built-in layout). The currently-configured value is always
|
|
included in the enum for the same reason: the dropdown must be able
|
|
to display a selection whose skin was removed.
|
|
"""
|
|
# A per-mode mapping ({"live": ..., "recent": ...}) can't be edited
|
|
# through a string dropdown — injecting one would let the form save
|
|
# a string over the mapping. Leave the schema alone; per-mode users
|
|
# edit via the raw JSON config editor.
|
|
if isinstance(current_value, dict):
|
|
return schema
|
|
|
|
try:
|
|
from src.skin_system import skin_runtime
|
|
matching = skin_runtime.skins_for_plugin(plugin_id)
|
|
except Exception as e:
|
|
self.logger.debug(f"Skin discovery failed for {plugin_id}: {e}")
|
|
return schema
|
|
|
|
choices = sorted(matching.keys())
|
|
if isinstance(current_value, str) and current_value and \
|
|
current_value != "built-in" and current_value not in choices:
|
|
choices.append(current_value)
|
|
if not choices:
|
|
return schema
|
|
|
|
enhanced = copy.deepcopy(schema)
|
|
enhanced.setdefault("properties", {})
|
|
if "skin" not in enhanced["properties"]:
|
|
names = {sid: (matching.get(sid, {}).get("name") or sid) for sid in choices}
|
|
enhanced["properties"]["skin"] = {
|
|
"type": "string",
|
|
"title": "Visual Skin",
|
|
"description": "Replace this scoreboard's look with an installed skin "
|
|
"(data, scheduling, and vegas mode are unaffected)",
|
|
"enum": ["built-in", *choices],
|
|
"enumNames": ["Built-in", *(names[sid] for sid in choices)],
|
|
"default": "built-in"
|
|
}
|
|
return enhanced
|
|
|
|
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':
|
|
missing = error.validator_value
|
|
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
|
|
"""
|
|
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 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
|
|
|