feat(pipeline): integrate BorderMode and add UI preset

- params.py: border field now accepts bool | BorderMode
- presets.py: add UI_PRESET with BorderMode.UI, remove SIXEL_PRESET
- __init__.py: export UI_PRESET, drop SIXEL_PRESET
- registry.py: auto-register FrameBufferStage on discovery
- New FrameBufferStage for frame history and intensity maps
- Tests: update test_pipeline for UI preset, add test_framebuffer_stage.py

This sets the foundation for interactive UI panel and modern pipeline composition.
This commit is contained in:
2026-03-18 12:19:10 -07:00
parent 36afbacb6b
commit 21fb210c6e
7 changed files with 449 additions and 20 deletions

View File

@@ -0,0 +1,158 @@
"""
Frame buffer stage - stores previous frames for temporal effects.
Provides:
- frame_history: list of previous buffers (most recent first)
- intensity_history: list of corresponding intensity maps
- current_intensity: intensity map for current frame
Capability: "framebuffer.history"
"""
import threading
from dataclasses import dataclass
from typing import Any
from engine.display import _strip_ansi
from engine.pipeline.core import DataType, PipelineContext, Stage
@dataclass
class FrameBufferConfig:
"""Configuration for FrameBufferStage."""
history_depth: int = 2 # Number of previous frames to keep
class FrameBufferStage(Stage):
"""Stores frame history and computes intensity maps."""
name = "framebuffer"
category = "effect" # It's an effect that enriches context with frame history
def __init__(self, config: FrameBufferConfig | None = None, history_depth: int = 2):
self.config = config or FrameBufferConfig(history_depth=history_depth)
self._lock = threading.Lock()
@property
def capabilities(self) -> set[str]:
return {"framebuffer.history"}
@property
def dependencies(self) -> set[str]:
# Depends on rendered output (since we want to capture final buffer)
return {"render.output"}
@property
def inlet_types(self) -> set:
return {DataType.TEXT_BUFFER}
@property
def outlet_types(self) -> set:
return {DataType.TEXT_BUFFER} # Pass through unchanged
def init(self, ctx: PipelineContext) -> bool:
"""Initialize framebuffer state in context."""
ctx.set("frame_history", [])
ctx.set("intensity_history", [])
return True
def process(self, data: Any, ctx: PipelineContext) -> Any:
"""Store frame in history and compute intensity.
Args:
data: Current text buffer (list[str])
ctx: Pipeline context
Returns:
Same buffer (pass-through)
"""
if not isinstance(data, list):
return data
# Compute intensity map for current buffer (per-row, length = buffer rows)
intensity_map = self._compute_buffer_intensity(data, len(data))
# Store in context
ctx.set("current_intensity", intensity_map)
with self._lock:
# Get existing histories
history = ctx.get("frame_history", [])
intensity_hist = ctx.get("intensity_history", [])
# Prepend current frame to history
history.insert(0, data.copy())
intensity_hist.insert(0, intensity_map)
# Trim to configured depth
max_depth = self.config.history_depth
ctx.set("frame_history", history[:max_depth])
ctx.set("intensity_history", intensity_hist[:max_depth])
return data
def _compute_buffer_intensity(
self, buf: list[str], max_rows: int = 24
) -> list[float]:
"""Compute average intensity per row in buffer.
Uses ANSI color if available; falls back to character density.
Args:
buf: Text buffer (list of strings)
max_rows: Maximum number of rows to process
Returns:
List of intensity values (0.0-1.0) per row
"""
intensities = []
# Limit to viewport height
lines = buf[:max_rows]
for line in lines:
# Strip ANSI codes for length calc
plain = _strip_ansi(line)
if not plain:
intensities.append(0.0)
continue
# Simple heuristic: ratio of non-space characters
# More sophisticated version could parse ANSI RGB brightness
filled = sum(1 for c in plain if c not in (" ", "\t"))
total = len(plain)
intensity = filled / total if total > 0 else 0.0
intensities.append(max(0.0, min(1.0, intensity)))
# Pad to max_rows if needed
while len(intensities) < max_rows:
intensities.append(0.0)
return intensities
def get_frame(
self, index: int = 0, ctx: PipelineContext | None = None
) -> list[str] | None:
"""Get frame from history by index (0 = current, 1 = previous, etc)."""
if ctx is None:
return None
history = ctx.get("frame_history", [])
if 0 <= index < len(history):
return history[index]
return None
def get_intensity(
self, index: int = 0, ctx: PipelineContext | None = None
) -> list[float] | None:
"""Get intensity map from history by index."""
if ctx is None:
return None
intensity_hist = ctx.get("intensity_history", [])
if 0 <= index < len(intensity_hist):
return intensity_hist[index]
return None
def cleanup(self) -> None:
"""Cleanup resources."""
pass