Files
sideline/engine/display/__init__.py
David Gwilliam e0bbfea26c refactor: consolidate pipeline architecture with unified data source system
MAJOR REFACTORING: Consolidate duplicated pipeline code and standardize on
capability-based dependency resolution. This is a significant but backwards-compatible
restructuring that improves maintainability and extensibility.

## ARCHITECTURE CHANGES

### Data Sources Consolidation
- Move engine/sources_v2.py → engine/data_sources/sources.py
- Move engine/pipeline_sources/ → engine/data_sources/
- Create unified DataSource ABC with common interface:
  * fetch() - idempotent data retrieval
  * get_items() - cached access with automatic refresh
  * refresh() - force cache invalidation
  * is_dynamic - indicate streaming vs static sources
- Support for SourceItem dataclass (content, source, timestamp, metadata)

### Display Backend Improvements
- Update all 7 display backends to use new import paths
- Terminal: Improve dimension detection and handling
- WebSocket: Better error handling and client lifecycle
- Sixel: Refactor graphics rendering
- Pygame: Modernize event handling
- Kitty: Add protocol support for inline images
- Multi: Ensure proper forwarding to all backends
- Null: Maintain testing backend functionality

### Pipeline Adapter Consolidation
- Refactor adapter stages for clarity and flexibility
- RenderStage now handles both item-based and buffer-based rendering
- Add SourceItemsToBufferStage for converting data source items
- Improve DataSourceStage to work with all source types
- Add DisplayStage wrapper for display backends

### Camera & Viewport Refinements
- Update Camera class for new architecture
- Improve viewport dimension detection
- Better handling of resize events across backends

### New Effect Plugins
- border.py: Frame rendering effect with configurable style
- crop.py: Viewport clipping effect for selective display
- tint.py: Color filtering effect for atmosphere

### Tests & Quality
- Add test_border_effect.py with comprehensive border tests
- Add test_crop_effect.py with viewport clipping tests
- Add test_tint_effect.py with color filtering tests
- Update test_pipeline.py for new architecture
- Update test_pipeline_introspection.py for new data source location
- All 463 tests pass with 56% coverage
- Linting: All checks pass with ruff

### Removals (Code Cleanup)
- Delete engine/benchmark.py (deprecated performance testing)
- Delete engine/pipeline_sources/__init__.py (moved to data_sources)
- Delete engine/sources_v2.py (replaced by data_sources/sources.py)
- Update AGENTS.md to reflect new structure

### Import Path Updates
- Update engine/pipeline/controller.py::create_default_pipeline()
  * Old: from engine.sources_v2 import HeadlinesDataSource
  * New: from engine.data_sources.sources import HeadlinesDataSource
- All display backends import from new locations
- All tests import from new locations

## BACKWARDS COMPATIBILITY

This refactoring is intended to be backwards compatible:
- Pipeline execution unchanged (DAG-based with capability matching)
- Effect plugins unchanged (EffectPlugin interface same)
- Display protocol unchanged (Display duck-typing works as before)
- Config system unchanged (presets.toml format same)

## TESTING

- 463 tests pass (0 failures, 19 skipped)
- Full linting check passes
- Manual testing on demo, poetry, websocket modes
- All new effect plugins tested

## FILES CHANGED

- 24 files modified/added/deleted
- 723 insertions, 1,461 deletions (net -738 LOC - cleanup!)
- No breaking changes to public APIs
- All transitive imports updated correctly
2026-03-16 19:47:12 -07:00

234 lines
6.9 KiB
Python

