sgn.visualize

visualize(pipeline, label=None, path=None, expand_composed=False)

convert a pipeline to a graphviz.DiGraph object

Source pads are light green, sink pads are light blue, and unconnected pads are red.

When expand_composed is True, composed elements are rendered as labelled cluster subgraphs showing their full internal structure, including explicit boundary pad nodes where external connections enter and leave the composition. When False, each composed element is shown as a single opaque box with only its own boundary source and sink pads visible.

Parameters:

Name	Type	Description	Default
pipeline	Graph	The pipeline to visualize.	required
label	str | None	Label for graph.	None
path	Path | None	If set, save graph visualization to a file (format based on file extension).	None
expand_composed	bool	Whether to expand composed elements into cluster subgraphs. Defaults to False.	False

Returns:

Type	Description
	graphviz.Digraph: the graph object

Source code in src/sgn/visualize.py

def visualize(pipeline, label=None, path=None, expand_composed=False):
    """convert a pipeline to a graphviz.DiGraph object

    Source pads are light green, sink pads are light blue, and
    unconnected pads are red.

    When ``expand_composed`` is True, composed elements are rendered
    as labelled cluster subgraphs showing their full internal structure,
    including explicit boundary pad nodes where external connections
    enter and leave the composition.
    When False, each composed element is shown as a single opaque box
    with only its own boundary source and sink pads visible.

    Args:
        pipeline (Graph): The pipeline to visualize.
        label (str | None): Label for graph.
        path (Path | None): If set, save graph visualization to a file
            (format based on file extension).
        expand_composed (bool): Whether to expand composed elements into
            cluster subgraphs. Defaults to False.

    Returns:
        graphviz.Digraph: the graph object

    """
    try:
        import graphviz
    except ImportError:
        raise ImportError("graphviz needs to be installed to visualize pipelines")

    graph = graphviz.Digraph(
        "pipeline",
        graph_attr={
            "labelloc": "t",
            "rankdir": "LR",
            "ranksep": "2",
        },
        node_attr={
            "shape": "plaintext",
            "fontname": "times mono",
        },
    )
    if label:
        graph.graph_attr["label"] = (
            f"""<<font point-size="32"><b>{escape(label)}</b></font>>"""
        )

    top_level_names = pipeline.nodes(pads=False)

    if expand_composed:
        # Gather all composed elements at every level of nesting
        all_composed = []
        for name in top_level_names:
            all_composed.extend(_collect_composed(pipeline[name]))

        # Map each composed boundary pad to the internal pads it routes to
        pad_remap = _build_pad_remap(all_composed)

        # Internal boundary source pads shown as linked when their outer pad is linked
        linked_pads = _collect_linked_pads(pad_remap)

        # Route virtual-source edges: virtual_src → internal_snk
        #   becomes: outer_snk → internal_snk
        virtual_src_to_outer_snk = {
            vs: ce.snks[snk_name]
            for ce in all_composed
            for snk_name, vs in ce._virtual_sources.items()
        }

        # Route internal boundary source → InternalPad edges:
        #   internal_src → InternalPad  becomes: internal_src → outer_src
        internal_src_to_outer_src = {}
        for ce in all_composed:
            for pad_name, internal_src in ce._boundary_source_pads.items():
                for outer_src in ce.source_pads:
                    if outer_src.pad_name == pad_name:
                        internal_src_to_outer_src[internal_src] = outer_src
                        break

        # Boundary pads are rendered as standalone nodes; use pad ID directly
        boundary_pads = {
            pad for ce in all_composed for pad in (*ce.sink_pads, *ce.source_pads)
        }

        def _endpoint(pad):
            if pad in boundary_pads:
                elem_id = _id(pad.element.name)
                if isinstance(pad, SinkPad):
                    return elem_id + "__sinks:" + _id(pad.name)
                else:
                    return elem_id + "__srcs:" + _id(pad.name)
            return pad.element.name + ":" + _id(pad.name)

        # Create nodes for all top-level elements (clusters for composed elements)
        for name in top_level_names:
            _add_element_to_graph(graph, pipeline[name], linked_pads)

        # Connect element pads with boundary-aware edge routing
        for sname, tname in pipeline.edges(pads=True, intra=False):
            source = pipeline[sname]
            target = pipeline[tname]

            # Virtual-source edge → outer sink pad → internal element sink pad
            if source in virtual_src_to_outer_snk:
                outer_snk = virtual_src_to_outer_snk[source]
                graph.edge(_endpoint(outer_snk), _endpoint(target))
                continue

            # Internal boundary source → InternalPad → internal source → outer source
            if isinstance(target, InternalPad):
                if source in internal_src_to_outer_src:
                    outer_src = internal_src_to_outer_src[source]
                    graph.edge(_endpoint(source), _endpoint(outer_src))
                continue

            graph.edge(_endpoint(source), _endpoint(target))

    else:
        from sgn.compose import ComposedElementMixin

        # Render every top-level element (including composed) as a plain box
        for name in top_level_names:
            element = pipeline[name]
            is_composed = isinstance(element, ComposedElementMixin)
            bg = _element_bg(element, shade=1 if is_composed else 0)
            graph.node(element.name, _element_struct_plaintext(element, bg_color=bg))

        # Only draw edges between top-level elements; skip internal infrastructure
        for sname, tname in pipeline.edges(pads=True, intra=False):
            source = pipeline[sname]
            target = pipeline[tname]
            if (
                source.element.name in top_level_names
                and target.element.name in top_level_names
            ):
                graph.edge(
                    source.element.name + ":" + _id(source.name),
                    target.element.name + ":" + _id(target.name),
                )

    if path:
        graph.render(
            outfile=path,
            cleanup=True,
        )

    return graph