From 8e27f89fa4b2230b7e5a131c8884f53acfa73340 Mon Sep 17 00:00:00 2001 From: David Gwilliam Date: Mon, 16 Mar 2026 01:58:54 -0700 Subject: [PATCH] feat(pipeline): add self-documenting pipeline introspection - Add --pipeline-diagram flag to generate mermaid diagrams - Create engine/pipeline.py with PipelineIntrospector - Outputs flowchart, sequence diagram, and camera state diagram - Run with: python mainline.py --pipeline-diagram --- engine/app.py | 7 ++ engine/config.py | 3 + engine/pipeline.py | 265 +++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 275 insertions(+) create mode 100644 engine/pipeline.py diff --git a/engine/app.py b/engine/app.py index 806e2a7..1b39ad1 100644 --- a/engine/app.py +++ b/engine/app.py @@ -559,6 +559,13 @@ def run_demo_mode(): def main(): + from engine import config + from engine.pipeline import generate_pipeline_diagram + + if config.PIPELINE_DIAGRAM: + print(generate_pipeline_diagram()) + return + if config.DEMO: run_demo_mode() return diff --git a/engine/config.py b/engine/config.py index 6542563..c43475f 100644 --- a/engine/config.py +++ b/engine/config.py @@ -245,6 +245,9 @@ WEBSOCKET_PORT = _arg_int("--websocket-port", 8765) DEMO = "--demo" in sys.argv DEMO_EFFECT_DURATION = 5.0 # seconds per effect +# ─── PIPELINE DIAGRAM ──────────────────────────────────── +PIPELINE_DIAGRAM = "--pipeline-diagram" in sys.argv + def set_font_selection(font_path=None, font_index=None): """Set runtime primary font selection.""" diff --git a/engine/pipeline.py b/engine/pipeline.py new file mode 100644 index 0000000..70e0f63 --- /dev/null +++ b/engine/pipeline.py @@ -0,0 +1,265 @@ +""" +Pipeline introspection - generates self-documenting diagrams of the render pipeline. +""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass +class PipelineNode: + """Represents a node in the pipeline.""" + + name: str + module: str + class_name: str | None = None + func_name: str | None = None + description: str = "" + inputs: list[str] | None = None + outputs: list[str] | None = None + + +class PipelineIntrospector: + """Introspects the render pipeline and generates documentation.""" + + def __init__(self): + self.nodes: list[PipelineNode] = [] + + def add_node(self, node: PipelineNode) -> None: + self.nodes.append(node) + + def generate_mermaid_flowchart(self) -> str: + """Generate a Mermaid flowchart of the pipeline.""" + lines = ["```mermaid", "flowchart TD"] + + for node in self.nodes: + node_id = node.name.replace("-", "_").replace(" ", "_") + label = node.name + if node.class_name: + label = f"{node.name}\\n({node.class_name})" + elif node.func_name: + label = f"{node.name}\\n({node.func_name})" + + if node.description: + label += f"\\n{node.description}" + + lines.append(f' {node_id}["{label}"]') + + lines.append("") + + for node in self.nodes: + node_id = node.name.replace("-", "_").replace(" ", "_") + if node.inputs: + for inp in node.inputs: + inp_id = inp.replace("-", "_").replace(" ", "_") + lines.append(f" {inp_id} --> {node_id}") + + lines.append("```") + return "\n".join(lines) + + def generate_mermaid_sequence(self) -> str: + """Generate a Mermaid sequence diagram of message flow.""" + lines = ["```mermaid", "sequenceDiagram"] + + lines.append(" participant Sources") + lines.append(" participant Fetch") + lines.append(" participant Scroll") + lines.append(" participant Effects") + lines.append(" participant Display") + + lines.append(" Sources->>Fetch: headlines") + lines.append(" Fetch->>Scroll: content blocks") + lines.append(" Scroll->>Effects: buffer") + lines.append(" Effects->>Effects: process chain") + lines.append(" Effects->>Display: rendered buffer") + + lines.append("```") + return "\n".join(lines) + + def generate_mermaid_state(self) -> str: + """Generate a Mermaid state diagram of camera modes.""" + lines = ["```mermaid", "stateDiagram-v2"] + + lines.append(" [*] --> Vertical") + lines.append(" Vertical --> Horizontal: set_mode()") + lines.append(" Horizontal --> Omni: set_mode()") + lines.append(" Omni --> Floating: set_mode()") + lines.append(" Floating --> Vertical: set_mode()") + + lines.append(" state Vertical {") + lines.append(" [*] --> ScrollUp") + lines.append(" ScrollUp --> ScrollUp: +y each frame") + lines.append(" }") + + lines.append(" state Horizontal {") + lines.append(" [*] --> ScrollLeft") + lines.append(" ScrollLeft --> ScrollLeft: +x each frame") + lines.append(" }") + + lines.append(" state Omni {") + lines.append(" [*] --> Diagonal") + lines.append(" Diagonal --> Diagonal: +x, +y") + lines.append(" }") + + lines.append(" state Floating {") + lines.append(" [*] --> Bobbing") + lines.append(" Bobbing --> Bobbing: sin(time)") + lines.append(" }") + + lines.append("```") + return "\n".join(lines) + + def generate_full_diagram(self) -> str: + """Generate full pipeline documentation.""" + lines = [ + "# Render Pipeline", + "", + "## Data Flow", + "", + self.generate_mermaid_flowchart(), + "", + "## Message Sequence", + "", + self.generate_mermaid_sequence(), + "", + "## Camera States", + "", + self.generate_mermaid_state(), + ] + return "\n".join(lines) + + def introspect_sources(self) -> None: + """Introspect data sources.""" + from engine import sources + + for name in dir(sources): + obj = getattr(sources, name) + if isinstance(obj, dict): + self.add_node( + PipelineNode( + name=f"Data Source: {name}", + module="engine.sources", + description=f"{len(obj)} feeds configured", + ) + ) + + def introspect_fetch(self) -> None: + """Introspect fetch layer.""" + self.add_node( + PipelineNode( + name="fetch_all", + module="engine.fetch", + func_name="fetch_all", + description="Fetch RSS feeds", + outputs=["items"], + ) + ) + + self.add_node( + PipelineNode( + name="fetch_poetry", + module="engine.fetch", + func_name="fetch_poetry", + description="Fetch Poetry DB", + outputs=["items"], + ) + ) + + def introspect_scroll(self) -> None: + """Introspect scroll engine.""" + self.add_node( + PipelineNode( + name="StreamController", + module="engine.controller", + class_name="StreamController", + description="Main render loop orchestrator", + inputs=["items", "ntfy_poller", "mic_monitor", "display"], + outputs=["buffer"], + ) + ) + + self.add_node( + PipelineNode( + name="render_ticker_zone", + module="engine.layers", + func_name="render_ticker_zone", + description="Render scrolling ticker content", + inputs=["active", "camera"], + outputs=["buffer"], + ) + ) + + def introspect_camera(self) -> None: + """Introspect camera system.""" + self.add_node( + PipelineNode( + name="Camera", + module="engine.camera", + class_name="Camera", + description="Viewport position controller", + inputs=["dt"], + outputs=["x", "y"], + ) + ) + + def introspect_effects(self) -> None: + """Introspect effect system.""" + self.add_node( + PipelineNode( + name="EffectChain", + module="engine.effects", + class_name="EffectChain", + description="Process effects in sequence", + inputs=["buffer", "context"], + outputs=["buffer"], + ) + ) + + self.add_node( + PipelineNode( + name="EffectRegistry", + module="engine.effects", + class_name="EffectRegistry", + description="Manage effect plugins", + ) + ) + + def introspect_display(self) -> None: + """Introspect display backends.""" + from engine.display import DisplayRegistry + + DisplayRegistry.initialize() + backends = DisplayRegistry.list_backends() + + for backend in backends: + self.add_node( + PipelineNode( + name=f"Display: {backend}", + module="engine.display.backends", + class_name=f"{backend.title()}Display", + description=f"Render to {backend}", + inputs=["buffer"], + ) + ) + + def run(self) -> str: + """Run full introspection.""" + self.introspect_sources() + self.introspect_fetch() + self.introspect_scroll() + self.introspect_camera() + self.introspect_effects() + self.introspect_display() + + return self.generate_full_diagram() + + +def generate_pipeline_diagram() -> str: + """Generate a self-documenting pipeline diagram.""" + introspector = PipelineIntrospector() + return introspector.run() + + +if __name__ == "__main__": + print(generate_pipeline_diagram())