From f27f3475c8c7c1c3569d84eeac3fb45135d07000 Mon Sep 17 00:00:00 2001 From: David Gwilliam Date: Sat, 21 Mar 2026 19:26:32 -0700 Subject: [PATCH] feat(graph): Add adapter to convert graphs to pipelines Bridge the new graph abstraction with existing Stage-based pipeline system for backward compatibility. - Add GraphAdapter class to map nodes to Stage implementations - Handle effect intensity configuration (sets global effect state) - Map camera modes to Camera factory methods (feed, scroll, horizontal, etc.) - Auto-inject required dependencies (render, camera_update) via pipeline capabilities - Support for all major node types: source, camera, effect, position, display The adapter ensures that graphs seamlessly integrate with the existing pipeline architecture while providing a cleaner abstraction layer. --- engine/pipeline/graph_adapter.py | 161 +++++++++++++++++++++++++++++++ 1 file changed, 161 insertions(+) create mode 100644 engine/pipeline/graph_adapter.py diff --git a/engine/pipeline/graph_adapter.py b/engine/pipeline/graph_adapter.py new file mode 100644 index 0000000..cb614ab --- /dev/null +++ b/engine/pipeline/graph_adapter.py @@ -0,0 +1,161 @@ +"""Adapter to convert Graph to Pipeline stages. + +This module bridges the new graph-based abstraction with the existing +Stage-based pipeline system for backward compatibility. +""" + +from typing import Dict, Any, List, Optional + +from engine.pipeline.graph import Graph, NodeType +from engine.pipeline.controller import Pipeline, PipelineConfig +from engine.pipeline.core import PipelineContext +from engine.pipeline.params import PipelineParams +from engine.pipeline.adapters import ( + CameraStage, + DataSourceStage, + DisplayStage, + EffectPluginStage, + FontStage, + MessageOverlayStage, + PositionStage, + ViewportFilterStage, + create_stage_from_display, + create_stage_from_effect, +) +from engine.pipeline.adapters.positioning import PositioningMode +from engine.display import DisplayRegistry +from engine.effects import get_registry +from engine.data_sources.sources import EmptyDataSource, HeadlinesDataSource +from engine.camera import Camera + + +class GraphAdapter: + """Converts Graph to Pipeline with existing Stage classes.""" + + def __init__(self, graph: Graph): + self.graph = graph + self.pipeline: Optional[Pipeline] = None + self.context: Optional[PipelineContext] = None + + def build_pipeline( + self, viewport_width: int = 80, viewport_height: int = 24 + ) -> Pipeline: + """Build a Pipeline from the Graph.""" + # Create pipeline context + self.context = PipelineContext() + self.context.terminal_width = viewport_width + self.context.terminal_height = viewport_height + + # Create params + params = PipelineParams( + viewport_width=viewport_width, + viewport_height=viewport_height, + ) + self.context.params = params + + # Create pipeline config + config = PipelineConfig() + + # Create pipeline + self.pipeline = Pipeline(config=config, context=self.context) + + # Map graph nodes to pipeline stages + self._map_nodes_to_stages() + + # Build pipeline + self.pipeline.build() + + return self.pipeline + + def _map_nodes_to_stages(self) -> None: + """Map graph nodes to pipeline stages.""" + for name, node in self.graph.nodes.items(): + if not node.enabled: + continue + + stage = self._create_stage_from_node(name, node) + if stage: + self.pipeline.add_stage(name, stage) + + def _create_stage_from_node(self, name: str, node) -> Optional: + """Create a pipeline stage from a graph node.""" + stage = None + + if node.type == NodeType.SOURCE: + source_type = node.config.get("source", "headlines") + if source_type == "headlines": + source = HeadlinesDataSource() + elif source_type == "empty": + source = EmptyDataSource( + width=self.context.terminal_width, + height=self.context.terminal_height, + ) + else: + source = EmptyDataSource( + width=self.context.terminal_width, + height=self.context.terminal_height, + ) + stage = DataSourceStage(source, name=name) + + elif node.type == NodeType.CAMERA: + mode = node.config.get("mode", "scroll") + speed = node.config.get("speed", 1.0) + # Map mode string to Camera factory method + mode_lower = mode.lower() + if hasattr(Camera, mode_lower): + camera_factory = getattr(Camera, mode_lower) + camera = camera_factory(speed=speed) + else: + # Fallback to scroll mode + camera = Camera.scroll(speed=speed) + stage = CameraStage(camera, name=name) + + elif node.type == NodeType.DISPLAY: + backend = node.config.get("backend", "terminal") + positioning = node.config.get("positioning", "mixed") + display = DisplayRegistry.create(backend) + if display: + stage = DisplayStage(display, name=name, positioning=positioning) + + elif node.type == NodeType.EFFECT: + effect_name = node.config.get("effect", "") + intensity = node.config.get("intensity", 1.0) + effect = get_registry().get(effect_name) + if effect: + # Set effect intensity (modifies global effect state) + effect.config.intensity = intensity + # Effects typically depend on rendered output + dependencies = {"render.output"} + stage = EffectPluginStage(effect, name=name, dependencies=dependencies) + + elif node.type == NodeType.RENDER: + stage = FontStage(name=name) + + elif node.type == NodeType.OVERLAY: + stage = MessageOverlayStage(name=name) + + elif node.type == NodeType.POSITION: + mode_str = node.config.get("mode", "mixed") + try: + mode = PositioningMode(mode_str) + except ValueError: + mode = PositioningMode.MIXED + stage = PositionStage(mode=mode, name=name) + + return stage + + +def graph_to_pipeline( + graph: Graph, viewport_width: int = 80, viewport_height: int = 24 +) -> Pipeline: + """Convert a Graph to a Pipeline.""" + adapter = GraphAdapter(graph) + return adapter.build_pipeline(viewport_width, viewport_height) + + +def dict_to_pipeline( + data: Dict[str, Any], viewport_width: int = 80, viewport_height: int = 24 +) -> Pipeline: + """Convert a dictionary to a Pipeline.""" + graph = Graph().from_dict(data) + return graph_to_pipeline(graph, viewport_width, viewport_height)