klubhaus/sideline
2
0
Fork 0
forked from genewildish/Mainline
Code Issues 5 Pull Requests Projects 1 Releases Wiki Activity

fix: Update imports to use engine.pipeline instead of engine.pipeline.core

Browse Source 
The old engine/pipeline/core.py file was removed as part of the Sideline/Mainline split.
All imports that referenced engine.pipeline.core have been updated to use engine.pipeline
which re-exports from sideline.pipeline.core.

This ensures consistency and avoids duplicate DataType enum instances.
This commit is contained in:
David Gwilliam
2026-03-23 20:45:40 -07:00
parent 98a5862c74
commit fc7f58685a
19 changed files with 66 additions and 373 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
engine/data_sources/pipeline_introspection.py
View File

@@ -67,13 +67,13 @@ class PipelineIntrospectionSource(DataSource):
@property
def inlet_types(self) -> set:
from engine.pipeline.core import DataType
from engine.pipeline import DataType
return {DataType.NONE}
@property
def outlet_types(self) -> set:
from engine.pipeline.core import DataType
from engine.pipeline import DataType
return {DataType.SOURCE_ITEMS}

4
engine/effects/plugins/tint.py
View File

@@ -20,13 +20,13 @@ class TintEffect(EffectPlugin):
# Define inlet types for PureData-style typing
@property
def inlet_types(self) -> set:
from engine.pipeline.core import DataType
from engine.pipeline import DataType
return {DataType.TEXT_BUFFER}
@property
def outlet_types(self) -> set:
from engine.pipeline.core import DataType
from engine.pipeline import DataType
return {DataType.TEXT_BUFFER}

2
engine/interfaces/__init__.py
View File

