Class WorkflowAnalyzer

preswald/interfaces/workflow.py:604–851

Provides visualization and analysis capabilities for workflow structures. Uses Plotly for interactive visualization and NetworkX for graph algorithms.

class WorkflowAnalyzer:
    """
    Provides visualization and analysis capabilities for workflow structures.
    Uses Plotly for interactive visualization and NetworkX for graph algorithms.
    """

    def __init__(self, workflow):
        self.workflow = workflow
        self.graph = nx.DiGraph()
        self._last_analysis_time = None

        # Define color scheme for different atom statuses
        self.status_colors = {
            "pending": "#E8E8E8",  # Light Gray
            "running": "#72B0DD",  # Blue
            "completed": "#72B7B7",  # Teal
            "failed": "#B76E79",  # Rose
            "retry": "#FFB347",  # Orange
            "skipped": "#D7BDE2",  # Light Purple
            "not_executed": "#C8C8C8",  # Gray
        }

    def build_graph(self) -> nx.DiGraph:
        """
        Constructs a NetworkX graph representation of the workflow.
        Includes rich metadata for visualization and analysis.
        """
        self.graph.clear()

        # Add nodes (atoms) with their metadata
        for atom_name, atom in self.workflow.atoms.items():
            result = self.workflow.context.results.get(atom_name)

            # Prepare node metadata with rich information for tooltips
            node_data = {
                "name": atom_name,
                "status": result.status.value if result else "not_executed",
                "execution_time": (
                    f"{result.execution_time:.2f}s"
                    if result and result.execution_time
                    else "N/A"
                ),
                "attempts": result.attempts if result else 0,
                "error": str(result.error) if result and result.error else None,
                "dependencies": list(atom.dependencies),
                "force_recompute": atom.force_recompute,
            }

            self.graph.add_node(atom_name, **node_data)

            # Add edges for dependencies
            for dep in atom.dependencies:
                self.graph.add_edge(dep, atom_name)

        self._last_analysis_time = datetime.now()
        return self.graph

    def get_critical_path(self) -> list[str]: