"""Hybrid Preset-Graph Configuration System This module provides a configuration format that combines the simplicity of presets with the flexibility of graphs. Example: [pipeline] source = "headlines" camera = { mode = "scroll", speed = 1.0 } effects = [ { name = "noise", intensity = 0.3 }, { name = "fade", intensity = 0.5 } ] display = { backend = "terminal" } This is much more concise than the verbose node-based graph DSL while providing the same flexibility. """ from dataclasses import dataclass, field from pathlib import Path from typing import Any from engine.pipeline.graph import Graph, NodeType from engine.pipeline.graph_adapter import graph_to_pipeline @dataclass class EffectConfig: """Configuration for a single effect.""" name: str intensity: float = 1.0 enabled: bool = True params: dict[str, Any] = field(default_factory=dict) @dataclass class CameraConfig: """Configuration for camera.""" mode: str = "scroll" speed: float = 1.0 @dataclass class DisplayConfig: """Configuration for display.""" backend: str = "terminal" positioning: str = "mixed" @dataclass class PipelineConfig: """Hybrid pipeline configuration combining preset simplicity with graph flexibility. This format provides a concise way to define pipelines that's 70% smaller than the verbose node-based DSL while maintaining full flexibility. Example: [pipeline] source = "headlines" camera = { mode = "scroll", speed = 1.0 } effects = [ { name = "noise", intensity = 0.3 }, { name = "fade", intensity = 0.5 } ] display = { backend = "terminal", positioning = "mixed" } """ source: str = "headlines" camera: CameraConfig | None = None effects: list[EffectConfig] = field(default_factory=list) display: DisplayConfig | None = None viewport_width: int = 80 viewport_height: int = 24 @classmethod def from_preset(cls, preset_name: str) -> "PipelineConfig": """Create PipelineConfig from a preset name. Args: preset_name: Name of preset (e.g., "upstream-default") Returns: PipelineConfig instance """ from engine.pipeline import get_preset preset = get_preset(preset_name) if not preset: raise ValueError(f"Preset '{preset_name}' not found") # Convert preset to PipelineConfig effects = [EffectConfig(name=e, intensity=1.0) for e in preset.effects] return cls( source=preset.source, camera=CameraConfig(mode=preset.camera, speed=preset.camera_speed), effects=effects, display=DisplayConfig( backend=preset.display, positioning=preset.positioning ), viewport_width=preset.viewport_width, viewport_height=preset.viewport_height, ) def to_graph(self) -> Graph: """Convert hybrid config to Graph representation.""" graph = Graph() # Add source node graph.node("source", NodeType.SOURCE, source=self.source) # Add camera node if configured if self.camera: graph.node( "camera", NodeType.CAMERA, mode=self.camera.mode, speed=self.camera.speed, ) # Add effect nodes for effect in self.effects: # Handle both EffectConfig objects and dictionaries if isinstance(effect, dict): name = effect.get("name", "") intensity = effect.get("intensity", 1.0) enabled = effect.get("enabled", True) params = effect.get("params", {}) else: name = effect.name intensity = effect.intensity enabled = effect.enabled params = effect.params if name: graph.node( name, NodeType.EFFECT, effect=name, intensity=intensity, enabled=enabled, **params, ) # Add display node if isinstance(self.display, dict): display_backend = self.display.get("backend", "terminal") display_positioning = self.display.get("positioning", "mixed") elif self.display: display_backend = self.display.backend display_positioning = self.display.positioning else: display_backend = "terminal" display_positioning = "mixed" graph.node( "display", NodeType.DISPLAY, backend=display_backend, positioning=display_positioning, ) # Create linear connections # Build chain: source -> camera -> effects... -> display chain = ["source"] if self.camera: chain.append("camera") # Add all effects in order for effect in self.effects: name = effect.get("name", "") if isinstance(effect, dict) else effect.name if name: chain.append(name) chain.append("display") # Connect all nodes in chain for i in range(len(chain) - 1): graph.connect(chain[i], chain[i + 1]) return graph def to_pipeline(self, viewport_width: int = 80, viewport_height: int = 24): """Convert to Pipeline instance.""" graph = self.to_graph() return graph_to_pipeline(graph, viewport_width, viewport_height) def load_hybrid_config(toml_path: str | Path) -> PipelineConfig: """Load hybrid configuration from TOML file. Args: toml_path: Path to TOML file Returns: PipelineConfig instance """ import tomllib with open(toml_path, "rb") as f: data = tomllib.load(f) return parse_hybrid_config(data) def parse_hybrid_config(data: dict[str, Any]) -> PipelineConfig: """Parse hybrid configuration from dictionary. Expected format: { "pipeline": { "source": "headlines", "camera": {"mode": "scroll", "speed": 1.0}, "effects": [ {"name": "noise", "intensity": 0.3}, {"name": "fade", "intensity": 0.5} ], "display": {"backend": "terminal"} } } """ pipeline_data = data.get("pipeline", {}) # Parse camera config camera = None if "camera" in pipeline_data: camera_data = pipeline_data["camera"] if isinstance(camera_data, dict): camera = CameraConfig( mode=camera_data.get("mode", "scroll"), speed=camera_data.get("speed", 1.0), ) elif isinstance(camera_data, str): camera = CameraConfig(mode=camera_data) # Parse effects list effects = [] if "effects" in pipeline_data: effects_data = pipeline_data["effects"] if isinstance(effects_data, list): for effect_item in effects_data: if isinstance(effect_item, dict): effects.append( EffectConfig( name=effect_item.get("name", ""), intensity=effect_item.get("intensity", 1.0), enabled=effect_item.get("enabled", True), params=effect_item.get("params", {}), ) ) elif isinstance(effect_item, str): effects.append(EffectConfig(name=effect_item)) # Parse display config display = None if "display" in pipeline_data: display_data = pipeline_data["display"] if isinstance(display_data, dict): display = DisplayConfig( backend=display_data.get("backend", "terminal"), positioning=display_data.get("positioning", "mixed"), ) elif isinstance(display_data, str): display = DisplayConfig(backend=display_data) # Parse viewport settings viewport_width = pipeline_data.get("viewport_width", 80) viewport_height = pipeline_data.get("viewport_height", 24) return PipelineConfig( source=pipeline_data.get("source", "headlines"), camera=camera, effects=effects, display=display, viewport_width=viewport_width, viewport_height=viewport_height, )