feat(pipeline): add PureData-style inlet/outlet typing

- Add DataType enum (SOURCE_ITEMS, TEXT_BUFFER, etc.)
- Add inlet_types and outlet_types to Stage
- Add _validate_types() for type checking at build time
- Update tests with proper type annotations
This commit is contained in:
2026-03-16 15:39:36 -07:00
parent 4616a21359
commit 76126bdaac
4 changed files with 717 additions and 2 deletions

View File

@@ -5,17 +5,40 @@ This module provides the foundation for a clean, dependency-managed pipeline:
- Stage: Base class for all pipeline components (sources, effects, displays, cameras)
- PipelineContext: Dependency injection context for runtime data exchange
- Capability system: Explicit capability declarations with duck-typing support
- DataType: PureData-style inlet/outlet typing for validation
"""
from abc import ABC, abstractmethod
from collections.abc import Callable
from dataclasses import dataclass, field
from enum import Enum, auto
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from engine.pipeline.params import PipelineParams
class DataType(Enum):
"""PureData-style data types for inlet/outlet validation.
Each type represents a specific data format that flows through the pipeline.
This enables compile-time-like validation of connections.
Examples:
SOURCE_ITEMS: List[SourceItem] - raw items from sources
ITEM_TUPLES: List[tuple] - (title, source, timestamp) tuples
TEXT_BUFFER: List[str] - rendered ANSI buffer for display
RAW_TEXT: str - raw text strings
"""
SOURCE_ITEMS = auto() # List[SourceItem] - from DataSource
ITEM_TUPLES = auto() # List[tuple] - (title, source, ts)
TEXT_BUFFER = auto() # List[str] - ANSI buffer
RAW_TEXT = auto() # str - raw text
ANY = auto() # Accepts any type
NONE = auto() # No data (terminator)
@dataclass
class StageConfig:
"""Configuration for a single stage."""
@@ -35,18 +58,78 @@ class Stage(ABC):
- Effects: Post-processors (noise, fade, glitch, hud)
- Displays: Output backends (terminal, pygame, websocket)
- Cameras: Viewport controllers (vertical, horizontal, omni)
- Overlays: UI elements that compose on top (HUD)
Stages declare:
- capabilities: What they provide to other stages
- dependencies: What they need from other stages
- stage_type: Category of stage (source, effect, overlay, display)
- render_order: Execution order within category
- is_overlay: If True, output is composited on top, not passed downstream
Duck-typing is supported: any class with the required methods can act as a Stage.
"""
name: str
category: str # "source", "effect", "display", "camera"
category: str # "source", "effect", "overlay", "display", "camera"
optional: bool = False # If True, pipeline continues even if stage fails
@property
def stage_type(self) -> str:
"""Category of stage for ordering.
Valid values: "source", "effect", "overlay", "display", "camera"
Defaults to category for backwards compatibility.
"""
return self.category
@property
def render_order(self) -> int:
"""Execution order within stage_type group.
Higher values execute later. Useful for ordering overlays
or effects that need specific execution order.
"""
return 0
@property
def is_overlay(self) -> bool:
"""If True, this stage's output is composited on top of the buffer.
Overlay stages don't pass their output to the next stage.
Instead, their output is layered on top of the final buffer.
Use this for HUD, status displays, and similar UI elements.
"""
return False
@property
def inlet_types(self) -> set[DataType]:
"""Return set of data types this stage accepts.
PureData-style inlet typing. If the connected upstream stage's
outlet_type is not in this set, the pipeline will raise an error.
Examples:
- Source stages: {DataType.NONE} (no input needed)
- Transform stages: {DataType.ITEM_TUPLES, DataType.TEXT_BUFFER}
- Display stages: {DataType.TEXT_BUFFER}
"""
return {DataType.ANY}
@property
def outlet_types(self) -> set[DataType]:
"""Return set of data types this stage produces.
PureData-style outlet typing. Downstream stages must accept
this type in their inlet_types.
Examples:
- Source stages: {DataType.SOURCE_ITEMS}
- Transform stages: {DataType.TEXT_BUFFER}
- Display stages: {DataType.NONE} (consumes data)
"""
return {DataType.ANY}
@property
def capabilities(self) -> set[str]:
"""Return set of capabilities this stage provides.