David Gwilliam
bcb4ef0cfe
- Add engine/pipeline/ module with Stage ABC, PipelineContext, PipelineParams - Stage provides unified interface for sources, effects, displays, cameras - Pipeline class handles DAG-based execution with dependency resolution - PipelinePreset for pre-configured pipelines (demo, poetry, pipeline, etc.) - Add PipelineParams as params layer for animation-driven config - Add StageRegistry for unified stage registration - Add sources_v2.py with DataSource.is_dynamic property - Add animation.py with Preset and AnimationController - Skip ntfy integration tests by default (require -m integration) - Skip e2e tests by default (require -m e2e) - Update pipeline.py with comprehensive introspection methods
156 lines
4.2 KiB
Python
156 lines
4.2 KiB
Python
"""
|
|
Pipeline presets - Pre-configured pipeline configurations.
|
|
|
|
Provides PipelinePreset as a unified preset system that wraps
|
|
the existing Preset class from animation.py for backwards compatibility.
|
|
"""
|
|
|
|
from dataclasses import dataclass, field
|
|
|
|
from engine.animation import Preset as AnimationPreset
|
|
from engine.pipeline.params import PipelineParams
|
|
|
|
|
|
@dataclass
|
|
class PipelinePreset:
|
|
"""Pre-configured pipeline with stages and animation.
|
|
|
|
A PipelinePreset packages:
|
|
- Initial params: Starting configuration
|
|
- Stages: List of stage configurations to create
|
|
- Animation: Optional animation controller
|
|
|
|
This is the new unified preset that works with the Pipeline class.
|
|
"""
|
|
|
|
name: str
|
|
description: str = ""
|
|
source: str = "headlines"
|
|
display: str = "terminal"
|
|
camera: str = "vertical"
|
|
effects: list[str] = field(default_factory=list)
|
|
initial_params: PipelineParams | None = None
|
|
animation_preset: AnimationPreset | None = None
|
|
|
|
def to_params(self) -> PipelineParams:
|
|
"""Convert to PipelineParams."""
|
|
if self.initial_params:
|
|
return self.initial_params.copy()
|
|
params = PipelineParams()
|
|
params.source = self.source
|
|
params.display = self.display
|
|
params.camera_mode = self.camera
|
|
params.effect_order = self.effects.copy()
|
|
return params
|
|
|
|
@classmethod
|
|
def from_animation_preset(cls, preset: AnimationPreset) -> "PipelinePreset":
|
|
"""Create a PipelinePreset from an existing animation Preset."""
|
|
params = preset.initial_params
|
|
return cls(
|
|
name=preset.name,
|
|
description=preset.description,
|
|
source=params.source,
|
|
display=params.display,
|
|
camera=params.camera_mode,
|
|
effects=params.effect_order.copy(),
|
|
initial_params=params,
|
|
animation_preset=preset,
|
|
)
|
|
|
|
def create_animation_controller(self):
|
|
"""Create an AnimationController from this preset."""
|
|
if self.animation_preset:
|
|
return self.animation_preset.create_controller()
|
|
return None
|
|
|
|
|
|
# Built-in presets
|
|
DEMO_PRESET = PipelinePreset(
|
|
name="demo",
|
|
description="Demo mode with effect cycling and camera modes",
|
|
source="headlines",
|
|
display="terminal",
|
|
camera="vertical",
|
|
effects=["noise", "fade", "glitch", "firehose", "hud"],
|
|
)
|
|
|
|
POETRY_PRESET = PipelinePreset(
|
|
name="poetry",
|
|
description="Poetry feed with subtle effects",
|
|
source="poetry",
|
|
display="terminal",
|
|
camera="vertical",
|
|
effects=["fade", "hud"],
|
|
)
|
|
|
|
PIPELINE_VIZ_PRESET = PipelinePreset(
|
|
name="pipeline",
|
|
description="Pipeline visualization mode",
|
|
source="pipeline",
|
|
display="terminal",
|
|
camera="trace",
|
|
effects=["hud"],
|
|
)
|
|
|
|
WEBSOCKET_PRESET = PipelinePreset(
|
|
name="websocket",
|
|
description="WebSocket display mode",
|
|
source="headlines",
|
|
display="websocket",
|
|
camera="vertical",
|
|
effects=["noise", "fade", "glitch", "hud"],
|
|
)
|
|
|
|
SIXEL_PRESET = PipelinePreset(
|
|
name="sixel",
|
|
description="Sixel graphics display mode",
|
|
source="headlines",
|
|
display="sixel",
|
|
camera="vertical",
|
|
effects=["noise", "fade", "glitch", "hud"],
|
|
)
|
|
|
|
FIREHOSE_PRESET = PipelinePreset(
|
|
name="firehose",
|
|
description="High-speed firehose mode",
|
|
source="headlines",
|
|
display="terminal",
|
|
camera="vertical",
|
|
effects=["noise", "fade", "glitch", "firehose", "hud"],
|
|
)
|
|
|
|
|
|
PRESETS: dict[str, PipelinePreset] = {
|
|
"demo": DEMO_PRESET,
|
|
"poetry": POETRY_PRESET,
|
|
"pipeline": PIPELINE_VIZ_PRESET,
|
|
"websocket": WEBSOCKET_PRESET,
|
|
"sixel": SIXEL_PRESET,
|
|
"firehose": FIREHOSE_PRESET,
|
|
}
|
|
|
|
|
|
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(),
|
|
initial_params=params,
|
|
)
|