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
268 lines
7.9 KiB
Python
268 lines
7.9 KiB
Python
"""
|
|
Camera system for viewport scrolling.
|
|
|
|
Provides abstraction for camera motion in different modes:
|
|
- Vertical: traditional upward scroll
|
|
- Horizontal: left/right movement
|
|
- Omni: combination of both
|
|
- Floating: sinusoidal/bobbing motion
|
|
|
|
The camera defines a visible viewport into a larger Canvas.
|
|
"""
|
|
|
|
import math
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum, auto
|
|
|
|
|
|
class CameraMode(Enum):
|
|
VERTICAL = auto()
|
|
HORIZONTAL = auto()
|
|
OMNI = auto()
|
|
FLOATING = auto()
|
|
BOUNCE = auto()
|
|
|
|
|
|
@dataclass
|
|
class CameraViewport:
|
|
"""Represents the visible viewport."""
|
|
|
|
x: int
|
|
y: int
|
|
width: int
|
|
height: int
|
|
|
|
|
|
@dataclass
|
|
class Camera:
|
|
"""Camera for viewport scrolling.
|
|
|
|
The camera defines a visible viewport into a Canvas.
|
|
It can be smaller than the canvas to allow scrolling,
|
|
and supports zoom to scale the view.
|
|
|
|
Attributes:
|
|
x: Current horizontal offset (positive = scroll left)
|
|
y: Current vertical offset (positive = scroll up)
|
|
mode: Current camera mode
|
|
speed: Base scroll speed
|
|
zoom: Zoom factor (1.0 = 100%, 2.0 = 200% zoom out)
|
|
canvas_width: Width of the canvas being viewed
|
|
canvas_height: Height of the canvas being viewed
|
|
custom_update: Optional custom update function
|
|
"""
|
|
|
|
x: int = 0
|
|
y: int = 0
|
|
mode: CameraMode = CameraMode.VERTICAL
|
|
speed: float = 1.0
|
|
zoom: float = 1.0
|
|
canvas_width: int = 200 # Larger than viewport for scrolling
|
|
canvas_height: int = 200
|
|
custom_update: Callable[["Camera", float], None] | None = None
|
|
_time: float = field(default=0.0, repr=False)
|
|
|
|
@property
|
|
def w(self) -> int:
|
|
"""Shorthand for viewport_width."""
|
|
return self.viewport_width
|
|
|
|
@property
|
|
def h(self) -> int:
|
|
"""Shorthand for viewport_height."""
|
|
return self.viewport_height
|
|
|
|
@property
|
|
def viewport_width(self) -> int:
|
|
"""Get the visible viewport width.
|
|
|
|
This is the canvas width divided by zoom.
|
|
"""
|
|
return max(1, int(self.canvas_width / self.zoom))
|
|
|
|
@property
|
|
def viewport_height(self) -> int:
|
|
"""Get the visible viewport height.
|
|
|
|
This is the canvas height divided by zoom.
|
|
"""
|
|
return max(1, int(self.canvas_height / self.zoom))
|
|
|
|
def get_viewport(self) -> CameraViewport:
|
|
"""Get the current viewport bounds.
|
|
|
|
Returns:
|
|
CameraViewport with position and size (clamped to canvas bounds)
|
|
"""
|
|
vw = self.viewport_width
|
|
vh = self.viewport_height
|
|
|
|
clamped_x = max(0, min(self.x, self.canvas_width - vw))
|
|
clamped_y = max(0, min(self.y, self.canvas_height - vh))
|
|
|
|
return CameraViewport(
|
|
x=clamped_x,
|
|
y=clamped_y,
|
|
width=vw,
|
|
height=vh,
|
|
)
|
|
|
|
def set_zoom(self, zoom: float) -> None:
|
|
"""Set the zoom factor.
|
|
|
|
Args:
|
|
zoom: Zoom factor (1.0 = 100%, 2.0 = zoomed out 2x, 0.5 = zoomed in 2x)
|
|
"""
|
|
self.zoom = max(0.1, min(10.0, zoom))
|
|
|
|
def update(self, dt: float) -> None:
|
|
"""Update camera position based on mode.
|
|
|
|
Args:
|
|
dt: Delta time in seconds
|
|
"""
|
|
self._time += dt
|
|
|
|
if self.custom_update:
|
|
self.custom_update(self, dt)
|
|
return
|
|
|
|
if self.mode == CameraMode.VERTICAL:
|
|
self._update_vertical(dt)
|
|
elif self.mode == CameraMode.HORIZONTAL:
|
|
self._update_horizontal(dt)
|
|
elif self.mode == CameraMode.OMNI:
|
|
self._update_omni(dt)
|
|
elif self.mode == CameraMode.FLOATING:
|
|
self._update_floating(dt)
|
|
elif self.mode == CameraMode.BOUNCE:
|
|
self._update_bounce(dt)
|
|
|
|
# Bounce mode handles its own bounds checking
|
|
if self.mode != CameraMode.BOUNCE:
|
|
self._clamp_to_bounds()
|
|
|
|
def _clamp_to_bounds(self) -> None:
|
|
"""Clamp camera position to stay within canvas bounds.
|
|
|
|
Only clamps if the viewport is smaller than the canvas.
|
|
If viewport equals canvas (no scrolling needed), allows any position
|
|
for backwards compatibility with original behavior.
|
|
"""
|
|
vw = self.viewport_width
|
|
vh = self.viewport_height
|
|
|
|
# Only clamp if there's room to scroll
|
|
if vw < self.canvas_width:
|
|
self.x = max(0, min(self.x, self.canvas_width - vw))
|
|
if vh < self.canvas_height:
|
|
self.y = max(0, min(self.y, self.canvas_height - vh))
|
|
|
|
def _update_vertical(self, dt: float) -> None:
|
|
self.y += int(self.speed * dt * 60)
|
|
|
|
def _update_horizontal(self, dt: float) -> None:
|
|
self.x += int(self.speed * dt * 60)
|
|
|
|
def _update_omni(self, dt: float) -> None:
|
|
speed = self.speed * dt * 60
|
|
self.y += int(speed)
|
|
self.x += int(speed * 0.5)
|
|
|
|
def _update_floating(self, dt: float) -> None:
|
|
base = self.speed * 30
|
|
self.y = int(math.sin(self._time * 2) * base)
|
|
self.x = int(math.cos(self._time * 1.5) * base * 0.5)
|
|
|
|
def _update_bounce(self, dt: float) -> None:
|
|
"""Bouncing DVD-style camera that bounces off canvas edges."""
|
|
vw = self.viewport_width
|
|
vh = self.viewport_height
|
|
|
|
# Initialize direction if not set
|
|
if not hasattr(self, "_bounce_dx"):
|
|
self._bounce_dx = 1
|
|
self._bounce_dy = 1
|
|
|
|
# Calculate max positions
|
|
max_x = max(0, self.canvas_width - vw)
|
|
max_y = max(0, self.canvas_height - vh)
|
|
|
|
# Move
|
|
move_speed = self.speed * dt * 60
|
|
|
|
# Bounce off edges - reverse direction when hitting bounds
|
|
self.x += int(move_speed * self._bounce_dx)
|
|
self.y += int(move_speed * self._bounce_dy)
|
|
|
|
# Bounce horizontally
|
|
if self.x <= 0:
|
|
self.x = 0
|
|
self._bounce_dx = 1
|
|
elif self.x >= max_x:
|
|
self.x = max_x
|
|
self._bounce_dx = -1
|
|
|
|
# Bounce vertically
|
|
if self.y <= 0:
|
|
self.y = 0
|
|
self._bounce_dy = 1
|
|
elif self.y >= max_y:
|
|
self.y = max_y
|
|
self._bounce_dy = -1
|
|
|
|
def reset(self) -> None:
|
|
"""Reset camera position."""
|
|
self.x = 0
|
|
self.y = 0
|
|
self._time = 0.0
|
|
self.zoom = 1.0
|
|
|
|
def set_canvas_size(self, width: int, height: int) -> None:
|
|
"""Set the canvas size and clamp position if needed.
|
|
|
|
Args:
|
|
width: New canvas width
|
|
height: New canvas height
|
|
"""
|
|
self.canvas_width = width
|
|
self.canvas_height = height
|
|
self._clamp_to_bounds()
|
|
|
|
@classmethod
|
|
def vertical(cls, speed: float = 1.0) -> "Camera":
|
|
"""Create a vertical scrolling camera."""
|
|
return cls(mode=CameraMode.VERTICAL, speed=speed, canvas_height=200)
|
|
|
|
@classmethod
|
|
def horizontal(cls, speed: float = 1.0) -> "Camera":
|
|
"""Create a horizontal scrolling camera."""
|
|
return cls(mode=CameraMode.HORIZONTAL, speed=speed, canvas_width=200)
|
|
|
|
@classmethod
|
|
def omni(cls, speed: float = 1.0) -> "Camera":
|
|
"""Create an omnidirectional scrolling camera."""
|
|
return cls(
|
|
mode=CameraMode.OMNI, speed=speed, canvas_width=200, canvas_height=200
|
|
)
|
|
|
|
@classmethod
|
|
def floating(cls, speed: float = 1.0) -> "Camera":
|
|
"""Create a floating/bobbing camera."""
|
|
return cls(
|
|
mode=CameraMode.FLOATING, speed=speed, canvas_width=200, canvas_height=200
|
|
)
|
|
|
|
@classmethod
|
|
def bounce(cls, speed: float = 1.0) -> "Camera":
|
|
"""Create a bouncing DVD-style camera that bounces off canvas edges."""
|
|
return cls(
|
|
mode=CameraMode.BOUNCE, speed=speed, canvas_width=200, canvas_height=200
|
|
)
|
|
|
|
@classmethod
|
|
def custom(cls, update_fn: Callable[["Camera", float], None]) -> "Camera":
|
|
"""Create a camera with custom update function."""
|
|
return cls(custom_update=update_fn)
|