Source code for labscheduler.dev_tools.visualization

import graphviz

from labscheduler.structures import MoveOperation, Operation




def visualize_workflow(operations: list[Operation]) -> None:
    """
    Visualizes a directed graph of operations using Graphviz.

    Node properties:

      - Color:
          • Red if both start and finish are None.
          • Yellow if start is not None but finish is None.
          • Green if neither start nor finish are None.
      - Label:
          • The first line is taken from operation.main_machine.preferred if it exists;
            otherwise, operation.main_machine.type is used.
          • The second line shows the duration.
          • If no main_machine is provided, the operation's name is used as a fallback.

    Edge properties:

      - For each edge from a preceding operation to an operation, the label is constructed from:
          • "Min:" from min_wait if the value exists and is non-zero.
          • "Max:" from max_wait if the value exists and is not infinity.
          • "Cost:" from wait_cost if the value exists.

        The three parts are concatenated together with commas. If none of the values are relevant,
        the edge is unlabeled.

    The graph is oriented top-to-bottom (TB) and is saved locally as "workflow_graph.png",
    which will be overwritten each time. The image is opened in the default viewer.
    """
    dot = graphviz.Digraph(comment="Workflow Graph", format="png")
    dot.attr(rankdir="TB")

    # Create nodes with updated labels and coloring.
    for op in operations:
        # Determine node color.
        if op.start is None and op.finish is None:
            fillcolor = "red"
        elif op.finish is None:
            fillcolor = "yellow"
        else:
            fillcolor = "green"

        # Determine label from main_machine if available.
        main_machine = op.main_machine
        primary = main_machine.preferred if main_machine.preferred is not None else main_machine.type

        if isinstance(op, MoveOperation):
            primary = f"{op.origin_machine.preferred}--{primary}-->{op.target_machine.preferred}"
        # Add duration on the second line.
        label = f"{primary}\nDuration: {op.duration}"
        dot.node(op.name, label=label, style="filled", fillcolor=fillcolor)

    # Create edges from each operations preceding_operations.
    for op in operations:
        for predecessor in op.preceding_operations:
            label_parts = []
            # Add min_wait label if relevant.
            if predecessor in op.min_wait and op.min_wait[predecessor] != 0:
                label_parts.append(f"Min={op.min_wait[predecessor]}")
            # Add max_wait label if relevant.
            if predecessor in op.max_wait:
                max_val = op.max_wait[predecessor]
                if max_val != float("inf") and max_val != "inf":
                    label_parts.append(f"Max={max_val}")
            # Add wait_cost label if available.
            if predecessor in op.wait_cost:
                label_parts.append(f"C={round(op.wait_cost[predecessor])}")

            edge_label = ", ".join(label_parts) if label_parts else ""
            dot.edge(predecessor, op.name, label=edge_label)

    # Render to a local file "workflow_graph" (overwritten each time).
    dot.render("workflow_graph", view=True, cleanup=True)