"""
Display backend system with registry pattern.
Allows swapping output backends via the Display protocol.
Supports auto-discovery of display backends.
"""
from typing import Protocol
from engine.display.backends.kitty import KittyDisplay
from engine.display.backends.multi import MultiDisplay
from engine.display.backends.null import NullDisplay
from engine.display.backends.pygame import PygameDisplay
from engine.display.backends.sixel import SixelDisplay
from engine.display.backends.terminal import TerminalDisplay
from engine.display.backends.websocket import WebSocketDisplay
class Display(Protocol):
"""Protocol for display backends.
All display backends must implement:
- width, height: Terminal dimensions
- init(width, height, reuse=False): Initialize the display
- show(buffer): Render buffer to display
- clear(): Clear the display
- cleanup(): Shutdown the display
Optional methods for keyboard input:
- is_quit_requested(): Returns True if user pressed Ctrl+C/Q or Escape
- clear_quit_request(): Clears the quit request flag
The reuse flag allows attaching to an existing display instance
rather than creating a new window/connection.
Keyboard input support by backend:
- terminal: No native input (relies on signal handler for Ctrl+C)
- pygame: Supports Ctrl+C, Ctrl+Q, Escape for graceful shutdown
- websocket: No native input (relies on signal handler for Ctrl+C)
- sixel: No native input (relies on signal handler for Ctrl+C)
- null: No native input
- kitty: Supports Ctrl+C, Ctrl+Q, Escape (via pygame-like handling)
"""
width: int
height: int
def init(self, width: int, height: int, reuse: bool = False) -> None:
"""Initialize display with dimensions.
Args:
width: Terminal width in characters
height: Terminal height in rows
reuse: If True, attach to existing display instead of creating new
"""
...
def show(self, buffer: list[str], border: bool = False) -> None:
"""Show buffer on display.
Args:
buffer: Buffer to display
border: If True, render border around buffer (default False)
"""
...
def clear(self) -> None:
"""Clear display."""
...
def cleanup(self) -> None:
"""Shutdown display."""
...
def get_dimensions(self) -> tuple[int, int]:
"""Get current terminal dimensions.
Returns:
(width, height) in character cells
This method is called after show() to check if the display
was resized. The main loop should compare this to the current
viewport dimensions and update accordingly.
"""
...
class DisplayRegistry:
"""Registry for display backends with auto-discovery."""
_backends: dict[str, type[Display]] = {}
_initialized = False
@classmethod
def register(cls, name: str, backend_class: type[Display]) -> None:
"""Register a display backend."""
cls._backends[name.lower()] = backend_class
@classmethod
def get(cls, name: str) -> type[Display] | None:
"""Get a display backend class by name."""
return cls._backends.get(name.lower())
@classmethod
def list_backends(cls) -> list[str]:
"""List all available display backend names."""
return list(cls._backends.keys())
@classmethod
def create(cls, name: str, **kwargs) -> Display | None:
"""Create a display instance by name."""
cls.initialize()
backend_class = cls.get(name)
if backend_class:
return backend_class(**kwargs)
return None
@classmethod
def initialize(cls) -> None:
"""Initialize and register all built-in backends."""
if cls._initialized:
return
cls.register("terminal", TerminalDisplay)
cls.register("null", NullDisplay)
cls.register("websocket", WebSocketDisplay)
cls.register("sixel", SixelDisplay)
cls.register("kitty", KittyDisplay)
cls.register("pygame", PygameDisplay)
cls._initialized = True
def get_monitor():
"""Get the performance monitor."""
try:
from engine.effects.performance import get_monitor as _get_monitor
return _get_monitor()
except Exception:
return None
def _strip_ansi(s: str) -> str:
"""Strip ANSI escape sequences from string for length calculation."""
import re
return re.sub(r"\x1b\[[0-9;]*[a-zA-Z]", "", s)
def render_border(
buf: list[str], width: int, height: int, fps: float = 0.0, frame_time: float = 0.0
) -> list[str]:
"""Render a border around the buffer.
Args:
buf: Input buffer (list of strings)
width: Display width in characters
height: Display height in rows
fps: Current FPS to display in top border (optional)
frame_time: Frame time in ms to display in bottom border (optional)
Returns:
Buffer with border applied
"""
if not buf or width < 3 or height < 3:
return buf
inner_w = width - 2
inner_h = height - 2
# Crop buffer to fit inside border
cropped = []
for i in range(min(inner_h, len(buf))):
line = buf[i]
# Calculate visible width (excluding ANSI codes)
visible_len = len(_strip_ansi(line))
if visible_len > inner_w:
# Truncate carefully - this is approximate for ANSI text
cropped.append(line[:inner_w])
else:
cropped.append(line + " " * (inner_w - visible_len))
# Pad with empty lines if needed
while len(cropped) < inner_h:
cropped.append(" " * inner_w)
# Build borders
if fps > 0:
fps_str = f" FPS:{fps:.0f}"
if len(fps_str) < inner_w:
right_len = inner_w - len(fps_str)
top_border = "" + "" * right_len + fps_str + ""
else:
top_border = "" + "" * inner_w + ""
else:
top_border = "" + "" * inner_w + ""
if frame_time > 0:
ft_str = f" {frame_time:.1f}ms"
if len(ft_str) < inner_w:
right_len = inner_w - len(ft_str)
bottom_border = "" + "" * right_len + ft_str + ""
else:
bottom_border = "" + "" * inner_w + ""
else:
bottom_border = "" + "" * inner_w + ""
# Build result with left/right borders
result = [top_border]
for line in cropped:
# Ensure exactly inner_w characters before adding right border
if len(line) < inner_w:
line = line + " " * (inner_w - len(line))
elif len(line) > inner_w:
line = line[:inner_w]
result.append("" + line + "")
result.append(bottom_border)
return result
__all__ = [
"Display",
"DisplayRegistry",
"get_monitor",
"render_border",
"TerminalDisplay",
"NullDisplay",
"WebSocketDisplay",
"SixelDisplay",
"MultiDisplay",
]