@@ -28,7 +28,7 @@ from engine.effects.types import (
apply_param_bindings,
create_effect_context,
)
from engine.pipeline.core import (
from engine.pipeline import (
DataType,
Stage,
StageConfig,

7
engine/pipeline/__init__.py
View File

@@ -26,6 +26,11 @@ from sideline.pipeline import (
register_source,
)
# Also re-export from sideline.core for compatibility
from sideline.pipeline.core import (
DataType,
)
# Re-export from engine.pipeline.presets (Mainline-specific)
from engine.pipeline.presets import (
DEMO_PRESET,
@@ -91,4 +96,6 @@ __all__ = [
"register_effect",
"register_display",
"register_camera",
# Core types (from sideline)
"DataType",
]

2
engine/pipeline/adapters/camera.py
View File

@@ -3,7 +3,7 @@
import time
from typing import Any
from engine.pipeline.core import DataType, PipelineContext, Stage
from engine.pipeline import DataType, PipelineContext, Stage
class CameraClockStage(Stage):

2
engine/pipeline/adapters/data_source.py
View File

@@ -8,7 +8,7 @@ This module provides adapters that wrap existing components
from typing import Any
from engine.data_sources import SourceItem
from engine.pipeline.core import DataType, PipelineContext, Stage
from engine.pipeline import DataType, PipelineContext, Stage
class DataSourceStage(Stage):

6
engine/pipeline/adapters/display.py
View File

@@ -2,7 +2,7 @@
from typing import Any
from engine.pipeline.core import PipelineContext, Stage
from engine.pipeline import PipelineContext, Stage
class DisplayStage(Stage):
@@ -59,13 +59,13 @@ class DisplayStage(Stage):
@property
def inlet_types(self) -> set:
from engine.pipeline.core import DataType
from engine.pipeline import DataType
return {DataType.TEXT_BUFFER} # Display consumes rendered text
@property
def outlet_types(self) -> set:
from engine.pipeline.core import DataType
from engine.pipeline import DataType
return {DataType.NONE} # Display is a terminal stage (no output)

6
engine/pipeline/adapters/effect_plugin.py
View File

@@ -2,7 +2,7 @@
from typing import Any
from engine.pipeline.core import PipelineContext, Stage
from engine.pipeline import PipelineContext, Stage
class EffectPluginStage(Stage):
@@ -69,13 +69,13 @@ class EffectPluginStage(Stage):
@property
def inlet_types(self) -> set:
from engine.pipeline.core import DataType
from engine.pipeline import DataType
return {DataType.TEXT_BUFFER}
@property
def outlet_types(self) -> set:
from engine.pipeline.core import DataType
from engine.pipeline import DataType
return {DataType.TEXT_BUFFER}

2
engine/pipeline/adapters/frame_capture.py
View File

@@ -7,7 +7,7 @@ Wraps pipeline stages to capture frames for animation report generation.
from typing import Any
from engine.display.backends.animation_report import AnimationReportDisplay
from engine.pipeline.core import PipelineContext, Stage
from engine.pipeline import PipelineContext, Stage
class FrameCaptureStage(Stage):

2
engine/pipeline/adapters/message_overlay.py
View File

@@ -12,7 +12,7 @@ from datetime import datetime
from engine import config
from engine.effects.legacy import vis_trunc
from engine.pipeline.core import DataType, PipelineContext, Stage
from engine.pipeline import DataType, PipelineContext, Stage
from engine.render.blocks import big_wrap
from engine.render.gradient import msg_gradient

2
engine/pipeline/adapters/positioning.py
View File

@@ -10,7 +10,7 @@ different ANSI positioning approaches:
from enum import Enum
from typing import Any
from engine.pipeline.core import DataType, PipelineContext, Stage
from engine.pipeline import DataType, PipelineContext, Stage
class PositioningMode(Enum):

2
engine/pipeline/adapters/transform.py
View File

@@ -4,7 +4,7 @@ from typing import Any
import engine.render
from engine.data_sources import SourceItem
from engine.pipeline.core import DataType, PipelineContext, Stage
from engine.pipeline import DataType, PipelineContext, Stage
def estimate_simple_height(text: str, width: int) -> int:

4
engine/pipeline/controller.py
View File

@@ -9,7 +9,7 @@ import time
from dataclasses import dataclass, field
from typing import Any
from engine.pipeline.core import PipelineContext, Stage, StageError, StageResult
from engine.pipeline import PipelineContext, Stage, StageError, StageResult
from engine.pipeline.params import PipelineParams
from engine.pipeline.registry import StageRegistry
@@ -640,7 +640,7 @@ class Pipeline:
Raises StageError if type mismatch is detected.
"""
from engine.pipeline.core import DataType
from engine.pipeline import DataType
errors: list[str] = []

321
engine/pipeline/core.py
View File

@@ -1,321 +0,0 @@
"""
Pipeline core - Unified Stage abstraction and PipelineContext.
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
PIL_IMAGE: PIL Image object
"""
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
PIL_IMAGE = auto() # PIL Image object
ANY = auto() # Accepts any type
NONE = auto() # No data (terminator)
@dataclass
class StageConfig:
"""Configuration for a single stage."""
name: str
category: str
enabled: bool = True
optional: bool = False
params: dict[str, Any] = field(default_factory=dict)
class Stage(ABC):
"""Abstract base class for all pipeline stages.
A Stage is a single component in the rendering pipeline. Stages can be:
- Sources: Data providers (headlines, poetry, pipeline viz)
- 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", "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.
Examples:
- "source.headlines"
- "effect.noise"
- "display.output"
- "camera"
"""
return {f"{self.category}.{self.name}"}
@property
def dependencies(self) -> set[str]:
"""Return set of capability names this stage needs.
Examples:
- {"display.output"}
- {"source.headlines"}
- {"camera"}
"""
return set()
@property
def stage_dependencies(self) -> set[str]:
"""Return set of stage names this stage must connect to directly.
This allows explicit stage-to-stage dependencies, useful for enforcing
pipeline structure when capability matching alone is insufficient.
Examples:
- {"viewport_filter"} # Must connect to viewport_filter stage
- {"camera_update"} # Must connect to camera_update stage
NOTE: These are stage names (as added to pipeline), not capabilities.
"""
return set()
def init(self, ctx: "PipelineContext") -> bool:
"""Initialize stage with pipeline context.
Args:
ctx: PipelineContext for accessing services
Returns:
True if initialization succeeded, False otherwise
"""
return True
@abstractmethod
def process(self, data: Any, ctx: "PipelineContext") -> Any:
"""Process input data and return output.
Args:
data: Input data from previous stage (or initial data for first stage)
ctx: PipelineContext for accessing services and state
Returns:
Processed data for next stage
"""
...
def cleanup(self) -> None: # noqa: B027
"""Clean up resources when pipeline shuts down."""
pass
def get_config(self) -> StageConfig:
"""Return current configuration of this stage."""
return StageConfig(
name=self.name,
category=self.category,
optional=self.optional,
)
def set_enabled(self, enabled: bool) -> None:
"""Enable or disable this stage."""
self._enabled = enabled # type: ignore[attr-defined]
def is_enabled(self) -> bool:
"""Check if stage is enabled."""
return getattr(self, "_enabled", True)
@dataclass
class StageResult:
"""Result of stage processing, including success/failure info."""
success: bool
data: Any
error: str | None = None
stage_name: str = ""
class PipelineContext:
"""Dependency injection context passed through the pipeline.
Provides:
- services: Named services (display, config, event_bus, etc.)
- state: Runtime state shared between stages
- params: PipelineParams for animation-driven config
Services can be injected at construction time or lazily resolved.
"""
def __init__(
self,
services: dict[str, Any] | None = None,
initial_state: dict[str, Any] | None = None,
):
self.services: dict[str, Any] = services or {}
self.state: dict[str, Any] = initial_state or {}
self._params: PipelineParams | None = None
# Lazy resolvers for common services
self._lazy_resolvers: dict[str, Callable[[], Any]] = {
"config": self._resolve_config,
"event_bus": self._resolve_event_bus,
}
def _resolve_config(self) -> Any:
from engine.config import get_config
return get_config()
def _resolve_event_bus(self) -> Any:
from engine.eventbus import get_event_bus
return get_event_bus()
def get(self, key: str, default: Any = None) -> Any:
"""Get a service or state value by key.
First checks services, then state, then lazy resolution.
"""
if key in self.services:
return self.services[key]
if key in self.state:
return self.state[key]
if key in self._lazy_resolvers:
try:
return self._lazy_resolvers[key]()
except Exception:
return default
return default
def set(self, key: str, value: Any) -> None:
"""Set a service or state value."""
self.services[key] = value
def set_state(self, key: str, value: Any) -> None:
"""Set a runtime state value."""
self.state[key] = value
def get_state(self, key: str, default: Any = None) -> Any:
"""Get a runtime state value."""
return self.state.get(key, default)
@property
def params(self) -> "PipelineParams | None":
"""Get current pipeline params (for animation)."""
return self._params
@params.setter
def params(self, value: "PipelineParams") -> None:
"""Set pipeline params (from animation controller)."""
self._params = value
def has_capability(self, capability: str) -> bool:
"""Check if a capability is available."""
return capability in self.services or capability in self._lazy_resolvers
class StageError(Exception):
"""Raised when a stage fails to process."""
def __init__(self, stage_name: str, message: str, is_optional: bool = False):
self.stage_name = stage_name
self.message = message
self.is_optional = is_optional
super().__init__(f"Stage '{stage_name}' failed: {message}")
def create_stage_error(
stage_name: str, error: Exception, is_optional: bool = False
) -> StageError:
"""Helper to create a StageError from an exception."""
return StageError(stage_name, str(error), is_optional)

4
engine/pipeline/registry.py
View File

@@ -8,10 +8,10 @@ from __future__ import annotations
from typing import TYPE_CHECKING, Any, TypeVar
from engine.pipeline.core import Stage
from engine.pipeline import Stage
if TYPE_CHECKING:
from engine.pipeline.core import Stage
from engine.pipeline import Stage
T = TypeVar("T")

2
engine/pipeline/stages/framebuffer.py
View File

@@ -14,7 +14,7 @@ from dataclasses import dataclass
from typing import Any
from engine.display import _strip_ansi
from engine.pipeline.core import DataType, PipelineContext, Stage
from engine.pipeline import DataType, PipelineContext, Stage
@dataclass

6
engine/sensors/__init__.py
View File

@@ -25,7 +25,7 @@ from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from engine.pipeline.core import PipelineContext
from engine.pipeline import PipelineContext
@dataclass
@@ -166,13 +166,13 @@ class SensorStage:
@property
def inlet_types(self) -> set:
from engine.pipeline.core import DataType
from engine.pipeline import DataType
return {DataType.ANY}
@property
def outlet_types(self) -> set:
from engine.pipeline.core import DataType
from engine.pipeline import DataType
return {DataType.ANY}