Files
Mainline/engine/pipeline/params.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

146 lines
4.7 KiB
Python

"""
Pipeline parameters - Runtime configuration layer for animation control.
PipelineParams is the target for AnimationController - animation events
modify these params, which the pipeline then applies to its stages.
"""
from dataclasses import dataclass, field
from typing import Any
@dataclass
class PipelineParams:
"""Runtime configuration for the pipeline.
This is the canonical config object that AnimationController modifies.
Stages read from these params to adjust their behavior.
"""
# Source config
source: str = "headlines"
source_refresh_interval: float = 60.0
# Display config
display: str = "terminal"
border: bool = False
# Camera config
camera_mode: str = "vertical"
camera_speed: float = 1.0
camera_x: int = 0 # For horizontal scrolling
# Effect config
effect_order: list[str] = field(
default_factory=lambda: ["noise", "fade", "glitch", "firehose", "hud"]
)
effect_enabled: dict[str, bool] = field(default_factory=dict)
effect_intensity: dict[str, float] = field(default_factory=dict)
# Animation-driven state (set by AnimationController)
pulse: float = 0.0
current_effect: str | None = None
path_progress: float = 0.0
# Viewport
viewport_width: int = 80
viewport_height: int = 24
# Firehose
firehose_enabled: bool = False
# Runtime state
frame_number: int = 0
fps: float = 60.0
def get_effect_config(self, name: str) -> tuple[bool, float]:
"""Get (enabled, intensity) for an effect."""
enabled = self.effect_enabled.get(name, True)
intensity = self.effect_intensity.get(name, 1.0)
return enabled, intensity
def set_effect_config(self, name: str, enabled: bool, intensity: float) -> None:
"""Set effect configuration."""
self.effect_enabled[name] = enabled
self.effect_intensity[name] = intensity
def is_effect_enabled(self, name: str) -> bool:
"""Check if an effect is enabled."""
if name not in self.effect_enabled:
return True # Default to enabled
return self.effect_enabled.get(name, True)
def get_effect_intensity(self, name: str) -> float:
"""Get effect intensity (0.0 to 1.0)."""
return self.effect_intensity.get(name, 1.0)
def to_dict(self) -> dict[str, Any]:
"""Convert to dictionary for serialization."""
return {
"source": self.source,
"display": self.display,
"camera_mode": self.camera_mode,
"camera_speed": self.camera_speed,
"effect_order": self.effect_order,
"effect_enabled": self.effect_enabled.copy(),
"effect_intensity": self.effect_intensity.copy(),
"pulse": self.pulse,
"current_effect": self.current_effect,
"viewport_width": self.viewport_width,
"viewport_height": self.viewport_height,
"firehose_enabled": self.firehose_enabled,
}
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "PipelineParams":
"""Create from dictionary."""
params = cls()
for key, value in data.items():
if hasattr(params, key):
setattr(params, key, value)
return params
def copy(self) -> "PipelineParams":
"""Create a copy of this params object."""
params = PipelineParams()
params.source = self.source
params.display = self.display
params.camera_mode = self.camera_mode
params.camera_speed = self.camera_speed
params.camera_x = self.camera_x
params.effect_order = self.effect_order.copy()
params.effect_enabled = self.effect_enabled.copy()
params.effect_intensity = self.effect_intensity.copy()
params.pulse = self.pulse
params.current_effect = self.current_effect
params.path_progress = self.path_progress
params.viewport_width = self.viewport_width
params.viewport_height = self.viewport_height
params.firehose_enabled = self.firehose_enabled
params.frame_number = self.frame_number
params.fps = self.fps
return params
# Default params for different modes
DEFAULT_HEADLINE_PARAMS = PipelineParams(
source="headlines",
display="terminal",
camera_mode="vertical",
effect_order=["noise", "fade", "glitch", "firehose", "hud"],
)
DEFAULT_PYGAME_PARAMS = PipelineParams(
source="headlines",
display="pygame",
camera_mode="vertical",
effect_order=["noise", "fade", "glitch", "firehose", "hud"],
)
DEFAULT_PIPELINE_PARAMS = PipelineParams(
source="pipeline",
display="pygame",
camera_mode="trace",
effect_order=["hud"], # Just HUD for pipeline viz
)