forked from genewildish/Mainline
David Gwilliam
6646ed78b3
- Detect REPL effect in pipeline and enable interactive mode - Enable raw terminal mode for REPL input capture - Add keyboard input loop for REPL commands (return, up/down arrows, backspace) - Process commands and handle pipeline mutations from REPL - Fix lint issues in graph and REPL modules (type annotations, imports)
283 lines
8.4 KiB
Python
283 lines
8.4 KiB
Python
"""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,
|
|
)
|