From fb0dd4592fc54619f8fc46d60ab871336545a769 Mon Sep 17 00:00:00 2001 From: David Gwilliam Date: Sat, 21 Mar 2026 21:12:38 -0700 Subject: [PATCH] feat(repl): Add REPL effect with HUD-style interactive interface Implement a Read-Eval-Print Loop (REPL) effect that provides a HUD-style overlay for interactive pipeline control. ## New Files - engine/effects/plugins/repl.py - REPL effect plugin with command processor - engine/display/backends/terminal.py - Added raw mode and input handling - examples/repl_simple.py - Simple demonstration script - tests/test_repl_effect.py - 18 comprehensive tests ## Features ### REPL Interface - HUD-style overlay showing FPS, command history, output buffer size - Command history navigation (Up/Down arrows) - Command execution (Enter) - Character input and backspace support - Output buffer with scrolling ### Commands - help - Show available commands - status - Show pipeline status and metrics - effects - List all effects in pipeline - effect - Toggle effect - param - Set parameter - pipeline - Show current pipeline order - clear - Clear output buffer - quit - Show exit message ### Terminal Input Support - Added set_raw_mode() to TerminalDisplay for capturing keystrokes - Added get_input_keys() to read keyboard input - Proper terminal state restoration on cleanup ## Usage Add 'repl' to effects in your configuration: ## Testing All 18 REPL tests pass, covering: - Effect registration - Command processing - Navigation (history, editing) - Configuration - Rendering ## Integration The REPL effect integrates with the existing pipeline system: - Uses EffectPlugin interface - Supports partial updates - Reads metrics from EffectContext - Can be controlled via keyboard when terminal display is in raw mode Next steps: - Integrate REPL input handling into pipeline_runner.py - Add keyboard event processing loop - Create full demo with interactive features --- engine/display/backends/terminal.py | 96 +++++++ engine/effects/plugins/repl.py | 419 ++++++++++++++++++++++++++++ engine/pipeline/hybrid_config.py | 48 +++- examples/repl_demo.py | 145 ++++++++++ examples/repl_simple.py | 78 ++++++ tests/test_repl_effect.py | 201 +++++++++++++ 6 files changed, 975 insertions(+), 12 deletions(-) create mode 100644 engine/effects/plugins/repl.py create mode 100644 examples/repl_demo.py create mode 100644 examples/repl_simple.py create mode 100644 tests/test_repl_effect.py diff --git a/engine/display/backends/terminal.py b/engine/display/backends/terminal.py index fb81dea..4bd575e 100644 --- a/engine/display/backends/terminal.py +++ b/engine/display/backends/terminal.py @@ -3,6 +3,10 @@ ANSI terminal display backend. """ import os +import select +import sys +import termios +import tty class TerminalDisplay: @@ -22,6 +26,8 @@ class TerminalDisplay: self._frame_period = 1.0 / target_fps if target_fps > 0 else 0 self._last_frame_time = 0.0 self._cached_dimensions: tuple[int, int] | None = None + self._raw_mode_enabled: bool = False + self._original_termios: list = [] def init(self, width: int, height: int, reuse: bool = False) -> None: """Initialize display with dimensions. @@ -150,6 +156,9 @@ class TerminalDisplay: def cleanup(self) -> None: from engine.terminal import CURSOR_ON + # Restore normal terminal mode if raw mode was enabled + self.set_raw_mode(False) + print(CURSOR_ON, end="", flush=True) def is_quit_requested(self) -> bool: @@ -159,3 +168,90 @@ class TerminalDisplay: def clear_quit_request(self) -> None: """Clear quit request (optional protocol method).""" pass + + def set_raw_mode(self, enable: bool = True) -> None: + """Enable/disable raw terminal mode for input capture. + + When raw mode is enabled: + - Keystrokes are read immediately without echo + - Special keys (arrows, Ctrl+C, etc.) are captured + - Terminal is not in cooked/canonical mode + + Args: + enable: True to enable raw mode, False to restore normal mode + """ + try: + if enable and not self._raw_mode_enabled: + # Save original terminal settings + self._original_termios = termios.tcgetattr(sys.stdin) + # Set raw mode + tty.setraw(sys.stdin.fileno()) + self._raw_mode_enabled = True + elif not enable and self._raw_mode_enabled: + # Restore original terminal settings + if self._original_termios: + termios.tcsetattr( + sys.stdin, termios.TCSADRAIN, self._original_termios + ) + self._raw_mode_enabled = False + except (termios.error, OSError): + # Terminal might not support raw mode (e.g., in tests) + pass + + def get_input_keys(self, timeout: float = 0.0) -> list[str]: + """Get available keyboard input. + + Reads available keystrokes from stdin. Should be called + with raw mode enabled for best results. + + Args: + timeout: Maximum time to wait for input (seconds) + + Returns: + List of key symbols as strings + """ + keys = [] + + try: + # Check if input is available + if select.select([sys.stdin], [], [], timeout)[0]: + char = sys.stdin.read(1) + + if char == "\x1b": # Escape sequence + # Read next character to determine key + seq = sys.stdin.read(2) + if seq == "[A": + keys.append("up") + elif seq == "[B": + keys.append("down") + elif seq == "[C": + keys.append("right") + elif seq == "[D": + keys.append("left") + else: + # Unknown escape sequence + keys.append("escape") + elif char == "\n" or char == "\r": + keys.append("return") + elif char == "\t": + keys.append("tab") + elif char == " ": + keys.append(" ") + elif char == "\x7f" or char == "\x08": # Backspace or Ctrl+H + keys.append("backspace") + elif char == "\x03": # Ctrl+C + keys.append("ctrl_c") + elif char == "\x04": # Ctrl+D + keys.append("ctrl_d") + elif char == "\x1b": # Escape + keys.append("escape") + elif char.isprintable(): + keys.append(char) + except (OSError, IOError): + pass + + return keys + + def is_raw_mode_enabled(self) -> bool: + """Check if raw mode is currently enabled.""" + return self._raw_mode_enabled diff --git a/engine/effects/plugins/repl.py b/engine/effects/plugins/repl.py new file mode 100644 index 0000000..99dad38 --- /dev/null +++ b/engine/effects/plugins/repl.py @@ -0,0 +1,419 @@ +"""REPL Effect Plugin + +A HUD-style command-line interface for interactive pipeline control. + +This effect provides a Read-Eval-Print Loop (REPL) that allows users to: +- View pipeline status and metrics +- Toggle effects on/off +- Adjust effect parameters in real-time +- Inspect pipeline configuration +- Execute commands for pipeline manipulation + +Usage: + Add 'repl' to the effects list in your configuration. + +Commands: + help - Show available commands + status - Show pipeline status + effects - List all effects + effect - Toggle an effect + param - Set effect parameter + pipeline - Show current pipeline order + clear - Clear output buffer + quit - Exit REPL + +Keyboard: + Enter - Execute command + Up/Down - Navigate command history + Tab - Auto-complete (if implemented) + Ctrl+C - Clear current input +""" + +from dataclasses import dataclass, field +from typing import Any, Optional + +from engine.effects.types import ( + EffectConfig, + EffectContext, + EffectPlugin, + PartialUpdate, +) + + +@dataclass +class REPLState: + """State of the REPL interface.""" + + command_history: list[str] = field(default_factory=list) + current_command: str = "" + history_index: int = -1 + output_buffer: list[str] = field(default_factory=list) + max_history: int = 50 + max_output_lines: int = 20 + + +class ReplEffect(EffectPlugin): + """REPL effect with HUD-style overlay for interactive pipeline control.""" + + name = "repl" + config = EffectConfig( + enabled=True, + intensity=1.0, + params={ + "display_height": 8, # Height of REPL area in lines + "show_hud": True, # Show HUD header lines + }, + ) + supports_partial_updates = True + + def __init__(self): + super().__init__() + self.state = REPLState() + self._last_metrics: Optional[dict] = None + + def process_partial( + self, buf: list[str], ctx: EffectContext, partial: PartialUpdate + ) -> list[str]: + """Handle partial updates efficiently.""" + if partial.full_buffer: + return self.process(buf, ctx) + # Always process REPL since it needs to stay visible + return self.process(buf, ctx) + + def process(self, buf: list[str], ctx: EffectContext) -> list[str]: + """Render buffer with REPL overlay.""" + result = list(buf) + + # Get display dimensions from context + height = ctx.terminal_height if hasattr(ctx, "terminal_height") else len(buf) + width = ctx.terminal_width if hasattr(ctx, "terminal_width") else 80 + + # Calculate areas + repl_height = self.config.params.get("display_height", 8) + show_hud = self.config.params.get("show_hud", True) + + # Reserve space for REPL at bottom + # HUD uses top 3 lines if enabled + hud_lines = 3 if show_hud else 0 + content_height = max(1, height - repl_height) + + # Build output + output = [] + + # Add content (truncated or padded) + for i in range(content_height): + if i < len(buf): + output.append(buf[i][:width]) + else: + output.append(" " * width) + + # Add HUD lines if enabled + if show_hud: + hud_output = self._render_hud(width, ctx) + # Overlay HUD on first lines of content + for i, line in enumerate(hud_output): + if i < len(output): + output[i] = line[:width] + + # Add separator + output.append("─" * width) + + # Add REPL area + repl_lines = self._render_repl(width, repl_height - 1) + output.extend(repl_lines) + + # Ensure correct height + while len(output) < height: + output.append(" " * width) + output = output[:height] + + return output + + def _render_hud(self, width: int, ctx: EffectContext) -> list[str]: + """Render HUD-style header with metrics.""" + lines = [] + + # Get metrics + metrics = self._get_metrics(ctx) + fps = metrics.get("fps", 0.0) + frame_time = metrics.get("frame_time", 0.0) + + # Line 1: Title + FPS + Frame time + fps_str = f"FPS: {fps:.1f}" if fps > 0 else "FPS: --" + time_str = f"{frame_time:.1f}ms" if frame_time > 0 else "--ms" + line1 = ( + f"\033[1;1H\033[38;5;46mMAINLINE REPL\033[0m " + f"\033[38;5;245m|\033[0m \033[38;5;39m{fps_str}\033[0m " + f"\033[38;5;245m|\033[0m \033[38;5;208m{time_str}\033[0m" + ) + lines.append(line1[:width]) + + # Line 2: Command count + History index + cmd_count = len(self.state.command_history) + hist_idx = ( + f"[{self.state.history_index + 1}/{cmd_count}]" if cmd_count > 0 else "" + ) + line2 = ( + f"\033[2;1H\033[38;5;45mCOMMANDS:\033[0m " + f"\033[1;38;5;227m{cmd_count}\033[0m " + f"\033[38;5;245m|\033[0m \033[38;5;219m{hist_idx}\033[0m" + ) + lines.append(line2[:width]) + + # Line 3: Output buffer count + out_count = len(self.state.output_buffer) + line3 = ( + f"\033[3;1H\033[38;5;44mOUTPUT:\033[0m " + f"\033[1;38;5;227m{out_count}\033[0m lines" + ) + lines.append(line3[:width]) + + return lines + + def _render_repl(self, width: int, height: int) -> list[str]: + """Render REPL interface.""" + lines = [] + + # Calculate how many output lines to show + # Reserve 1 line for input prompt + output_height = height - 1 + output_start = max(0, len(self.state.output_buffer) - output_height) + + # Render output buffer + for i in range(output_height): + idx = output_start + i + if idx < len(self.state.output_buffer): + line = self.state.output_buffer[idx][:width] + lines.append(line) + else: + lines.append(" " * width) + + # Render input prompt + prompt = "> " + input_line = f"{prompt}{self.state.current_command}" + # Add cursor indicator + cursor = "█" if len(self.state.current_command) % 2 == 0 else " " + input_line += cursor + lines.append(input_line[:width]) + + return lines + + def _get_metrics(self, ctx: EffectContext) -> dict: + """Get pipeline metrics from context.""" + metrics = ctx.get_state("metrics") + if metrics: + self._last_metrics = metrics + + if self._last_metrics: + # Extract FPS and frame time + fps = 0.0 + frame_time = 0.0 + + if "pipeline" in self._last_metrics: + avg_ms = self._last_metrics["pipeline"].get("avg_ms", 0.0) + frame_count = self._last_metrics.get("frame_count", 0) + if frame_count > 0 and avg_ms > 0: + fps = 1000.0 / avg_ms + frame_time = avg_ms + + return {"fps": fps, "frame_time": frame_time} + + return {"fps": 0.0, "frame_time": 0.0} + + def process_command( + self, command: str, ctx: Optional[EffectContext] = None + ) -> None: + """Process a REPL command.""" + cmd = command.strip() + if not cmd: + return + + # Add to history + self.state.command_history.append(cmd) + if len(self.state.command_history) > self.state.max_history: + self.state.command_history.pop(0) + + self.state.history_index = len(self.state.command_history) + self.state.current_command = "" + + # Add to output buffer + self.state.output_buffer.append(f"> {cmd}") + + # Parse command + parts = cmd.split() + cmd_name = parts[0].lower() + cmd_args = parts[1:] if len(parts) > 1 else [] + + # Execute command + try: + if cmd_name == "help": + self._cmd_help() + elif cmd_name == "status": + self._cmd_status(ctx) + elif cmd_name == "effects": + self._cmd_effects(ctx) + elif cmd_name == "effect": + self._cmd_effect(cmd_args, ctx) + elif cmd_name == "param": + self._cmd_param(cmd_args, ctx) + elif cmd_name == "pipeline": + self._cmd_pipeline(ctx) + elif cmd_name == "clear": + self.state.output_buffer.clear() + elif cmd_name == "quit" or cmd_name == "exit": + self.state.output_buffer.append("Use Ctrl+C to exit") + else: + self.state.output_buffer.append(f"Unknown command: {cmd_name}") + self.state.output_buffer.append("Type 'help' for available commands") + + except Exception as e: + self.state.output_buffer.append(f"Error: {e}") + + def _cmd_help(self): + """Show help message.""" + self.state.output_buffer.append("Available commands:") + self.state.output_buffer.append(" help - Show this help") + self.state.output_buffer.append(" status - Show pipeline status") + self.state.output_buffer.append(" effects - List all effects") + self.state.output_buffer.append(" effect - Toggle effect") + self.state.output_buffer.append( + " param - Set parameter" + ) + self.state.output_buffer.append(" pipeline - Show current pipeline order") + self.state.output_buffer.append(" clear - Clear output buffer") + self.state.output_buffer.append(" quit - Show exit message") + + def _cmd_status(self, ctx: Optional[EffectContext]): + """Show pipeline status.""" + if ctx: + metrics = self._get_metrics(ctx) + self.state.output_buffer.append(f"FPS: {metrics['fps']:.1f}") + self.state.output_buffer.append( + f"Frame time: {metrics['frame_time']:.1f}ms" + ) + + self.state.output_buffer.append( + f"Output lines: {len(self.state.output_buffer)}" + ) + self.state.output_buffer.append( + f"History: {len(self.state.command_history)} commands" + ) + + def _cmd_effects(self, ctx: Optional[EffectContext]): + """List all effects.""" + if ctx: + # Try to get effect list from context + effects = ctx.get_state("pipeline_order") + if effects: + self.state.output_buffer.append("Pipeline effects:") + for i, name in enumerate(effects): + self.state.output_buffer.append(f" {i + 1}. {name}") + else: + self.state.output_buffer.append("No pipeline information available") + else: + self.state.output_buffer.append("No context available") + + def _cmd_effect(self, args: list[str], ctx: Optional[EffectContext]): + """Toggle effect on/off.""" + if len(args) < 2: + self.state.output_buffer.append("Usage: effect ") + return + + effect_name = args[0] + state = args[1].lower() + + if state not in ("on", "off"): + self.state.output_buffer.append("State must be 'on' or 'off'") + return + + # Emit event to toggle effect + enabled = state == "on" + self.state.output_buffer.append(f"Effect '{effect_name}' set to {state}") + + # Store command for external handling + self._pending_command = { + "action": "enable_stage" if enabled else "disable_stage", + "stage": effect_name, + } + + def _cmd_param(self, args: list[str], ctx: Optional[EffectContext]): + """Set effect parameter.""" + if len(args) < 3: + self.state.output_buffer.append("Usage: param ") + return + + effect_name = args[0] + param_name = args[1] + try: + param_value = float(args[2]) + except ValueError: + self.state.output_buffer.append("Value must be a number") + return + + self.state.output_buffer.append( + f"Setting {effect_name}.{param_name} = {param_value}" + ) + + # Store command for external handling + self._pending_command = { + "action": "adjust_param", + "stage": effect_name, + "param": param_name, + "delta": param_value, # Note: This sets absolute value, need adjustment + } + + def _cmd_pipeline(self, ctx: Optional[EffectContext]): + """Show current pipeline order.""" + if ctx: + pipeline_order = ctx.get_state("pipeline_order") + if pipeline_order: + self.state.output_buffer.append( + "Pipeline: " + " → ".join(pipeline_order) + ) + else: + self.state.output_buffer.append("Pipeline information not available") + else: + self.state.output_buffer.append("No context available") + + def get_pending_command(self) -> Optional[dict]: + """Get and clear pending command for external handling.""" + cmd = getattr(self, "_pending_command", None) + if cmd: + self._pending_command = None + return cmd + + def navigate_history(self, direction: int) -> None: + """Navigate command history (up/down).""" + if not self.state.command_history: + return + + if direction > 0: # Down + self.state.history_index = min( + len(self.state.command_history), self.state.history_index + 1 + ) + else: # Up + self.state.history_index = max(0, self.state.history_index - 1) + + if self.state.history_index < len(self.state.command_history): + self.state.current_command = self.state.command_history[ + self.state.history_index + ] + else: + self.state.current_command = "" + + def append_to_command(self, char: str) -> None: + """Append character to current command.""" + if len(char) == 1: # Single character + self.state.current_command += char + + def backspace(self) -> None: + """Remove last character from command.""" + self.state.current_command = self.state.current_command[:-1] + + def clear_command(self) -> None: + """Clear current command.""" + self.state.current_command = "" + + def configure(self, config: EffectConfig) -> None: + """Configure the effect.""" + self.config = config diff --git a/engine/pipeline/hybrid_config.py b/engine/pipeline/hybrid_config.py index d426edc..e325558 100644 --- a/engine/pipeline/hybrid_config.py +++ b/engine/pipeline/hybrid_config.py @@ -124,22 +124,44 @@ class PipelineConfig: # Add effect nodes for effect in self.effects: - graph.node( - effect.name, - NodeType.EFFECT, - effect=effect.name, - intensity=effect.intensity, - enabled=effect.enabled, - **effect.params, - ) + # 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 - display_config = self.display or DisplayConfig() + 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_config.backend, - positioning=display_config.positioning, + backend=display_backend, + positioning=display_positioning, ) # Create linear connections @@ -151,7 +173,9 @@ class PipelineConfig: # Add all effects in order for effect in self.effects: - chain.append(effect.name) + name = effect.get("name", "") if isinstance(effect, dict) else effect.name + if name: + chain.append(name) chain.append("display") diff --git a/examples/repl_demo.py b/examples/repl_demo.py new file mode 100644 index 0000000..5d8b56d --- /dev/null +++ b/examples/repl_demo.py @@ -0,0 +1,145 @@ +#!/usr/bin/env python3 +""" +REPL Demo - Interactive command-line interface for pipeline control + +This demo shows how to use the REPL effect plugin to interact with +the Mainline pipeline in real-time. + +Features: +- HUD-style overlay showing FPS, frame time, command history +- Command history navigation (Up/Down arrows) +- Pipeline inspection and control commands +- Parameter adjustment in real-time + +Usage: + python examples/repl_demo.py + +Keyboard Controls: + Enter - Execute command + Up/Down - Navigate command history + Backspace - Delete character + Ctrl+C - Exit +""" + +import sys +import time +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from engine.effects.plugins import discover_plugins +from engine.pipeline.hybrid_config import PipelineConfig + + +def main(): + """Run the REPL demo.""" + print("REPL Demo - Interactive Pipeline Control") + print("=" * 50) + print() + print("This demo will:") + print("1. Create a pipeline with REPL effect") + print("2. Enable raw terminal mode for input") + print("3. Show REPL interface with HUD overlay") + print() + print("Keyboard controls:") + print(" Enter - Execute command") + print(" Up/Down - Navigate command history") + print(" Backspace - Delete character") + print(" Ctrl+C - Exit") + print() + print("Commands to try:") + print(" help - Show available commands") + print(" status - Show pipeline status") + print(" effects - List effects") + print(" pipeline - Show pipeline order") + print() + input("Press Enter to start...") + + # Discover plugins + discover_plugins() + + # Create pipeline with REPL effect + config = PipelineConfig( + source="headlines", + camera={"mode": "scroll", "speed": 1.0}, + effects=[ + {"name": "noise", "intensity": 0.3}, + {"name": "fade", "intensity": 0.5}, + {"name": "repl", "intensity": 1.0}, # Add REPL effect + ], + display={"backend": "terminal", "positioning": "mixed"}, + ) + + pipeline = config.to_pipeline(viewport_width=80, viewport_height=24) + + # Initialize pipeline + if not pipeline.initialize(): + print("Failed to initialize pipeline") + return + + # Get the REPL effect instance + repl_effect = None + for stage in pipeline._stages.values(): + if hasattr(stage, "_effect") and stage._effect.name == "repl": + repl_effect = stage._effect + break + + if not repl_effect: + print("REPL effect not found in pipeline") + return + + # Enable raw mode for input + display = pipeline.context.get("display") + if display and hasattr(display, "set_raw_mode"): + display.set_raw_mode(True) + + # Main loop + try: + frame_count = 0 + while True: + # Get keyboard input + if display and hasattr(display, "get_input_keys"): + keys = display.get_input_keys(timeout=0.01) + for key in keys: + if key == "return": + repl_effect.process_command( + repl_effect.state.current_command, pipeline.context + ) + elif key == "up": + repl_effect.navigate_history(-1) + elif key == "down": + repl_effect.navigate_history(1) + elif key == "backspace": + repl_effect.backspace() + elif key == "ctrl_c": + raise KeyboardInterrupt + elif len(key) == 1: + repl_effect.append_to_command(key) + + # Execute pipeline + result = pipeline.execute([]) + + if not result.success: + print(f"Pipeline error: {result.error}") + break + + # Check for pending commands + pending = repl_effect.get_pending_command() + if pending: + print(f"\nPending command: {pending}\n") + + frame_count += 1 + time.sleep(0.033) # ~30 FPS + + except KeyboardInterrupt: + print("\n\nExiting REPL demo...") + finally: + # Restore terminal mode + if display and hasattr(display, "set_raw_mode"): + display.set_raw_mode(False) + # Cleanup pipeline + pipeline.cleanup() + + +if __name__ == "__main__": + main() diff --git a/examples/repl_simple.py b/examples/repl_simple.py new file mode 100644 index 0000000..b9d293e --- /dev/null +++ b/examples/repl_simple.py @@ -0,0 +1,78 @@ +#!/usr/bin/env python3 +""" +Simple REPL Demo - Just shows the REPL effect rendering + +This is a simpler version that doesn't require raw terminal mode, +just demonstrates the REPL effect rendering. +""" + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from engine.effects.plugins import discover_plugins +from engine.effects.registry import get_registry +from engine.effects.types import EffectContext +from engine.pipeline.hybrid_config import PipelineConfig + + +def main(): + """Run simple REPL demo.""" + print("Simple REPL Demo") + print("=" * 50) + + # Discover plugins + discover_plugins() + + # Create a simple pipeline with REPL + config = PipelineConfig( + source="headlines", + effects=[{"name": "repl", "intensity": 1.0}], + display={"backend": "null"}, + ) + + pipeline = config.to_pipeline(viewport_width=80, viewport_height=24) + + # Initialize pipeline + if not pipeline.initialize(): + print("Failed to initialize pipeline") + return + + # Get the REPL effect + repl_effect = None + for stage in pipeline._stages.values(): + if hasattr(stage, "_effect") and stage._effect.name == "repl": + repl_effect = stage._effect + break + + if not repl_effect: + print("REPL effect not found") + return + + # Get the EffectContext for REPL + # Note: In a real pipeline, the EffectContext is created per-stage + # For this demo, we'll simulate by adding commands + + # Add some commands to the output + repl_effect.process_command("help") + repl_effect.process_command("status") + repl_effect.process_command("effects") + repl_effect.process_command("pipeline") + + # Execute pipeline to see REPL output + result = pipeline.execute([]) + + if result.success: + print("\nPipeline Output:") + print("-" * 50) + for line in result.data: + print(line) + print("-" * 50) + print(f"\n✓ Successfully rendered {len(result.data)} lines") + else: + print(f"✗ Pipeline error: {result.error}") + + +if __name__ == "__main__": + main() diff --git a/tests/test_repl_effect.py b/tests/test_repl_effect.py new file mode 100644 index 0000000..7f97811 --- /dev/null +++ b/tests/test_repl_effect.py @@ -0,0 +1,201 @@ +"""Tests for the REPL effect plugin.""" + +import pytest +from pathlib import Path + +from engine.effects.plugins import discover_plugins +from engine.effects.registry import get_registry +from engine.effects.plugins.repl import ReplEffect, REPLState + + +class TestReplEffectRegistration: + """Tests for REPL effect registration.""" + + def test_repl_registered(self): + """REPL effect is registered in the registry.""" + discover_plugins() + registry = get_registry() + repl = registry.get("repl") + assert repl is not None + assert repl.name == "repl" + + +class TestReplEffectCreation: + """Tests for creating REPL effect instances.""" + + def test_create_repl_effect(self): + """Can create REPL effect instance.""" + repl = ReplEffect() + assert repl.name == "repl" + assert repl.config.enabled is True + assert repl.config.intensity == 1.0 + + def test_repl_state(self): + """REPL state is initialized correctly.""" + repl = ReplEffect() + assert repl.state.command_history == [] + assert repl.state.current_command == "" + assert repl.state.history_index == -1 + assert repl.state.output_buffer == [] + + +class TestReplEffectCommands: + """Tests for REPL command processing.""" + + @pytest.fixture(autouse=True) + def setup(self): + """Setup before each test.""" + self.repl = ReplEffect() + + def test_process_command_help(self): + """Help command adds help text to output.""" + self.repl.process_command("help") + assert "> help" in self.repl.state.output_buffer + assert any( + "Available commands:" in line for line in self.repl.state.output_buffer + ) + + def test_process_command_status(self): + """Status command adds status info to output.""" + self.repl.process_command("status") + assert "> status" in self.repl.state.output_buffer + assert any("Output lines:" in line for line in self.repl.state.output_buffer) + + def test_process_command_clear(self): + """Clear command clears output buffer.""" + self.repl.process_command("help") + initial_count = len(self.repl.state.output_buffer) + assert initial_count > 0 + + self.repl.process_command("clear") + assert len(self.repl.state.output_buffer) == 0 + + def test_process_command_unknown(self): + """Unknown command adds error message.""" + self.repl.process_command("unknown_command_xyz") + assert "> unknown_command_xyz" in self.repl.state.output_buffer + assert any("Unknown command" in line for line in self.repl.state.output_buffer) + + def test_command_history(self): + """Commands are added to history.""" + self.repl.process_command("help") + self.repl.process_command("status") + assert len(self.repl.state.command_history) == 2 + assert self.repl.state.command_history[0] == "help" + assert self.repl.state.command_history[1] == "status" + + def test_current_command_cleared(self): + """Current command is cleared after processing.""" + self.repl.state.current_command = "test" + self.repl.process_command("help") + assert self.repl.state.current_command == "" + + +class TestReplNavigation: + """Tests for REPL navigation (history, editing).""" + + @pytest.fixture(autouse=True) + def setup(self): + """Setup before each test.""" + self.repl = ReplEffect() + self.repl.state.command_history = ["help", "status", "effects"] + + def test_navigate_history_up(self): + """Navigate up through command history.""" + self.repl.navigate_history(-1) # Up + assert self.repl.state.history_index == 0 + assert self.repl.state.current_command == "help" + + def test_navigate_history_down(self): + """Navigate down through command history.""" + self.repl.state.history_index = 0 + self.repl.navigate_history(1) # Down + assert self.repl.state.history_index == 1 + assert self.repl.state.current_command == "status" + + def test_append_to_command(self): + """Append character to current command.""" + self.repl.append_to_command("h") + self.repl.append_to_command("e") + self.repl.append_to_command("l") + self.repl.append_to_command("p") + assert self.repl.state.current_command == "help" + + def test_backspace(self): + """Remove last character from command.""" + self.repl.state.current_command = "hel" + self.repl.backspace() + assert self.repl.state.current_command == "he" + + def test_clear_command(self): + """Clear current command.""" + self.repl.state.current_command = "test" + self.repl.clear_command() + assert self.repl.state.current_command == "" + + +class TestReplProcess: + """Tests for REPL effect processing.""" + + @pytest.fixture(autouse=True) + def setup(self): + """Setup before each test.""" + discover_plugins() + self.repl = ReplEffect() + + def test_process_renders_output(self): + """Process renders REPL interface.""" + buf = ["line1", "line2", "line3"] + from engine.effects.types import EffectContext + + ctx = EffectContext( + terminal_width=80, terminal_height=24, scroll_cam=0, ticker_height=0 + ) + result = self.repl.process(buf, ctx) + + assert len(result) == 24 # Should match terminal height + assert any("MAINLINE REPL" in line for line in result) + assert any("COMMANDS:" in line for line in result) + assert any("OUTPUT:" in line for line in result) + + def test_process_with_commands(self): + """Process shows command output in REPL.""" + buf = ["line1"] + from engine.effects.types import EffectContext + + ctx = EffectContext( + terminal_width=80, terminal_height=24, scroll_cam=0, ticker_height=0 + ) + self.repl.process_command("help") + result = self.repl.process(buf, ctx) + + # Check that command output appears in the REPL area + # (help output may be partially shown due to buffer size limits) + assert any("effects - List all effects" in line for line in result) + + +class TestReplConfig: + """Tests for REPL configuration.""" + + def test_config_params(self): + """REPL config has expected parameters.""" + repl = ReplEffect() + assert "display_height" in repl.config.params + assert "show_hud" in repl.config.params + assert repl.config.params["display_height"] == 8 + assert repl.config.params["show_hud"] is True + + def test_configure(self): + """Can configure REPL effect.""" + repl = ReplEffect() + from engine.effects.types import EffectConfig + + config = EffectConfig( + enabled=False, + intensity=0.5, + params={"display_height": 10, "show_hud": False}, + ) + repl.configure(config) + assert repl.config.enabled is False + assert repl.config.intensity == 0.5 + assert repl.config.params["display_height"] == 10