""" 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