David Gwilliam
0eb5f1d5ff
- Fixes issue #45: Add state property to EffectContext for motionblur/afterimage effects - Fixes issue #44: Reset camera bounce direction state in reset() method - Fixes issue #43: Implement pipeline hot-rebuild with state preservation - Adds radial camera mode for polar coordinate scanning - Adds afterimage and motionblur effects - Adds acceptance tests for camera and pipeline rebuild Closes #43, #44, #45
282 lines
8.9 KiB
Python
282 lines
8.9 KiB
Python
"""
|
|
Visual effects type definitions and base classes.
|
|
|
|
EffectPlugin Architecture:
|
|
- Uses ABC (Abstract Base Class) for interface enforcement
|
|
- Runtime discovery via directory scanning (effects_plugins/)
|
|
- Configuration via EffectConfig dataclass
|
|
- Context passed through EffectContext dataclass
|
|
|
|
Plugin System Research (see AGENTS.md for references):
|
|
- VST: Standardized audio interfaces, chaining, presets (FXP/FXB)
|
|
- Python Entry Points: Namespace packages, importlib.metadata discovery
|
|
- Shadertoy: Shader-based with uniforms as context
|
|
|
|
Current gaps vs industry patterns:
|
|
- No preset save/load system
|
|
- No external plugin distribution via entry points
|
|
- No plugin metadata (version, author, description)
|
|
"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
|
|
@dataclass
|
|
class PartialUpdate:
|
|
"""Represents a partial buffer update for optimized rendering.
|
|
|
|
Instead of processing the full buffer every frame, effects that support
|
|
partial updates can process only changed regions.
|
|
|
|
Attributes:
|
|
rows: Row indices that changed (None = all rows)
|
|
cols: Column range that changed (None = full width)
|
|
dirty: Set of dirty row indices
|
|
"""
|
|
|
|
rows: tuple[int, int] | None = None # (start, end) inclusive
|
|
cols: tuple[int, int] | None = None # (start, end) inclusive
|
|
dirty: set[int] | None = None # Set of dirty row indices
|
|
full_buffer: bool = True # If True, process entire buffer
|
|
|
|
|
|
@dataclass
|
|
class EffectContext:
|
|
"""Context passed to effect plugins during processing.
|
|
|
|
Contains terminal dimensions, camera state, frame info, and real-time sensor values.
|
|
"""
|
|
|
|
terminal_width: int
|
|
terminal_height: int
|
|
scroll_cam: int
|
|
ticker_height: int
|
|
camera_x: int = 0
|
|
mic_excess: float = 0.0
|
|
grad_offset: float = 0.0
|
|
frame_number: int = 0
|
|
has_message: bool = False
|
|
items: list = field(default_factory=list)
|
|
_state: dict[str, Any] = field(default_factory=dict, repr=False)
|
|
|
|
def compute_entropy(self, effect_name: str, data: Any) -> float:
|
|
"""Compute entropy score for an effect based on its output.
|
|
|
|
Args:
|
|
effect_name: Name of the effect
|
|
data: Processed buffer or effect-specific data
|
|
|
|
Returns:
|
|
Entropy score 0.0-1.0 representing visual chaos
|
|
"""
|
|
# Default implementation: use effect name as seed for deterministic randomness
|
|
# Better implementations can analyze actual buffer content
|
|
import hashlib
|
|
|
|
data_str = str(data)[:100] if data else ""
|
|
hash_val = hashlib.md5(f"{effect_name}:{data_str}".encode()).hexdigest()
|
|
# Convert hash to float 0.0-1.0
|
|
entropy = int(hash_val[:8], 16) / 0xFFFFFFFF
|
|
return min(max(entropy, 0.0), 1.0)
|
|
|
|
def get_sensor_value(self, sensor_name: str) -> float | None:
|
|
"""Get a sensor value from context state.
|
|
|
|
Args:
|
|
sensor_name: Name of the sensor (e.g., "mic", "camera")
|
|
|
|
Returns:
|
|
Sensor value as float, or None if not available.
|
|
"""
|
|
return self._state.get(f"sensor.{sensor_name}")
|
|
|
|
def set_state(self, key: str, value: Any) -> None:
|
|
"""Set a state value in the context."""
|
|
self._state[key] = value
|
|
|
|
def get_state(self, key: str, default: Any = None) -> Any:
|
|
"""Get a state value from the context."""
|
|
return self._state.get(key, default)
|
|
|
|
@property
|
|
def state(self) -> dict[str, Any]:
|
|
"""Get the state dictionary for direct access by effects."""
|
|
return self._state
|
|
|
|
|
|
@dataclass
|
|
class EffectConfig:
|
|
enabled: bool = True
|
|
intensity: float = 1.0
|
|
entropy: float = 0.0 # Visual chaos metric (0.0 = calm, 1.0 = chaotic)
|
|
params: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
class EffectPlugin(ABC):
|
|
"""Abstract base class for effect plugins.
|
|
|
|
Subclasses must define:
|
|
- name: str - unique identifier for the effect
|
|
- config: EffectConfig - current configuration
|
|
|
|
Optional class attribute:
|
|
- param_bindings: dict - Declarative sensor-to-param bindings
|
|
Example:
|
|
param_bindings = {
|
|
"intensity": {"sensor": "mic", "transform": "linear"},
|
|
"rate": {"sensor": "mic", "transform": "exponential"},
|
|
}
|
|
|
|
And implement:
|
|
- process(buf, ctx) -> list[str]
|
|
- configure(config) -> None
|
|
|
|
Effect Behavior with ticker_height=0:
|
|
- NoiseEffect: Returns buffer unchanged (no ticker to apply noise to)
|
|
- FadeEffect: Returns buffer unchanged (no ticker to fade)
|
|
- GlitchEffect: Processes normally (doesn't depend on ticker_height)
|
|
- FirehoseEffect: Returns buffer unchanged if no items in context
|
|
|
|
Effects should handle missing or zero context values gracefully by
|
|
returning the input buffer unchanged rather than raising errors.
|
|
|
|
The param_bindings system enables PureData-style signal routing:
|
|
- Sensors emit values that effects can bind to
|
|
- Transform functions map sensor values to param ranges
|
|
- Effects read bound values from context.state["sensor.{name}"]
|
|
"""
|
|
|
|
name: str
|
|
config: EffectConfig
|
|
param_bindings: dict[str, dict[str, str | float]] = {}
|
|
supports_partial_updates: bool = False # Override in subclasses for optimization
|
|
|
|
@abstractmethod
|
|
def process(self, buf: list[str], ctx: EffectContext) -> list[str]:
|
|
"""Process the buffer with this effect applied.
|
|
|
|
Args:
|
|
buf: List of lines to process
|
|
ctx: Effect context with terminal state
|
|
|
|
Returns:
|
|
Processed buffer (may be same object or new list)
|
|
"""
|
|
...
|
|
|
|
def process_partial(
|
|
self, buf: list[str], ctx: EffectContext, partial: PartialUpdate
|
|
) -> list[str]:
|
|
"""Process a partial buffer for optimized rendering.
|
|
|
|
Override this in subclasses that support partial updates for performance.
|
|
Default implementation falls back to full buffer processing.
|
|
|
|
Args:
|
|
buf: List of lines to process
|
|
ctx: Effect context with terminal state
|
|
partial: PartialUpdate indicating which regions changed
|
|
|
|
Returns:
|
|
Processed buffer (may be same object or new list)
|
|
"""
|
|
# Default: fall back to full processing
|
|
return self.process(buf, ctx)
|
|
|
|
@abstractmethod
|
|
def configure(self, config: EffectConfig) -> None:
|
|
"""Configure the effect with new settings.
|
|
|
|
Args:
|
|
config: New configuration to apply
|
|
"""
|
|
...
|
|
|
|
|
|
def create_effect_context(
|
|
terminal_width: int = 80,
|
|
terminal_height: int = 24,
|
|
scroll_cam: int = 0,
|
|
ticker_height: int = 0,
|
|
mic_excess: float = 0.0,
|
|
grad_offset: float = 0.0,
|
|
frame_number: int = 0,
|
|
has_message: bool = False,
|
|
items: list | None = None,
|
|
) -> EffectContext:
|
|
"""Factory function to create EffectContext with sensible defaults."""
|
|
return EffectContext(
|
|
terminal_width=terminal_width,
|
|
terminal_height=terminal_height,
|
|
scroll_cam=scroll_cam,
|
|
ticker_height=ticker_height,
|
|
mic_excess=mic_excess,
|
|
grad_offset=grad_offset,
|
|
frame_number=frame_number,
|
|
has_message=has_message,
|
|
items=items or [],
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class PipelineConfig:
|
|
order: list[str] = field(default_factory=list)
|
|
effects: dict[str, EffectConfig] = field(default_factory=dict)
|
|
|
|
|
|
def apply_param_bindings(
|
|
effect: "EffectPlugin",
|
|
ctx: EffectContext,
|
|
) -> EffectConfig:
|
|
"""Apply sensor bindings to effect config.
|
|
|
|
This resolves param_bindings declarations by reading sensor values
|
|
from the context and applying transform functions.
|
|
|
|
Args:
|
|
effect: The effect with param_bindings to apply
|
|
ctx: EffectContext containing sensor values
|
|
|
|
Returns:
|
|
Modified EffectConfig with sensor-driven values applied.
|
|
"""
|
|
import copy
|
|
|
|
if not effect.param_bindings:
|
|
return effect.config
|
|
|
|
config = copy.deepcopy(effect.config)
|
|
|
|
for param_name, binding in effect.param_bindings.items():
|
|
sensor_name: str = binding.get("sensor", "")
|
|
transform: str = binding.get("transform", "linear")
|
|
|
|
if not sensor_name:
|
|
continue
|
|
|
|
sensor_value = ctx.get_sensor_value(sensor_name)
|
|
if sensor_value is None:
|
|
continue
|
|
|
|
if transform == "linear":
|
|
applied_value: float = sensor_value
|
|
elif transform == "exponential":
|
|
applied_value = sensor_value**2
|
|
elif transform == "threshold":
|
|
threshold = float(binding.get("threshold", 0.5))
|
|
applied_value = 1.0 if sensor_value > threshold else 0.0
|
|
elif transform == "inverse":
|
|
applied_value = 1.0 - sensor_value
|
|
else:
|
|
applied_value = sensor_value
|
|
|
|
config.params[f"{param_name}_sensor"] = applied_value
|
|
|
|
if param_name == "intensity":
|
|
base_intensity = effect.config.intensity
|
|
config.intensity = base_intensity * (0.5 + applied_value * 0.5)
|
|
|
|
return config
|