feat(app): add direct CLI mode, validation framework, fixtures, and UI panel integration
- Add run_pipeline_mode_direct() for constructing pipelines from CLI flags - Add engine/pipeline/validation.py with validate_pipeline_config() and MVP rules - Add fixtures system: engine/fixtures/headlines.json for cached test data - Enhance fetch.py to use fixtures cache path - Support fixture source in run_pipeline_mode() - Add --pipeline-* CLI flags: source, effects, camera, display, UI, border - Integrate UIPanel: raw mode, preset picker, event callbacks, param adjustment - Add UI_PRESET support in app and hot-rebuild pipeline on preset change - Add test UIPanel rendering and interaction tests This provides a flexible pipeline construction interface with validation and interactive control. Fixes #29, #30, #31
This commit is contained in:
549
engine/pipeline/ui.py
Normal file
549
engine/pipeline/ui.py
Normal file
@@ -0,0 +1,549 @@
|
||||
"""
|
||||
Pipeline UI panel - Interactive controls for pipeline configuration.
|
||||
|
||||
Provides:
|
||||
- Stage list with enable/disable toggles
|
||||
- Parameter sliders for selected effect
|
||||
- Keyboard/mouse interaction
|
||||
|
||||
This module implements the right-side UI panel that appears in border="ui" mode.
|
||||
"""
|
||||
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
|
||||
@dataclass
|
||||
class UIConfig:
|
||||
"""Configuration for the UI panel."""
|
||||
|
||||
panel_width: int = 24 # Characters wide
|
||||
stage_list_height: int = 12 # Number of stages to show at once
|
||||
param_height: int = 8 # Space for parameter controls
|
||||
scroll_offset: int = 0 # Scroll position in stage list
|
||||
start_with_preset_picker: bool = False # Show preset picker immediately
|
||||
|
||||
|
||||
@dataclass
|
||||
class StageControl:
|
||||
"""Represents a stage in the UI panel with its toggle state."""
|
||||
|
||||
name: str
|
||||
stage_name: str # Actual pipeline stage name
|
||||
category: str
|
||||
enabled: bool = True
|
||||
selected: bool = False
|
||||
params: dict[str, Any] = field(default_factory=dict) # Current param values
|
||||
param_schema: dict[str, dict] = field(default_factory=dict) # Param metadata
|
||||
|
||||
def toggle(self) -> None:
|
||||
"""Toggle enabled state."""
|
||||
self.enabled = not self.enabled
|
||||
|
||||
def get_param(self, name: str) -> Any:
|
||||
"""Get current parameter value."""
|
||||
return self.params.get(name)
|
||||
|
||||
def set_param(self, name: str, value: Any) -> None:
|
||||
"""Set parameter value."""
|
||||
self.params[name] = value
|
||||
|
||||
|
||||
class UIPanel:
|
||||
"""Interactive UI panel for pipeline configuration.
|
||||
|
||||
Manages:
|
||||
- Stage list with enable/disable checkboxes
|
||||
- Parameter sliders for selected stage
|
||||
- Keyboard/mouse event handling
|
||||
- Scroll state for long stage lists
|
||||
|
||||
The panel is rendered as a right border (panel_width characters wide)
|
||||
alongside the main viewport.
|
||||
"""
|
||||
|
||||
def __init__(self, config: UIConfig | None = None):
|
||||
self.config = config or UIConfig()
|
||||
self.stages: dict[str, StageControl] = {} # stage_name -> StageControl
|
||||
self.scroll_offset = 0
|
||||
self.selected_stage: str | None = None
|
||||
self._focused_param: str | None = None # For slider adjustment
|
||||
self._callbacks: dict[str, Callable] = {} # Event callbacks
|
||||
self._presets: list[str] = [] # Available preset names
|
||||
self._current_preset: str = "" # Current preset name
|
||||
self._show_preset_picker: bool = (
|
||||
config.start_with_preset_picker if config else False
|
||||
) # Picker overlay visible
|
||||
self._show_panel: bool = True # UI panel visibility
|
||||
self._preset_scroll_offset: int = 0 # Scroll in preset list
|
||||
|
||||
def register_stage(self, stage: Any, enabled: bool = True) -> StageControl:
|
||||
"""Register a stage for UI control.
|
||||
|
||||
Args:
|
||||
stage: Stage instance (must have .name, .category attributes)
|
||||
enabled: Initial enabled state
|
||||
|
||||
Returns:
|
||||
The created StageControl instance
|
||||
"""
|
||||
control = StageControl(
|
||||
name=stage.name,
|
||||
stage_name=stage.name,
|
||||
category=stage.category,
|
||||
enabled=enabled,
|
||||
)
|
||||
self.stages[stage.name] = control
|
||||
return control
|
||||
|
||||
def unregister_stage(self, stage_name: str) -> None:
|
||||
"""Remove a stage from UI control."""
|
||||
if stage_name in self.stages:
|
||||
del self.stages[stage_name]
|
||||
|
||||
def get_enabled_stages(self) -> list[str]:
|
||||
"""Get list of stage names that are currently enabled."""
|
||||
return [name for name, ctrl in self.stages.items() if ctrl.enabled]
|
||||
|
||||
def select_stage(self, stage_name: str | None = None) -> None:
|
||||
"""Select a stage (for parameter editing)."""
|
||||
if stage_name in self.stages:
|
||||
self.selected_stage = stage_name
|
||||
self.stages[stage_name].selected = True
|
||||
# Deselect others
|
||||
for name, ctrl in self.stages.items():
|
||||
if name != stage_name:
|
||||
ctrl.selected = False
|
||||
# Auto-focus first parameter when stage selected
|
||||
if self.stages[stage_name].params:
|
||||
self._focused_param = next(iter(self.stages[stage_name].params.keys()))
|
||||
else:
|
||||
self._focused_param = None
|
||||
|
||||
def toggle_stage(self, stage_name: str) -> bool:
|
||||
"""Toggle a stage's enabled state.
|
||||
|
||||
Returns:
|
||||
New enabled state
|
||||
"""
|
||||
if stage_name in self.stages:
|
||||
ctrl = self.stages[stage_name]
|
||||
ctrl.enabled = not ctrl.enabled
|
||||
return ctrl.enabled
|
||||
return False
|
||||
|
||||
def adjust_selected_param(self, delta: float) -> None:
|
||||
"""Adjust the currently focused parameter of selected stage.
|
||||
|
||||
Args:
|
||||
delta: Amount to add (positive or negative)
|
||||
"""
|
||||
if self.selected_stage and self._focused_param:
|
||||
ctrl = self.stages[self.selected_stage]
|
||||
if self._focused_param in ctrl.params:
|
||||
current = ctrl.params[self._focused_param]
|
||||
# Determine step size from schema
|
||||
schema = ctrl.param_schema.get(self._focused_param, {})
|
||||
step = schema.get("step", 0.1 if isinstance(current, float) else 1)
|
||||
new_val = current + delta * step
|
||||
# Clamp to min/max if specified
|
||||
if "min" in schema:
|
||||
new_val = max(schema["min"], new_val)
|
||||
if "max" in schema:
|
||||
new_val = min(schema["max"], new_val)
|
||||
# Only emit if value actually changed
|
||||
if new_val != current:
|
||||
ctrl.params[self._focused_param] = new_val
|
||||
self._emit_event(
|
||||
"param_changed",
|
||||
stage_name=self.selected_stage,
|
||||
param_name=self._focused_param,
|
||||
value=new_val,
|
||||
)
|
||||
|
||||
def scroll_stages(self, delta: int) -> None:
|
||||
"""Scroll the stage list."""
|
||||
max_offset = max(0, len(self.stages) - self.config.stage_list_height)
|
||||
self.scroll_offset = max(0, min(max_offset, self.scroll_offset + delta))
|
||||
|
||||
def render(self, width: int, height: int) -> list[str]:
|
||||
"""Render the UI panel.
|
||||
|
||||
Args:
|
||||
width: Total display width (panel uses last `panel_width` cols)
|
||||
height: Total display height
|
||||
|
||||
Returns:
|
||||
List of strings, each of length `panel_width`, to overlay on right side
|
||||
"""
|
||||
panel_width = min(
|
||||
self.config.panel_width, width - 4
|
||||
) # Reserve at least 2 for main
|
||||
lines = []
|
||||
|
||||
# If panel is hidden, render empty space
|
||||
if not self._show_panel:
|
||||
return [" " * panel_width for _ in range(height)]
|
||||
|
||||
# If preset picker is active, render that overlay instead of normal panel
|
||||
if self._show_preset_picker:
|
||||
picker_lines = self._render_preset_picker(panel_width)
|
||||
# Pad to full panel height if needed
|
||||
while len(picker_lines) < height:
|
||||
picker_lines.append(" " * panel_width)
|
||||
return [
|
||||
line.ljust(panel_width)[:panel_width] for line in picker_lines[:height]
|
||||
]
|
||||
|
||||
# Header
|
||||
title_line = "┌" + "─" * (panel_width - 2) + "┐"
|
||||
lines.append(title_line)
|
||||
|
||||
# Stage list section (occupies most of the panel)
|
||||
list_height = self.config.stage_list_height
|
||||
stage_names = list(self.stages.keys())
|
||||
for i in range(list_height):
|
||||
idx = i + self.scroll_offset
|
||||
if idx < len(stage_names):
|
||||
stage_name = stage_names[idx]
|
||||
ctrl = self.stages[stage_name]
|
||||
status = "✓" if ctrl.enabled else "✗"
|
||||
sel = ">" if ctrl.selected else " "
|
||||
# Truncate to fit panel (leave room for ">✓ " prefix and padding)
|
||||
max_name_len = panel_width - 5
|
||||
display_name = ctrl.name[:max_name_len]
|
||||
line = f"│{sel}{status} {display_name:<{max_name_len}}"
|
||||
lines.append(line[:panel_width])
|
||||
else:
|
||||
lines.append("│" + " " * (panel_width - 2) + "│")
|
||||
|
||||
# Separator
|
||||
lines.append("├" + "─" * (panel_width - 2) + "┤")
|
||||
|
||||
# Parameter section (if stage selected)
|
||||
if self.selected_stage and self.selected_stage in self.stages:
|
||||
ctrl = self.stages[self.selected_stage]
|
||||
if ctrl.params:
|
||||
# Render each parameter as "name: [=====] value" with focus indicator
|
||||
for param_name, param_value in ctrl.params.items():
|
||||
schema = ctrl.param_schema.get(param_name, {})
|
||||
is_focused = param_name == self._focused_param
|
||||
# Format value based on type
|
||||
if isinstance(param_value, float):
|
||||
val_str = f"{param_value:.2f}"
|
||||
elif isinstance(param_value, int):
|
||||
val_str = f"{param_value}"
|
||||
elif isinstance(param_value, bool):
|
||||
val_str = str(param_value)
|
||||
else:
|
||||
val_str = str(param_value)
|
||||
|
||||
# Build parameter line
|
||||
if (
|
||||
isinstance(param_value, (int, float))
|
||||
and "min" in schema
|
||||
and "max" in schema
|
||||
):
|
||||
# Render as slider
|
||||
min_val = schema["min"]
|
||||
max_val = schema["max"]
|
||||
# Normalize to 0-1 for bar length
|
||||
if max_val != min_val:
|
||||
ratio = (param_value - min_val) / (max_val - min_val)
|
||||
else:
|
||||
ratio = 0
|
||||
bar_width = (
|
||||
panel_width - len(param_name) - len(val_str) - 10
|
||||
) # approx space for "[] : ="
|
||||
if bar_width < 1:
|
||||
bar_width = 1
|
||||
filled = int(round(ratio * bar_width))
|
||||
bar = "[" + "=" * filled + " " * (bar_width - filled) + "]"
|
||||
param_line = f"│ {param_name}: {bar} {val_str}"
|
||||
else:
|
||||
# Simple name=value
|
||||
param_line = f"│ {param_name}={val_str}"
|
||||
|
||||
# Highlight focused parameter
|
||||
if is_focused:
|
||||
# Invert colors conceptually - for now use > prefix
|
||||
param_line = "│> " + param_line[2:]
|
||||
|
||||
# Truncate to fit panel width
|
||||
if len(param_line) > panel_width - 1:
|
||||
param_line = param_line[: panel_width - 1]
|
||||
lines.append(param_line + "│")
|
||||
else:
|
||||
lines.append("│ (no params)".ljust(panel_width - 1) + "│")
|
||||
else:
|
||||
lines.append("│ (select a stage)".ljust(panel_width - 1) + "│")
|
||||
|
||||
# Info line before footer
|
||||
info_parts = []
|
||||
if self._current_preset:
|
||||
info_parts.append(f"Preset: {self._current_preset}")
|
||||
if self._presets:
|
||||
info_parts.append("[P] presets")
|
||||
info_str = " | ".join(info_parts) if info_parts else ""
|
||||
if info_str:
|
||||
padded = info_str.ljust(panel_width - 2)
|
||||
lines.append("│" + padded + "│")
|
||||
|
||||
# Footer with instructions
|
||||
footer_line = self._render_footer(panel_width)
|
||||
lines.append(footer_line)
|
||||
|
||||
# Ensure all lines are exactly panel_width
|
||||
return [line.ljust(panel_width)[:panel_width] for line in lines]
|
||||
|
||||
def _render_footer(self, width: int) -> str:
|
||||
"""Render footer with key hints."""
|
||||
if width >= 40:
|
||||
# Show preset name and key hints
|
||||
preset_info = (
|
||||
f"Preset: {self._current_preset}" if self._current_preset else ""
|
||||
)
|
||||
hints = " [S]elect [Space]UI [Tab]Params [Arrows/HJKL]Adjust "
|
||||
if self._presets:
|
||||
hints += "[P]Preset "
|
||||
combined = f"{preset_info}{hints}"
|
||||
if len(combined) > width - 4:
|
||||
combined = combined[: width - 4]
|
||||
footer = "└" + "─" * (width - 2) + "┘"
|
||||
return footer # Just the line, we'll add info above in render
|
||||
else:
|
||||
return "└" + "─" * (width - 2) + "┘"
|
||||
|
||||
def process_key_event(self, key: str | int, modifiers: int = 0) -> bool:
|
||||
"""Process a keyboard event.
|
||||
|
||||
Args:
|
||||
key: Key symbol (e.g., ' ', 's', pygame.K_UP, etc.)
|
||||
modifiers: Modifier bits (Shift, Ctrl, Alt)
|
||||
|
||||
Returns:
|
||||
True if event was handled, False if not
|
||||
"""
|
||||
# Normalize to string for simplicity
|
||||
key_str = self._normalize_key(key, modifiers)
|
||||
|
||||
# Space: toggle UI panel visibility (only when preset picker not active)
|
||||
if key_str == " " and not self._show_preset_picker:
|
||||
self._show_panel = not getattr(self, "_show_panel", True)
|
||||
return True
|
||||
|
||||
# Space: toggle UI panel visibility (only when preset picker not active)
|
||||
if key_str == " " and not self._show_preset_picker:
|
||||
self._show_panel = not getattr(self, "_show_panel", True)
|
||||
return True
|
||||
|
||||
# S: select stage (cycle)
|
||||
if key_str == "s" and modifiers == 0:
|
||||
stages = list(self.stages.keys())
|
||||
if not stages:
|
||||
return False
|
||||
if self.selected_stage:
|
||||
current_idx = stages.index(self.selected_stage)
|
||||
next_idx = (current_idx + 1) % len(stages)
|
||||
else:
|
||||
next_idx = 0
|
||||
self.select_stage(stages[next_idx])
|
||||
return True
|
||||
|
||||
# P: toggle preset picker (only when panel is visible)
|
||||
if key_str == "p" and self._show_panel:
|
||||
self._show_preset_picker = not self._show_preset_picker
|
||||
if self._show_preset_picker:
|
||||
self._preset_scroll_offset = 0
|
||||
return True
|
||||
|
||||
# HJKL or Arrow Keys: scroll stage list, preset list, or adjust param
|
||||
# vi-style: K=up, J=down (J is actually next line in vi, but we use for down)
|
||||
# We'll use J for down, K for up, H for left, L for right
|
||||
elif key_str in ("up", "down", "kp8", "kp2", "j", "k"):
|
||||
# If preset picker is open, scroll preset list
|
||||
if self._show_preset_picker:
|
||||
delta = -1 if key_str in ("up", "kp8", "k") else 1
|
||||
self._preset_scroll_offset = max(0, self._preset_scroll_offset + delta)
|
||||
# Ensure scroll doesn't go past end
|
||||
max_offset = max(0, len(self._presets) - 1)
|
||||
self._preset_scroll_offset = min(max_offset, self._preset_scroll_offset)
|
||||
return True
|
||||
# If param is focused, adjust param value
|
||||
elif self.selected_stage and self._focused_param:
|
||||
delta = -1.0 if key_str in ("up", "kp8", "k") else 1.0
|
||||
self.adjust_selected_param(delta)
|
||||
return True
|
||||
# Otherwise scroll stages
|
||||
else:
|
||||
delta = -1 if key_str in ("up", "kp8", "k") else 1
|
||||
self.scroll_stages(delta)
|
||||
return True
|
||||
|
||||
# Left/Right or H/L: adjust param (if param selected)
|
||||
elif key_str in ("left", "right", "kp4", "kp6", "h", "l"):
|
||||
if self.selected_stage:
|
||||
delta = -0.1 if key_str in ("left", "kp4", "h") else 0.1
|
||||
self.adjust_selected_param(delta)
|
||||
return True
|
||||
|
||||
# Tab: cycle through parameters
|
||||
if key_str == "tab" and self.selected_stage:
|
||||
ctrl = self.stages[self.selected_stage]
|
||||
param_names = list(ctrl.params.keys())
|
||||
if param_names:
|
||||
if self._focused_param in param_names:
|
||||
current_idx = param_names.index(self._focused_param)
|
||||
next_idx = (current_idx + 1) % len(param_names)
|
||||
else:
|
||||
next_idx = 0
|
||||
self._focused_param = param_names[next_idx]
|
||||
return True
|
||||
|
||||
# Preset picker navigation
|
||||
if self._show_preset_picker:
|
||||
# Enter: select currently highlighted preset
|
||||
if key_str == "return":
|
||||
if self._presets:
|
||||
idx = self._preset_scroll_offset
|
||||
if idx < len(self._presets):
|
||||
self._current_preset = self._presets[idx]
|
||||
self._emit_event(
|
||||
"preset_changed", preset_name=self._current_preset
|
||||
)
|
||||
self._show_preset_picker = False
|
||||
return True
|
||||
# Escape: close picker without changing
|
||||
elif key_str == "escape":
|
||||
self._show_preset_picker = False
|
||||
return True
|
||||
|
||||
# Escape: deselect stage (only when picker not active)
|
||||
elif key_str == "escape" and self.selected_stage:
|
||||
self.selected_stage = None
|
||||
for ctrl in self.stages.values():
|
||||
ctrl.selected = False
|
||||
self._focused_param = None
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def _normalize_key(self, key: str | int, modifiers: int) -> str:
|
||||
"""Normalize key to a string identifier."""
|
||||
# Handle pygame keysyms if imported
|
||||
try:
|
||||
import pygame
|
||||
|
||||
if isinstance(key, int):
|
||||
# Map pygame constants to strings
|
||||
key_map = {
|
||||
pygame.K_UP: "up",
|
||||
pygame.K_DOWN: "down",
|
||||
pygame.K_LEFT: "left",
|
||||
pygame.K_RIGHT: "right",
|
||||
pygame.K_SPACE: " ",
|
||||
pygame.K_ESCAPE: "escape",
|
||||
pygame.K_s: "s",
|
||||
pygame.K_w: "w",
|
||||
# HJKL navigation (vi-style)
|
||||
pygame.K_h: "h",
|
||||
pygame.K_j: "j",
|
||||
pygame.K_k: "k",
|
||||
pygame.K_l: "l",
|
||||
}
|
||||
# Check for keypad keys with KP prefix
|
||||
if hasattr(pygame, "K_KP8") and key == pygame.K_KP8:
|
||||
return "kp8"
|
||||
if hasattr(pygame, "K_KP2") and key == pygame.K_KP2:
|
||||
return "kp2"
|
||||
if hasattr(pygame, "K_KP4") and key == pygame.K_KP4:
|
||||
return "kp4"
|
||||
if hasattr(pygame, "K_KP6") and key == pygame.K_KP6:
|
||||
return "kp6"
|
||||
return key_map.get(key, f"pygame_{key}")
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
# Already a string?
|
||||
if isinstance(key, str):
|
||||
return key.lower()
|
||||
|
||||
return str(key)
|
||||
|
||||
def set_event_callback(self, event_type: str, callback: Callable) -> None:
|
||||
"""Register a callback for UI events.
|
||||
|
||||
Args:
|
||||
event_type: Event type ("stage_toggled", "param_changed", "stage_selected", "preset_changed")
|
||||
callback: Function to call when event occurs
|
||||
"""
|
||||
self._callbacks[event_type] = callback
|
||||
|
||||
def _emit_event(self, event_type: str, **data) -> None:
|
||||
"""Emit an event to registered callbacks."""
|
||||
callback = self._callbacks.get(event_type)
|
||||
if callback:
|
||||
try:
|
||||
callback(**data)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def set_presets(self, presets: list[str], current: str) -> None:
|
||||
"""Set available presets and current selection.
|
||||
|
||||
Args:
|
||||
presets: List of preset names
|
||||
current: Currently active preset name
|
||||
"""
|
||||
self._presets = presets
|
||||
self._current_preset = current
|
||||
|
||||
def cycle_preset(self, direction: int = 1) -> str:
|
||||
"""Cycle to next/previous preset.
|
||||
|
||||
Args:
|
||||
direction: 1 for next, -1 for previous
|
||||
|
||||
Returns:
|
||||
New preset name
|
||||
"""
|
||||
if not self._presets:
|
||||
return self._current_preset
|
||||
try:
|
||||
current_idx = self._presets.index(self._current_preset)
|
||||
except ValueError:
|
||||
current_idx = 0
|
||||
next_idx = (current_idx + direction) % len(self._presets)
|
||||
self._current_preset = self._presets[next_idx]
|
||||
self._emit_event("preset_changed", preset_name=self._current_preset)
|
||||
return self._current_preset
|
||||
|
||||
def _render_preset_picker(self, panel_width: int) -> list[str]:
|
||||
"""Render a full-screen preset picker overlay."""
|
||||
lines = []
|
||||
picker_height = min(len(self._presets) + 2, self.config.stage_list_height)
|
||||
# Create a centered box
|
||||
title = " Select Preset "
|
||||
box_width = min(40, panel_width - 2)
|
||||
lines.append("┌" + "─" * (box_width - 2) + "┐")
|
||||
lines.append("│" + title.center(box_width - 2) + "│")
|
||||
lines.append("├" + "─" * (box_width - 2) + "┤")
|
||||
# List presets with selection
|
||||
visible_start = self._preset_scroll_offset
|
||||
visible_end = visible_start + picker_height - 2
|
||||
for i in range(visible_start, min(visible_end, len(self._presets))):
|
||||
preset_name = self._presets[i]
|
||||
is_current = preset_name == self._current_preset
|
||||
prefix = "▶ " if is_current else " "
|
||||
line = f"│ {prefix}{preset_name}"
|
||||
if len(line) < box_width - 1:
|
||||
line = line.ljust(box_width - 1)
|
||||
lines.append(line[: box_width - 1] + "│")
|
||||
# Footer with help
|
||||
help_text = "[P] close [↑↓] navigate [Enter] select"
|
||||
footer = "├" + "─" * (box_width - 2) + "┤"
|
||||
lines.append(footer)
|
||||
lines.append("│" + help_text.center(box_width - 2) + "│")
|
||||
lines.append("└" + "─" * (box_width - 2) + "┘")
|
||||
return lines
|
||||
Reference in New Issue
Block a user