""" Pipeline presets - Pre-configured pipeline configurations. Provides PipelinePreset as a unified preset system. Presets can be loaded from TOML files (presets.toml) or defined in code. Loading order: 1. Built-in presets.toml in the package 2. User config ~/.config/mainline/presets.toml 3. Local ./presets.toml (overrides earlier) """ from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any from engine.display import BorderMode from engine.pipeline.params import PipelineParams if TYPE_CHECKING: from engine.pipeline.controller import PipelineConfig def _load_toml_presets() -> dict[str, Any]: """Load presets from TOML file.""" try: from engine.pipeline.preset_loader import load_presets return load_presets() except Exception: return {} _YAML_PRESETS = _load_toml_presets() @dataclass class PipelinePreset: """Pre-configured pipeline with stages and animation. A PipelinePreset packages: - Initial params: Starting configuration - Stages: List of stage configurations to create This is the new unified preset that works with the Pipeline class. """ name: str description: str = "" source: str = "headlines" display: str = "terminal" camera: str = "scroll" effects: list[str] = field(default_factory=list) border: bool | BorderMode = ( False # Border mode: False=off, True=simple, BorderMode.UI for panel ) # Extended fields for fine-tuning camera_speed: float = 1.0 # Camera movement speed viewport_width: int = 80 # Viewport width in columns viewport_height: int = 24 # Viewport height in rows source_items: list[dict[str, Any]] | None = None # For ListDataSource enable_metrics: bool = True # Enable performance metrics collection enable_message_overlay: bool = False # Enable ntfy message overlay def to_params(self) -> PipelineParams: """Convert to PipelineParams (runtime configuration).""" from engine.display import BorderMode params = PipelineParams() params.source = self.source params.display = self.display params.border = ( self.border if isinstance(self.border, bool) else BorderMode.UI if self.border == BorderMode.UI else False ) params.camera_mode = self.camera params.effect_order = self.effects.copy() params.camera_speed = self.camera_speed # Note: viewport_width/height are read from PipelinePreset directly # in pipeline_runner.py, not from PipelineParams return params def to_config(self) -> "PipelineConfig": """Convert to PipelineConfig (static pipeline construction config). PipelineConfig is used once at pipeline initialization and contains the core settings that don't change during execution. """ from engine.pipeline.controller import PipelineConfig return PipelineConfig( source=self.source, display=self.display, camera=self.camera, effects=self.effects.copy(), enable_metrics=self.enable_metrics, ) @classmethod def from_yaml(cls, name: str, data: dict[str, Any]) -> "PipelinePreset": """Create a PipelinePreset from YAML data.""" return cls( name=name, description=data.get("description", ""), source=data.get("source", "headlines"), display=data.get("display", "terminal"), camera=data.get("camera", "vertical"), effects=data.get("effects", []), border=data.get("border", False), camera_speed=data.get("camera_speed", 1.0), viewport_width=data.get("viewport_width", 80), viewport_height=data.get("viewport_height", 24), source_items=data.get("source_items"), enable_metrics=data.get("enable_metrics", True), enable_message_overlay=data.get("enable_message_overlay", False), ) # Built-in presets DEMO_PRESET = PipelinePreset( name="demo", description="Demo mode with effect cycling and camera modes", source="headlines", display="pygame", camera="scroll", effects=["noise", "fade", "glitch", "firehose"], enable_message_overlay=True, ) UI_PRESET = PipelinePreset( name="ui", description="Interactive UI mode with right-side control panel", source="fixture", display="pygame", camera="scroll", effects=["noise", "fade", "glitch"], border=BorderMode.UI, enable_message_overlay=True, ) POETRY_PRESET = PipelinePreset( name="poetry", description="Poetry feed with subtle effects", source="poetry", display="pygame", camera="scroll", effects=["fade"], ) PIPELINE_VIZ_PRESET = PipelinePreset( name="pipeline", description="Pipeline visualization mode", source="pipeline", display="terminal", camera="trace", effects=[], ) WEBSOCKET_PRESET = PipelinePreset( name="websocket", description="WebSocket display mode", source="headlines", display="websocket", camera="scroll", effects=["noise", "fade", "glitch"], ) FIREHOSE_PRESET = PipelinePreset( name="firehose", description="High-speed firehose mode", source="headlines", display="pygame", camera="scroll", effects=["noise", "fade", "glitch", "firehose"], enable_message_overlay=True, ) FIXTURE_PRESET = PipelinePreset( name="fixture", description="Use cached headline fixtures", source="fixture", display="pygame", camera="scroll", effects=["noise", "fade"], border=False, ) # Build presets from YAML data def _build_presets() -> dict[str, PipelinePreset]: """Build preset dictionary from all sources.""" result = {} # Add YAML presets yaml_presets = _YAML_PRESETS.get("presets", {}) for name, data in yaml_presets.items(): result[name] = PipelinePreset.from_yaml(name, data) # Add built-in presets as fallback (if not in YAML) builtins = { "demo": DEMO_PRESET, "poetry": POETRY_PRESET, "pipeline": PIPELINE_VIZ_PRESET, "websocket": WEBSOCKET_PRESET, "firehose": FIREHOSE_PRESET, "ui": UI_PRESET, "fixture": FIXTURE_PRESET, } for name, preset in builtins.items(): if name not in result: result[name] = preset return result PRESETS: dict[str, PipelinePreset] = _build_presets() def get_preset(name: str) -> PipelinePreset | None: """Get a preset by name.""" return PRESETS.get(name) def list_presets() -> list[str]: """List all available preset names.""" return list(PRESETS.keys()) def create_preset_from_params( params: PipelineParams, name: str = "custom" ) -> PipelinePreset: """Create a preset from PipelineParams.""" return PipelinePreset( name=name, source=params.source, display=params.display, camera=params.camera_mode, effects=params.effect_order.copy() if hasattr(params, "effect_order") else [], )