diff --git a/engine/pipeline/preset_loader.py b/engine/pipeline/preset_loader.py new file mode 100644 index 0000000..cf9467c --- /dev/null +++ b/engine/pipeline/preset_loader.py @@ -0,0 +1,224 @@ +""" +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.""" + + pass + + +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_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) diff --git a/presets.toml b/presets.toml new file mode 100644 index 0000000..864e320 --- /dev/null +++ b/presets.toml @@ -0,0 +1,106 @@ +# Mainline Presets Configuration +# Human- and machine-readable preset definitions +# +# Format: TOML +# Usage: mainline --preset +# +# Built-in presets can be overridden by user presets in: +# - ~/.config/mainline/presets.toml +# - ./presets.toml (local override) + +[presets.demo] +description = "Demo mode with effect cycling and camera modes" +source = "headlines" +display = "pygame" +camera = "vertical" +effects = ["noise", "fade", "glitch", "firehose", "hud"] +viewport_width = 80 +viewport_height = 24 +camera_speed = 1.0 +firehose_enabled = true + +[presets.poetry] +description = "Poetry feed with subtle effects" +source = "poetry" +display = "pygame" +camera = "vertical" +effects = ["fade", "hud"] +viewport_width = 80 +viewport_height = 24 +camera_speed = 0.5 +firehose_enabled = false + +[presets.pipeline] +description = "Pipeline visualization mode" +source = "pipeline" +display = "terminal" +camera = "trace" +effects = ["hud"] +viewport_width = 80 +viewport_height = 24 +camera_speed = 1.0 +firehose_enabled = false + +[presets.websocket] +description = "WebSocket display mode" +source = "headlines" +display = "websocket" +camera = "vertical" +effects = ["noise", "fade", "glitch", "hud"] +viewport_width = 80 +viewport_height = 24 +camera_speed = 1.0 +firehose_enabled = false + +[presets.sixel] +description = "Sixel graphics display mode" +source = "headlines" +display = "sixel" +camera = "vertical" +effects = ["noise", "fade", "glitch", "hud"] +viewport_width = 80 +viewport_height = 24 +camera_speed = 1.0 +firehose_enabled = false + +[presets.firehose] +description = "High-speed firehose mode" +source = "headlines" +display = "pygame" +camera = "vertical" +effects = ["noise", "fade", "glitch", "firehose", "hud"] +viewport_width = 80 +viewport_height = 24 +camera_speed = 2.0 +firehose_enabled = true + +# Sensor configuration (for future use with param bindings) +[sensors.mic] +enabled = false +threshold_db = 50.0 + +[sensors.oscillator] +enabled = false +waveform = "sine" +frequency = 1.0 + +# Effect configurations +[effect_configs.noise] +enabled = true +intensity = 1.0 + +[effect_configs.fade] +enabled = true +intensity = 1.0 + +[effect_configs.glitch] +enabled = true +intensity = 0.5 + +[effect_configs.firehose] +enabled = true +intensity = 1.0 + +[effect_configs.hud] +enabled = true +intensity = 1.0