capturegraph.procedures.exporting.visualizer

Procedure Visualization

This module provides utilities for creating visual representations of procedure graphs using Graphviz. The visualizer converts procedure DAGs into flowcharts that help users understand workflow structure, data flow, and procedure dependencies.

The generated visualizations show: - Procedure nodes with their types and configurations - Data flow connections between procedures with input names - Sequential execution steps with step numbers - Hierarchical grouping options for complex workflows

procedure_to_graphviz(procedure, group_by_depth=False)

Converts a Procedure DAG into a Graphviz diagram for visualization.

Creates a directed graph showing the structure and data flow of a procedure workflow. Each procedure becomes a node showing its type, label, and settings. Edges show data flow (inputs) and control flow (sequential steps).

Parameters:

Name	Type	Description	Default
procedure	Procedure	The root procedure to visualize (any procedure in the DAG)	required
group_by_depth	bool	Whether to visually group nodes by their depth level in the procedure hierarchy for better layout	False

Returns:

Type	Description
Digraph	A Graphviz Digraph object that can be rendered to various formats
Digraph	(PNG, SVG, PDF, etc.)

Example

import capturegraph.procedures as cgp

# Create your procedure
my_procedure = my_capture_procedure()

# Generate visualization
graph = cgp.procedure_to_graphviz(my_procedure, group_by_depth=True)

# Render to file
graph.render('my_procedure', format='png', cleanup=True)

# Or view directly (if supported)
graph.view()

Source code in capturegraph-lib/capturegraph/procedures/exporting/visualizer.py

def procedure_to_graphviz(
    procedure: Procedure, group_by_depth: bool = False
) -> Digraph:
    """
    Converts a Procedure DAG into a Graphviz diagram for visualization.

    Creates a directed graph showing the structure and data flow of a procedure
    workflow. Each procedure becomes a node showing its type, label, and settings.
    Edges show data flow (inputs) and control flow (sequential steps).

    Args:
        procedure (Procedure): The root procedure to visualize (any procedure in the DAG)
        group_by_depth (bool): Whether to visually group nodes by their depth level
                              in the procedure hierarchy for better layout

    Returns:
        A Graphviz Digraph object that can be rendered to various formats
        (PNG, SVG, PDF, etc.)

    Example:
        ```python
        import capturegraph.procedures as cgp

        # Create your procedure
        my_procedure = my_capture_procedure()

        # Generate visualization
        graph = cgp.procedure_to_graphviz(my_procedure, group_by_depth=True)

        # Render to file
        graph.render('my_procedure', format='png', cleanup=True)

        # Or view directly (if supported)
        graph.view()
        ```
    """

    def graphviz_html(text: str) -> str:
        text = "".join(line.strip() for line in text.splitlines() if line.strip())

        return f"<{text}>"

    def shorten_value(value: str) -> str:
        value = f"{value}"
        if len(value) > 20:
            return value[:20] + "..."
        else:
            return value

    def list_item(name: str, value: str) -> str:
        return f"""<tr>
        <td align="left"><font point-size="11" face="Courier New">
            <b> • {name}:</b>
        </font></td>
        <td align="left"><font point-size="11">
            {shorten_value(value)}
        </font></td>
        </tr>"""

    def procedure_to_graph_label(procedure: Procedure) -> str:
        return graphviz_html(f"""
        <table border = "0" cellborder = "0" cellspacing = "0" align = "left">
        <tr>
            <td colspan = "2"> <font point-size = "14">
                <b> {procedure.label or procedure._kind.__name__} </b>
            </font> </td>
        </tr>
        {list_item("kind", procedure._kind.__name__) if procedure.label else ""}
        {"".join(list_item(name, value) for name, value in procedure._settings.items())}
        </table>
        """)

    def procedure_input_to_edge_label(name: str, return_type: str) -> str:
        return graphviz_html(f"""
        <table border = "1" cellborder = "0" cellspacing = "0" bgcolor = "white">
        <tr>
        <td>
            <font face = "Courier New" point-size = "10">
            <b> {name} </b>
            </font>
        </td>
        </tr>
        </table>
        """)

    def procedure_step_to_edge_label(number: int) -> str:
        return graphviz_html(f"""
        <table border = "1" cellborder = "0" cellspacing = "0" bgcolor = "white">
        <tr>
        <td align = "left">
            <font face = "Courier New" point-size = "11">
            <b> Step  # {number}</b>
            </font>
        </td>
        </tr>
        </table>
        """)

    node_depth: dict[str, int] = {}
    dot = Digraph(
        # 1) Default node style
        node_attr={
            "shape": "box",
            "style": "rounded,filled",
            "fillcolor": "lightgrey",
            "fontname": "Helvetica",
        },
        # 2) Global graph settings
        graph_attr={
            "rankdir": "TB",  # top->bottom tree
            "ranksep": "1.0",  # vertical spacing between ranks
            "nodesep": "0.1",  # vertical spacing between ranks
        },
    )

    def add_procedure(procedure: Procedure, depth: int):
        if procedure._uuid in node_depth:
            node_depth[procedure._uuid] = max(node_depth[procedure._uuid], depth)
            return
        else:
            node_depth[procedure._uuid] = depth

        dot.node(
            procedure._uuid, label=procedure_to_graph_label(procedure), labeljust="l"
        )

        for name, input_procedure in procedure._inputs.items():
            add_procedure(input_procedure, depth + 1)

            dot.edge(
                f"{input_procedure._uuid}",
                f"{procedure._uuid}",
                xlabel=procedure_input_to_edge_label(
                    name, input_procedure._return_type.__name__
                ),
            )

        for step, substep_procedure in enumerate(procedure._substeps):
            add_procedure(substep_procedure, depth + 1)

            dot.edge(
                f"{substep_procedure._uuid}",
                f"{procedure._uuid}",
                xlabel=procedure_step_to_edge_label(step + 1),
            )

    add_procedure(procedure, 0)

    if group_by_depth:
        for depth in sorted({d for _, d in node_depth.items()}):
            nodes = [uuid for uuid, d in node_depth.items() if d == depth]

            with dot.subgraph(name=f"cluster_{depth}") as s:  # type: ignore
                s.attr(style="invis")
                s.attr(rank="same")

                for uuid in nodes:
                    s.node(uuid)

    return dot