Pipeline Graph

PipelineGraph draws a DreamLake pipeline as an interactive flow-chart. It's purely presentational: you hand it one JSON object and it renders the DAG — dotted canvas, status-tinted node cards, orthogonal edges, live flow animation — with no data fetching of its own. PipelineSource is the paired read-only source inspector.

If you just want to render a graph, this page is the whole story. For the exact data shape see Pipeline Graph JSON; for a field-by-field visual map see Anatomy; for how the component is built and what's coming, see Architecture & Roadmap.

Background — what this renders and why

DreamLake runs an AI auto-labeling + human-review loop: models propose labels, reviewers gate them, accepted rows land in a dataset and the rest go back for rework. That loop is written as a pipeline — a plain Python @dl.pipeline function whose stages are @ls.udf functions.

You never draw this graph by hand. The dl_trace tracer reads the pipeline's Python statically (no import, no execution) and derives the node/edge graph from its dataflow. So:

• The graph is a derived view of code. Change the .py, re-trace, the graph updates. There is no separate diagram to keep in sync.
• Placeholder bodies (...) trace like real ones, so a graph renders long before any stage is implemented.
• This component only draws. Tracing happens elsewhere (a Python service); the runtime status that animates the graph streams in separately.

Who this is for: anyone rendering a traced pipeline in a UI — a Studio view, a dashboard, a docs page. You supply the traced JSON (and optionally a live status overlay); the component owns the canvas.

One stage, three faces

The thing that makes the data model click: a node is one stage seen three ways. You write a Python UDF; the tracer emits a JSON node; the component draws a card. They are the same object.

① The Python you write — a @ls.udf stage. Its parameters become input ports; its return columns become the result schema:

@ls.udf
def detect_objects(images: Tensor["N", "H", "W", 3]) -> Tuple["boxes", "classes", "confidence"]:
    """Detect objects in each image; one row per box."""
    ...

② The node you see, ③ the JSON in between — flip between the rendered card (Preview), the render call (Source), and the tracer's node JSON (Data). The card is the JSON: the kind dot ← kind, the title ← title, the 1→1 meta ← inputs/outputs lengths, the edge dots ← the inputs / outputs ports.

The return columns (boxes, classes, confidence) live in columns — the result's schema, not extra ports. A UDF returns one table; passing it downstream passes the whole table. See Anatomy for the full field-to-pixel map.

Basic

A freshly-traced graph is entirely idle. Drag nodes to rearrange, scroll (or two-finger drag) to pan, ⌘/ctrl-scroll or pinch to zoom, click a node to select it. Once the canvas is focused, arrow keys walk the selection — ↑ / ↓ step through the pipeline in topological order, ← / → jump to the upstream / downstream neighbour, Esc clears — and the selected node pans into view. Edges come in two kinds — data (solid) and mask (dashed gate); the Anatomy page shows both.

Graph + source, linked

The design layout is a canvas with a source right rail. The two share one selection: click a node and its source shows on the right; click the background to clear. Both components are controlled — you own the selectedNodeId state.

"""Recover the camera pose trajectory from a video.

frames → features → pose estimation → bundle adjustment; a confidence
mask (σ-algebra: intersection of two boolean columns) decides what goes
to the dataset vs. rework. EVERY function the pipeline calls is a
`@ls.udf` — the source and the two sinks included. UDF bodies are
placeholders.
"""

import dreamlake as dl
import lakeshore as ls
from dreamlake import batch, requeue, to_dataset
from lakeshore.types import Tensor, Tuple


@ls.udf(kind="source")
def load_videos() -> Tuple["videos"]:
    """Pull the batch of videos to process."""
    ...


@ls.udf
def extract_frames(videos: Tensor["N"]) -> Tuple["frames", "timestamps"]:
    """Decode each video into a frame column plus per-frame timestamps."""
    ...


@ls.udf
def detect_features(frames: Tensor["N", "H", "W", 3]) -> Tuple["keypoints", "descriptors"]:
    """Per-frame keypoints and descriptors (e.g. SuperPoint)."""
    ...


@ls.udf
def estimate_poses(keypoints, descriptors) -> Tuple["poses", "confidence"]:
    """Visual odometry / SfM placeholder; poses are 4x4 world-from-camera."""
    ...


@ls.udf
def bundle_adjust(poses: Tensor["N", 4, 4]) -> Tuple["poses", "residual"]:
    """Global refinement; residual is the reprojection error per frame."""
    ...


@ls.udf(kind="sink")
def save_dataset(rows):
    """Write the trajectory to the dataset (wraps dreamlake.to_dataset)."""
    to_dataset(rows)


@ls.udf(kind="sink")
def rework(rows):
    """Send low-confidence frames back (wraps dreamlake.requeue)."""
    requeue(rows)


@dl.pipeline
def camera_pose_trajectory():
    src = load_videos()
    for items in batch(src, n=8):               # batch = chunked/streamed run
        frames = extract_frames(items.videos)
        feats  = detect_features(frames.frames)
        poses  = estimate_poses(feats.keypoints, feats.descriptors)
        traj   = bundle_adjust(poses.poses)

        ok = (poses.confidence > 0.5) & (traj.residual < 1.0)   # A ∩ B

        save_dataset(traj[ok])   # sink: the trajectory
        rework(traj[~ok])        # low-confidence frames go back

Live status — a runnable pipeline

Edges carry no stored style. Each edge's visual flow is derived from the status of its two endpoint nodes, so animating a running pipeline is just a matter of feeding node statuses in via the statusById overlay. The six flow states, and how each is derived, are catalogued in Anatomy → Connector states.

The demo below is a tiny in-browser "runner". Each node, when it finishes, settles to a random outcome — mostly ok, occasionally stale or error. A node only starts once all its upstreams have settled and none errored, so a failure blocks everything downstream (those nodes never run and stay idle), exactly like a real pipeline. Nothing real executes — it only drives statusById. Press Run (each run re-rolls): running edges flow blue, ok edges go green, a stale node's edges turn amber, an error node's edges turn red, and nodes blocked by an upstream failure stay idle.

Props

PipelineGraph

PipelineSource

Both are theme-aware (uikit tone tokens) and load @dreamlake/uikit/styles.css for their colours.

Next: Anatomy maps every field to a pixel · Pipeline Graph JSON is the data-model reference · Architecture & Roadmap covers the internals and what's next.