Pipeline Visualization

SGN can generate Graphviz diagrams of your pipeline to help with debugging and documentation.

Prerequisites

Install the graphviz Python package and the Graphviz system package:

pip install graphviz
# On Debian/Ubuntu: sudo apt install graphviz
# On macOS: brew install graphviz
# On Fedora: sudo dnf install graphviz

Render to File

from sgn import Pipeline, IterSource, CollectSink, CallableTransform

src = IterSource(name="src", source_pad_names=["H1"],
                 iters={"H1": [1, 2, 3]})
trn = CallableTransform.from_callable(
    name="t1", sink_pad_names=["H1"],
    callable=lambda f: f.data * 2 if f.data is not None else None,
    output_pad_name="H1",
)
snk = CollectSink(name="snk", sink_pad_names=["H1"])

p = Pipeline()
p.connect(src, trn)
p.connect(trn, snk)

# Render to a file (format inferred from extension)
p.visualize("my_pipeline.png", label="My Pipeline")

This generates my_pipeline.png with a visual representation of the pipeline.

Get a Graphviz Object

Use to_graph() to get a graphviz.Digraph object for further customization:

graph = p.to_graph(label="My Pipeline")
graph.render("output", format="svg")  # Save as SVG
graph.view()                           # Open in system viewer

Get DOT Source

Use to_dot() to get the raw DOT language string:

dot_source = p.to_dot(label="My Pipeline")
print(dot_source)

Color Coding

The generated diagrams use color to indicate pad types:

Color	Meaning
Green	Source pads (producing data)
Blue	Sink pads (receiving data)
Red	Unconnected pads (likely a bug)

Debugging Unconnected Pads

Before running p.run(), the pipeline calls p.check() which raises a RuntimeError for any unlinked pads. Visualizing first can help identify the problem:

# Build pipeline but don't run yet
p = Pipeline()
p.connect(src, trn)
# Forgot to connect trn -> snk!

# Visualize to see what's missing
p.visualize("debug.png")
# The diagram will show trn's source pad and snk's sink pad in red