forked from genewildish/Mainline
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
341 lines
10 KiB
Python
341 lines
10 KiB
Python
"""
|
|
Animation system - Clock, events, triggers, durations, and animation controller.
|
|
"""
|
|
|
|
import time
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum, auto
|
|
from typing import Any
|
|
|
|
|
|
class Clock:
|
|
"""High-resolution clock for animation timing."""
|
|
|
|
def __init__(self):
|
|
self._start_time = time.perf_counter()
|
|
self._paused = False
|
|
self._pause_offset = 0.0
|
|
self._pause_start = 0.0
|
|
|
|
def reset(self) -> None:
|
|
self._start_time = time.perf_counter()
|
|
self._paused = False
|
|
self._pause_offset = 0.0
|
|
self._pause_start = 0.0
|
|
|
|
def elapsed(self) -> float:
|
|
if self._paused:
|
|
return self._pause_start - self._start_time - self._pause_offset
|
|
return time.perf_counter() - self._start_time - self._pause_offset
|
|
|
|
def elapsed_ms(self) -> float:
|
|
return self.elapsed() * 1000
|
|
|
|
def elapsed_frames(self, fps: float = 60.0) -> int:
|
|
return int(self.elapsed() * fps)
|
|
|
|
def pause(self) -> None:
|
|
if not self._paused:
|
|
self._paused = True
|
|
self._pause_start = time.perf_counter()
|
|
|
|
def resume(self) -> None:
|
|
if self._paused:
|
|
self._pause_offset += time.perf_counter() - self._pause_start
|
|
self._paused = False
|
|
|
|
|
|
class TriggerType(Enum):
|
|
TIME = auto() # Trigger after elapsed time
|
|
FRAME = auto() # Trigger after N frames
|
|
CYCLE = auto() # Trigger on cycle repeat
|
|
CONDITION = auto() # Trigger when condition is met
|
|
MANUAL = auto() # Trigger manually
|
|
|
|
|
|
@dataclass
|
|
class Trigger:
|
|
"""Event trigger configuration."""
|
|
|
|
type: TriggerType
|
|
value: float | int = 0
|
|
condition: Callable[["AnimationController"], bool] | None = None
|
|
repeat: bool = False
|
|
repeat_interval: float = 0.0
|
|
|
|
|
|
@dataclass
|
|
class Event:
|
|
"""An event with trigger, duration, and action."""
|
|
|
|
name: str
|
|
trigger: Trigger
|
|
action: Callable[["AnimationController", float], None]
|
|
duration: float = 0.0
|
|
ease: Callable[[float], float] | None = None
|
|
|
|
def __post_init__(self):
|
|
if self.ease is None:
|
|
self.ease = linear_ease
|
|
|
|
|
|
def linear_ease(t: float) -> float:
|
|
return t
|
|
|
|
|
|
def ease_in_out(t: float) -> float:
|
|
return t * t * (3 - 2 * t)
|
|
|
|
|
|
def ease_out_bounce(t: float) -> float:
|
|
if t < 1 / 2.75:
|
|
return 7.5625 * t * t
|
|
elif t < 2 / 2.75:
|
|
t -= 1.5 / 2.75
|
|
return 7.5625 * t * t + 0.75
|
|
elif t < 2.5 / 2.75:
|
|
t -= 2.25 / 2.75
|
|
return 7.5625 * t * t + 0.9375
|
|
else:
|
|
t -= 2.625 / 2.75
|
|
return 7.5625 * t * t + 0.984375
|
|
|
|
|
|
class AnimationController:
|
|
"""Controls animation parameters with clock and events."""
|
|
|
|
def __init__(self, fps: float = 60.0):
|
|
self.clock = Clock()
|
|
self.fps = fps
|
|
self.frame = 0
|
|
self._events: list[Event] = []
|
|
self._active_events: dict[str, float] = {}
|
|
self._params: dict[str, Any] = {}
|
|
self._cycled = 0
|
|
|
|
def add_event(self, event: Event) -> "AnimationController":
|
|
self._events.append(event)
|
|
return self
|
|
|
|
def set_param(self, key: str, value: Any) -> None:
|
|
self._params[key] = value
|
|
|
|
def get_param(self, key: str, default: Any = None) -> Any:
|
|
return self._params.get(key, default)
|
|
|
|
def update(self) -> dict[str, Any]:
|
|
"""Update animation state, return current params."""
|
|
elapsed = self.clock.elapsed()
|
|
|
|
for event in self._events:
|
|
triggered = False
|
|
|
|
if event.trigger.type == TriggerType.TIME:
|
|
if self.clock.elapsed() >= event.trigger.value:
|
|
triggered = True
|
|
elif event.trigger.type == TriggerType.FRAME:
|
|
if self.frame >= event.trigger.value:
|
|
triggered = True
|
|
elif event.trigger.type == TriggerType.CYCLE:
|
|
cycle_duration = event.trigger.value
|
|
if cycle_duration > 0:
|
|
current_cycle = int(elapsed / cycle_duration)
|
|
if current_cycle > self._cycled:
|
|
self._cycled = current_cycle
|
|
triggered = True
|
|
elif event.trigger.type == TriggerType.CONDITION:
|
|
if event.trigger.condition and event.trigger.condition(self):
|
|
triggered = True
|
|
elif event.trigger.type == TriggerType.MANUAL:
|
|
pass
|
|
|
|
if triggered:
|
|
if event.name not in self._active_events:
|
|
self._active_events[event.name] = 0.0
|
|
|
|
progress = 0.0
|
|
if event.duration > 0:
|
|
self._active_events[event.name] += 1 / self.fps
|
|
progress = min(
|
|
1.0, self._active_events[event.name] / event.duration
|
|
)
|
|
eased_progress = event.ease(progress)
|
|
event.action(self, eased_progress)
|
|
|
|
if progress >= 1.0:
|
|
if event.trigger.repeat:
|
|
self._active_events[event.name] = 0.0
|
|
else:
|
|
del self._active_events[event.name]
|
|
else:
|
|
event.action(self, 1.0)
|
|
if not event.trigger.repeat:
|
|
del self._active_events[event.name]
|
|
else:
|
|
self._active_events[event.name] = 0.0
|
|
|
|
self.frame += 1
|
|
return dict(self._params)
|
|
|
|
|
|
@dataclass
|
|
class PipelineParams:
|
|
"""Snapshot of pipeline parameters for animation."""
|
|
|
|
effect_enabled: dict[str, bool] = field(default_factory=dict)
|
|
effect_intensity: dict[str, float] = field(default_factory=dict)
|
|
camera_mode: str = "vertical"
|
|
camera_speed: float = 1.0
|
|
camera_x: int = 0
|
|
camera_y: int = 0
|
|
display_backend: str = "terminal"
|
|
scroll_speed: float = 1.0
|
|
|
|
|
|
class Preset:
|
|
"""Packages a starting pipeline config + Animation controller."""
|
|
|
|
def __init__(
|
|
self,
|
|
name: str,
|
|
description: str = "",
|
|
initial_params: PipelineParams | None = None,
|
|
animation: AnimationController | None = None,
|
|
):
|
|
self.name = name
|
|
self.description = description
|
|
self.initial_params = initial_params or PipelineParams()
|
|
self.animation = animation or AnimationController()
|
|
|
|
def create_controller(self) -> AnimationController:
|
|
controller = AnimationController()
|
|
for key, value in self.initial_params.__dict__.items():
|
|
controller.set_param(key, value)
|
|
for event in self.animation._events:
|
|
controller.add_event(event)
|
|
return controller
|
|
|
|
|
|
def create_demo_preset() -> Preset:
|
|
"""Create the demo preset with effect cycling and camera modes."""
|
|
animation = AnimationController(fps=60)
|
|
|
|
effects = ["noise", "fade", "glitch", "firehose"]
|
|
camera_modes = ["vertical", "horizontal", "omni", "floating", "trace"]
|
|
|
|
def make_effect_action(eff):
|
|
def action(ctrl, t):
|
|
ctrl.set_param("current_effect", eff)
|
|
ctrl.set_param("effect_intensity", t)
|
|
|
|
return action
|
|
|
|
def make_camera_action(cam_mode):
|
|
def action(ctrl, t):
|
|
ctrl.set_param("camera_mode", cam_mode)
|
|
|
|
return action
|
|
|
|
for i, effect in enumerate(effects):
|
|
effect_duration = 5.0
|
|
|
|
animation.add_event(
|
|
Event(
|
|
name=f"effect_{effect}",
|
|
trigger=Trigger(
|
|
type=TriggerType.TIME,
|
|
value=i * effect_duration,
|
|
repeat=True,
|
|
repeat_interval=len(effects) * effect_duration,
|
|
),
|
|
duration=effect_duration,
|
|
action=make_effect_action(effect),
|
|
ease=ease_in_out,
|
|
)
|
|
)
|
|
|
|
for i, mode in enumerate(camera_modes):
|
|
camera_duration = 10.0
|
|
animation.add_event(
|
|
Event(
|
|
name=f"camera_{mode}",
|
|
trigger=Trigger(
|
|
type=TriggerType.TIME,
|
|
value=i * camera_duration,
|
|
repeat=True,
|
|
repeat_interval=len(camera_modes) * camera_duration,
|
|
),
|
|
duration=0.5,
|
|
action=make_camera_action(mode),
|
|
)
|
|
)
|
|
|
|
animation.add_event(
|
|
Event(
|
|
name="pulse",
|
|
trigger=Trigger(type=TriggerType.CYCLE, value=2.0, repeat=True),
|
|
duration=1.0,
|
|
action=lambda ctrl, t: ctrl.set_param("pulse", t),
|
|
ease=ease_out_bounce,
|
|
)
|
|
)
|
|
|
|
return Preset(
|
|
name="demo",
|
|
description="Demo mode with effect cycling and camera modes",
|
|
initial_params=PipelineParams(
|
|
effect_enabled={
|
|
"noise": False,
|
|
"fade": False,
|
|
"glitch": False,
|
|
"firehose": False,
|
|
"hud": True,
|
|
},
|
|
effect_intensity={
|
|
"noise": 0.0,
|
|
"fade": 0.0,
|
|
"glitch": 0.0,
|
|
"firehose": 0.0,
|
|
},
|
|
camera_mode="vertical",
|
|
camera_speed=1.0,
|
|
display_backend="pygame",
|
|
),
|
|
animation=animation,
|
|
)
|
|
|
|
|
|
def create_pipeline_preset() -> Preset:
|
|
"""Create preset for pipeline visualization."""
|
|
animation = AnimationController(fps=60)
|
|
|
|
animation.add_event(
|
|
Event(
|
|
name="camera_trace",
|
|
trigger=Trigger(type=TriggerType.CYCLE, value=8.0, repeat=True),
|
|
duration=8.0,
|
|
action=lambda ctrl, t: ctrl.set_param("camera_mode", "trace"),
|
|
)
|
|
)
|
|
|
|
animation.add_event(
|
|
Event(
|
|
name="highlight_path",
|
|
trigger=Trigger(type=TriggerType.CYCLE, value=4.0, repeat=True),
|
|
duration=4.0,
|
|
action=lambda ctrl, t: ctrl.set_param("path_progress", t),
|
|
)
|
|
)
|
|
|
|
return Preset(
|
|
name="pipeline",
|
|
description="Pipeline visualization with trace camera",
|
|
initial_params=PipelineParams(
|
|
camera_mode="trace",
|
|
camera_speed=1.0,
|
|
display_backend="pygame",
|
|
),
|
|
animation=animation,
|
|
)
|