yaml_workflow.visualize

yaml_workflow.visualize

Workflow visualization utilities.

Functions

generate_mermaid(workflow: dict, flow: Optional[str] = None) -> str

Generate a Mermaid diagram string from a parsed workflow dict.

Parameters:

Name	Type	Description	Default
workflow	dict	Parsed workflow dictionary containing 'steps' and optionally 'flows'.	required
flow	Optional[str]	Optional flow name to determine step ordering.	None

Returns:

Type	Description
str	A Mermaid graph TD diagram string.

Source code in src/yaml_workflow/visualize.py

def generate_mermaid(workflow: dict, flow: Optional[str] = None) -> str:
    """Generate a Mermaid diagram string from a parsed workflow dict.

    Args:
        workflow: Parsed workflow dictionary containing 'steps' and optionally 'flows'.
        flow: Optional flow name to determine step ordering.

    Returns:
        A Mermaid graph TD diagram string.
    """
    # Handle both top-level and nested workflow formats
    if "workflow" in workflow:
        workflow = workflow["workflow"]

    workflow_name = workflow.get("name", "Workflow")
    steps = workflow.get("steps", [])

    # Build a lookup from step name to step dict
    step_map = {step["name"]: step for step in steps}

    # Determine the ordered list of step names
    ordered_names = _get_ordered_step_names(workflow, steps, flow)

    lines = []
    lines.append(f"%% {workflow_name}")
    lines.append("graph TD")

    # Generate node definitions
    for step_name in ordered_names:
        step = step_map.get(step_name)
        if step is None:
            continue
        task_type = step.get("task", "unknown")
        label = f"{step_name}<br/><small>{task_type}</small>"
        if "condition" in step:
            lines.append(f'    {step_name}{{"{label}"}}')
        else:
            lines.append(f'    {step_name}["{label}"]')

    # Generate sequential edges
    for i in range(len(ordered_names) - 1):
        current = ordered_names[i]
        next_step = ordered_names[i + 1]
        # Only add edge if both steps exist in step_map
        if current in step_map and next_step in step_map:
            lines.append(f"    {current} --> {next_step}")

    # Generate error edges
    for step_name in ordered_names:
        step = step_map.get(step_name)
        if step is None:
            continue
        on_error = step.get("on_error")
        if isinstance(on_error, dict):
            error_next = on_error.get("next")
            if error_next and error_next in step_map:
                lines.append(f"    {step_name} -.->|error| {error_next}")

    return "\n".join(lines)

generate_text(workflow: dict, flow: Optional[str] = None) -> str

Generate an ASCII DAG representation of a workflow.

Uses unicode box-drawing for regular steps and diamond shapes for conditional steps. Adjacent conditional steps are grouped as branches.

Parameters:

Name	Type	Description	Default
workflow	dict	Parsed workflow dictionary containing 'steps' and optionally 'flows'.	required
flow	Optional[str]	Optional flow name to determine step ordering.	None

Returns:

Type	Description
str	An ASCII text diagram string.

Source code in src/yaml_workflow/visualize.py

def generate_text(workflow: dict, flow: Optional[str] = None) -> str:
    """Generate an ASCII DAG representation of a workflow.

    Uses unicode box-drawing for regular steps and diamond shapes for
    conditional steps. Adjacent conditional steps are grouped as branches.

    Args:
        workflow: Parsed workflow dictionary containing 'steps' and optionally 'flows'.
        flow: Optional flow name to determine step ordering.

    Returns:
        An ASCII text diagram string.
    """
    # Handle both top-level and nested workflow formats
    if "workflow" in workflow:
        workflow = workflow["workflow"]

    workflow_name = workflow.get("name", "Workflow")
    steps = workflow.get("steps", [])
    step_map: Dict[str, dict] = {step["name"]: step for step in steps}
    ordered_names = _get_ordered_step_names(workflow, steps, flow)

    # Collect error edges for display
    error_edges: List[tuple] = []
    for name in ordered_names:
        step = step_map.get(name)
        if step:
            on_error = step.get("on_error")
            if isinstance(on_error, dict) and on_error.get("next"):
                error_edges.append((name, on_error["next"]))

    # Group steps into segments: sequential steps and branch groups
    segments = _group_into_segments(ordered_names, step_map)

    # Calculate widths
    max_name = max((len(n) for n in ordered_names), default=8)
    max_task = max(
        (len(step_map.get(n, {}).get("task", "")) for n in ordered_names), default=4
    )
    box_w = max(max_name, max_task, 10)  # inner content width

    lines: List[str] = []
    lines.append(f"  Workflow: {workflow_name}")
    if flow:
        lines.append(f"  Flow: {flow}")
    lines.append("")

    mid = box_w // 2 + 4

    for seg_i, segment in enumerate(segments):
        if segment["type"] == "step":
            name = segment["name"]
            step = step_map.get(name)
            if not step:
                continue
            task_type = step.get("task", "unknown")
            step_errors = [target for src, target in error_edges if src == name]
            _render_box(lines, name, task_type, box_w, step_errors)
        elif segment["type"] == "branch":
            _render_branch_group(lines, segment["steps"], step_map, error_edges, box_w)

        # Draw connector to next segment
        if seg_i < len(segments) - 1:
            lines.append(" " * mid + "\u2502")
            lines.append(" " * mid + "\u25bc")

    # Summary
    lines.append("")
    total = len(ordered_names)
    conditional = sum(1 for n in ordered_names if "condition" in step_map.get(n, {}))
    lines.append(
        f"  {total} steps ({conditional} conditional, {total - conditional} always-run)"
    )
    if error_edges:
        lines.append(
            f"  {len(error_edges)} error path(s): "
            + ", ".join(f"{s} \u2192 {t}" for s, t in error_edges)
        )

    return "\n".join(lines)