""" Preset loader - Loads presets from TOML files. Supports: - Built-in presets.toml in the package - User overrides in ~/.config/mainline/presets.toml - Local override in ./presets.toml - Fallback DEFAULT_PRESET when loading fails """ import os from pathlib import Path from typing import Any import tomllib DEFAULT_PRESET: dict[str, Any] = { "description": "Default fallback preset", "source": "headlines", "display": "terminal", "camera": "vertical", "effects": ["hud"], "viewport": {"width": 80, "height": 24}, "camera_speed": 1.0, "firehose_enabled": False, } def get_preset_paths() -> list[Path]: """Get list of preset file paths in load order (later overrides earlier).""" paths = [] builtin = Path(__file__).parent.parent / "presets.toml" if builtin.exists(): paths.append(builtin) user_config = Path(os.path.expanduser("~/.config/mainline/presets.toml")) if user_config.exists(): paths.append(user_config) local = Path("presets.toml") if local.exists(): paths.append(local) return paths def load_presets() -> dict[str, Any]: """Load all presets, merging from multiple sources.""" merged: dict[str, Any] = {"presets": {}, "sensors": {}, "effect_configs": {}} for path in get_preset_paths(): try: with open(path, "rb") as f: data = tomllib.load(f) if "presets" in data: merged["presets"].update(data["presets"]) if "sensors" in data: merged["sensors"].update(data["sensors"]) if "effect_configs" in data: merged["effect_configs"].update(data["effect_configs"]) except Exception as e: print(f"Warning: Failed to load presets from {path}: {e}") return merged def get_preset(name: str) -> dict[str, Any] | None: """Get a preset by name.""" presets = load_presets() return presets.get("presets", {}).get(name) def list_preset_names() -> list[str]: """List all available preset names.""" presets = load_presets() return list(presets.get("presets", {}).keys()) def get_sensor_config(name: str) -> dict[str, Any] | None: """Get sensor configuration by name.""" sensors = load_presets() return sensors.get("sensors", {}).get(name) def get_effect_config(name: str) -> dict[str, Any] | None: """Get effect configuration by name.""" configs = load_presets() return configs.get("effect_configs", {}).get(name) def get_all_effect_configs() -> dict[str, Any]: """Get all effect configurations.""" configs = load_presets() return configs.get("effect_configs", {}) def get_preset_or_default(name: str) -> dict[str, Any]: """Get a preset by name, or return DEFAULT_PRESET if not found.""" preset = get_preset(name) if preset is not None: return preset return DEFAULT_PRESET.copy() def ensure_preset_available(name: str | None) -> dict[str, Any]: """Ensure a preset is available, falling back to DEFAULT_PRESET.""" if name is None: return DEFAULT_PRESET.copy() return get_preset_or_default(name) class PresetValidationError(Exception): """Raised when preset validation fails.""" def validate_preset(preset: dict[str, Any]) -> list[str]: """Validate a preset and return list of errors (empty if valid).""" errors: list[str] = [] required_fields = ["source", "display", "effects"] for field in required_fields: if field not in preset: errors.append(f"Missing required field: {field}") if "effects" in preset: if not isinstance(preset["effects"], list): errors.append("'effects' must be a list") else: for effect in preset["effects"]: if not isinstance(effect, str): errors.append( f"Effect must be string, got {type(effect)}: {effect}" ) if "viewport" in preset: viewport = preset["viewport"] if not isinstance(viewport, dict): errors.append("'viewport' must be a dict") else: if "width" in viewport and not isinstance(viewport["width"], int): errors.append("'viewport.width' must be an int") if "height" in viewport and not isinstance(viewport["height"], int): errors.append("'viewport.height' must be an int") return errors def validate_signal_flow(stages: list[dict]) -> list[str]: """Validate signal flow based on inlet/outlet types. This validates that the preset's stage configuration produces valid data flow using the PureData-style type system. Args: stages: List of stage configs with 'name', 'category', 'inlet_types', 'outlet_types' Returns: List of errors (empty if valid) """ errors: list[str] = [] if not stages: errors.append("Signal flow is empty") return errors # Define expected types for each category type_map = { "source": {"inlet": "NONE", "outlet": "SOURCE_ITEMS"}, "data": {"inlet": "ANY", "outlet": "SOURCE_ITEMS"}, "transform": {"inlet": "SOURCE_ITEMS", "outlet": "TEXT_BUFFER"}, "effect": {"inlet": "TEXT_BUFFER", "outlet": "TEXT_BUFFER"}, "overlay": {"inlet": "TEXT_BUFFER", "outlet": "TEXT_BUFFER"}, "camera": {"inlet": "TEXT_BUFFER", "outlet": "TEXT_BUFFER"}, "display": {"inlet": "TEXT_BUFFER", "outlet": "NONE"}, "render": {"inlet": "SOURCE_ITEMS", "outlet": "TEXT_BUFFER"}, } # Check stage order and type compatibility for i, stage in enumerate(stages): category = stage.get("category", "unknown") name = stage.get("name", f"stage_{i}") if category not in type_map: continue # Skip unknown categories expected = type_map[category] # Check against previous stage if i > 0: prev = stages[i - 1] prev_category = prev.get("category", "unknown") if prev_category in type_map: prev_outlet = type_map[prev_category]["outlet"] inlet = expected["inlet"] # Validate type compatibility if inlet != "ANY" and prev_outlet != "ANY" and inlet != prev_outlet: errors.append( f"Type mismatch at '{name}': " f"expects {inlet} but previous stage outputs {prev_outlet}" ) return errors def validate_signal_path(stages: list[str]) -> list[str]: """Validate signal path for circular dependencies and connectivity. Args: stages: List of stage names in execution order Returns: List of errors (empty if valid) """ errors: list[str] = [] if not stages: errors.append("Signal path is empty") return errors seen: set[str] = set() for i, stage in enumerate(stages): if stage in seen: errors.append( f"Circular dependency: '{stage}' appears multiple times at index {i}" ) seen.add(stage) return errors def generate_preset_toml( name: str, source: str = "headlines", display: str = "terminal", effects: list[str] | None = None, viewport_width: int = 80, viewport_height: int = 24, camera: str = "vertical", camera_speed: float = 1.0, firehose_enabled: bool = False, ) -> str: """Generate a TOML preset skeleton with default values. Args: name: Preset name source: Data source name display: Display backend effects: List of effect names viewport_width: Viewport width in columns viewport_height: Viewport height in rows camera: Camera mode camera_speed: Camera scroll speed firehose_enabled: Enable firehose mode Returns: TOML string for the preset """ if effects is None: effects = ["fade", "hud"] output = [] output.append(f"[presets.{name}]") output.append(f'description = "Auto-generated preset: {name}"') output.append(f'source = "{source}"') output.append(f'display = "{display}"') output.append(f'camera = "{camera}"') output.append(f"effects = {effects}") output.append(f"viewport_width = {viewport_width}") output.append(f"viewport_height = {viewport_height}") output.append(f"camera_speed = {camera_speed}") output.append(f"firehose_enabled = {str(firehose_enabled).lower()}") return "\n".join(output)