Dijkstra Shortest Path Finder (python, written by Claude Code)

Claude Code wrote this python project from the task below. It installed and ran on a clean Ubuntu 22.04 machine as written.

Task given to the agent:

TASK: Dijkstra Shortest Path Finder

Write a program that finds the shortest path between nodes in a weighted graph using Dijkstra's algorithm, supporting multiple graph input formats, path visualization, and batch queries.

FUNCTIONAL REQUIREMENTS:
- Accept a graph definition file as a command-line argument (support adjacency list in JSON and edge list in CSV format)
- Accept source and destination nodes via --source and --destination flags
- Implement Dijkstra's algorithm with a priority queue (min-heap) for optimal performance
- Support both directed and undirected graphs via --directed flag (default: undirected)
- Report for the shortest path: total distance/weight, the complete node sequence, and the number of edges
- Support finding shortest paths from one source to all other nodes via --all flag (single-source shortest path)
- Support negative edge weight detection: warn if negative weights are found (Dijkstra doesn't handle them correctly) and suggest using Bellman-Ford instead
- Display the path in multiple formats via --format flag: text (default, showing node sequence with edge weights), json (structured output), and dot (Graphviz DOT format for visualization)
- Support batch queries: read multiple source-destination pairs from a file via --queries flag and compute shortest paths for all pairs
- Print graph statistics: total nodes, total edges, average degree, connected components count, and graph density
- Save results as JSON with --output flag (default: shortest_path.json)
- If no input is given, generate a sample weighted graph with 15 nodes and 25 edges, find shortest paths between several pairs of nodes, and demonstrate the all-pairs output
- Handle errors: disconnected nodes (no path exists), self-loops, duplicate edges, invalid node references, and malformed graph files

Create a complete Python project for a clean Ubuntu 22.04 machine with only Python 3.10+ installed. Include:
- Source code
- requirements.txt with all dependencies (direct and transitive) pinned to exact versions
- README.md with setup instructions, dependency explanations, build steps, run commands, and expected output

[]

"""
Dijkstra Shortest Path Finder

Finds shortest paths in weighted graphs using Dijkstra's algorithm with
priority queue optimization. Supports batch queries and graph visualization.

Dependencies:
    - networkx (3.2.1): Graph data structures and algorithms
    - matplotlib (3.8.2): Graph visualization
"""

import heapq
import json
import sys
from typing import Any, Optional

import matplotlib.pyplot as plt
import networkx as nx


class DijkstraPathFinder:
    """Finds shortest paths in weighted graphs using Dijkstra's algorithm."""

    def __init__(self):
        """Initialize the path finder with an empty graph."""
        self.graph = nx.DiGraph()

    def add_edge(self, source: str, target: str, weight: float) -> None:
        """Add a weighted directed edge to the graph.

        Args:
            source: The source node identifier.
            target: The target node identifier.
            weight: The edge weight (must be non-negative).

        Raises:
            ValueError: If weight is negative.
        """
        if weight < 0:
            raise ValueError(f"Edge weight must be non-negative, got {weight}")
        self.graph.add_edge(source, target, weight=weight)

    def add_undirected_edge(self, node_a: str, node_b: str, weight: float) -> None:
        """Add a weighted undirected edge (both directions).

        Args:
            node_a: First node identifier.
            node_b: Second node identifier.
            weight: The edge weight (must be non-negative).
        """
        self.add_edge(node_a, node_b, weight)
        self.add_edge(node_b, node_a, weight)

    def load_from_dict(self, graph_data: dict) -> None:
        """Load a graph from a dictionary representation.

        Args:
            graph_data: Dictionary with 'nodes' and 'edges' keys.
                edges should be list of dicts with 'source', 'target', 'weight'.
                Optionally include 'directed' (default True).
        """
        directed = graph_data.get("directed", True)

        for node in graph_data.get("nodes", []):
            if isinstance(node, dict):
                self.graph.add_node(node["id"], **{k: v for k, v in node.items() if k != "id"})
            else:
                self.graph.add_node(node)

        for edge in graph_data.get("edges", []):
            source = str(edge["source"])
            target = str(edge["target"])
            weight = float(edge["weight"])
            if directed:
                self.add_edge(source, target, weight)
            else:
                self.add_undirected_edge(source, target, weight)

    def load_from_json(self, filepath: str) -> None:
        """Load a graph from a JSON file.

        Args:
            filepath: Path to the JSON file.
        """
        with open(filepath, "r") as f:
            data = json.load(f)
        self.load_from_dict(data)

    def dijkstra(self, source: str, target: Optional[str] = None) -> dict:
        """Run Dijkstra's algorithm from a source node.

        Uses a priority queue (min-heap) for efficient extraction of the
        node with the smallest tentative distance.

        Args:
            source: The source node to start from.
            target: Optional target node. If provided, the algorithm stops
                    early once the target is reached.

        Returns:
            Dictionary with keys:
                - 'distances': dict mapping node -> shortest distance from source
                - 'predecessors': dict mapping node -> previous node on shortest path
                - 'target_distance': distance to target (if target specified)
                - 'target_path': list of nodes on shortest path to target (if reachable)

        Raises:
            ValueError: If source node is not in the graph.
        """
        if source not in self.graph:
            raise ValueError(f"Source node '{source}' not found in graph")
        if target is not None and target not in self.graph:
            raise ValueError(f"Target node '{target}' not found in graph")

        distances = {node: float("inf") for node in self.graph.nodes()}
        distances[source] = 0.0
        predecessors = {node: None for node in self.graph.nodes()}
        visited = set()

        # Priority queue: (distance, node)
        pq = [(0.0, source)]

        while pq:
            current_dist, current_node = heapq.heappop(pq)

            if current_node in visited:
                continue
            visited.add(current_node)

            # Early termination if we reached the target
            if target is not None and current_node == target:
                break

            for neighbor in self.graph.successors(current_node):
                if neighbor in visited:
                    continue
                edge_weight = self.graph[current_node][neighbor]["weight"]
                new_dist = current_dist + edge_weight

                if new_dist < distances[neighbor]:
                    distances[neighbor] = new_dist
                    predecessors[neighbor] = current_node
                    heapq.heappush(pq, (new_dist, neighbor))

        result = {
            "distances": distances,
            "predecessors": predecessors,
        }

        if target is not None:
            result["target_distance"] = distances.get(target, float("inf"))
            result["target_path"] = self._reconstruct_path(predecessors, source, target)

        return result

    def _reconstruct_path(
        self, predecessors: dict, source: str, target: str
    ) -> Optional[list]:
        """Reconstruct the shortest path from predecessors map.

        Args:
            predecessors: Dictionary mapping each node to its predecessor.
            source: The source node.
            target: The target node.

        Returns:
            List of nodes from source to target, or None if unreachable.
        """
        if predecessors.get(target) is None and target != source:
            return None

        path = []
        current = target
        while current is not None:
            path.append(current)
            current = predecessors.get(current)
        path.reverse()
        return path

    def batch_shortest_paths(self, queries: list) -> list:
        """Process multiple shortest path queries in batch.

        Args:
            queries: List of dicts, each with 'source' and 'target' keys.

        Returns:
            List of result dicts, each containing:
                - 'source': source node
                - 'target': target node
                - 'distance': shortest distance (inf if unreachable)
                - 'path': list of nodes on the path (None if unreachable)
        """
        results = []
        # Cache: source -> dijkstra result (reuse when same source appears)
        cache = {}

        for query in queries:
            source = str(query["source"])
            target = str(query["target"])

            if source not in cache:
                cache[source] = self.dijkstra(source)

            dijkstra_result = cache[source]
            distance = dijkstra_result["distances"].get(target, float("inf"))
            path = self._reconstruct_path(
                dijkstra_result["predecessors"], source, target
            )

            results.append(
                {
                    "source": source,
                    "target": target,
                    "distance": distance,
                    "path": path,
                }
            )

        return results

    def visualize(
        self,
        highlight_path: Optional[list] = None,
        output_file: Optional[str] = None,
        title: str = "Dijkstra Shortest Path",
    ) -> None:
        """Visualize the graph with optional path highlighting.

        Args:
            highlight_path: List of nodes forming a path to highlight.
            output_file: If provided, save the figure to this file path.
            title: Title for the plot.
        """
        fig, ax = plt.subplots(1, 1, figsize=(12, 8))

        pos = nx.spring_layout(self.graph, seed=42, k=2.0)

        # Draw all edges
        edge_colors = []
        edge_widths = []
        highlight_edges = set()

        if highlight_path and len(highlight_path) > 1:
            for i in range(len(highlight_path) - 1):
                highlight_edges.add((highlight_path[i], highlight_path[i + 1]))

        for edge in self.graph.edges():
            if edge in highlight_edges:
                edge_colors.append("#e74c3c")
                edge_widths.append(3.0)
            else:
                edge_colors.append("#95a5a6")
                edge_widths.append(1.0)

        nx.draw_edges = nx.draw_networkx_edges(
            self.graph, pos, edge_color=edge_colors, width=edge_widths,
            arrows=True, arrowsize=15, ax=ax, connectionstyle="arc3,rad=0.1"
        )

        # Draw nodes
        node_colors = []
        if highlight_path:
            path_set = set(highlight_path)
            for node in self.graph.nodes():
                if node == highlight_path[0]:
                    node_colors.append("#2ecc71")  # Green for source
                elif node == highlight_path[-1]:
                    node_colors.append("#e74c3c")  # Red for target
                elif node in path_set:
                    node_colors.append("#f39c12")  # Orange for path nodes
                else:
                    node_colors.append("#3498db")  # Blue for others
        else:
            node_colors = ["#3498db"] * len(self.graph.nodes())

        nx.draw_networkx_nodes(
            self.graph, pos, node_color=node_colors, node_size=600, ax=ax
        )
        nx.draw_networkx_labels(
            self.graph, pos, font_size=10, font_weight="bold", ax=ax
        )

        # Draw edge weight labels
        edge_labels = nx.get_edge_attributes(self.graph, "weight")
        nx.draw_networkx_edge_labels(
            self.graph, pos, edge_labels=edge_labels, font_size=8, ax=ax
        )

        ax.set_title(title, fontsize=14, fontweight="bold")
        plt.tight_layout()

        if output_file:
            plt.savefig(output_file, dpi=150, bbox_inches="tight")
            print(f"Graph saved to {output_file}")
        else:
            plt.show()

        plt.close(fig)

    def get_graph_stats(self) -> dict:
        """Return statistics about the graph.

        Returns:
            Dictionary with graph statistics.
        """
        return {
            "num_nodes": self.graph.number_of_nodes(),
            "num_edges": self.graph.number_of_edges(),
            "is_directed": self.graph.is_directed(),
            "nodes": list(self.graph.nodes()),
            "density": nx.density(self.graph),
        }


def create_sample_graph() -> dict:
    """Create a sample weighted graph for demonstration."""
    return {
        "directed": False,
        "nodes": ["A", "B", "C", "D", "E", "F", "G"],
        "edges": [
            {"source": "A", "target": "B", "weight": 4},
            {"source": "A", "target": "C", "weight": 2},
            {"source": "B", "target": "D", "weight": 5},
            {"source": "B", "target": "C", "weight": 1},
            {"source": "C", "target": "D", "weight": 8},
            {"source": "C", "target": "E", "weight": 10},
            {"source": "D", "target": "E", "weight": 2},
            {"source": "D", "target": "F", "weight": 6},
            {"source": "E", "target": "F", "weight": 3},
            {"source": "E", "target": "G", "weight": 1},
            {"source": "F", "target": "G", "weight": 7},
        ],
    }


def main():
    """Run the Dijkstra shortest path finder demonstration."""
    print("=" * 60)
    print("  Dijkstra Shortest Path Finder")
    print("=" * 60)

    # Create path finder and load sample graph
    finder = DijkstraPathFinder()
    sample_graph = create_sample_graph()
    finder.load_from_dict(sample_graph)

    # Display graph stats
    stats = finder.get_graph_stats()
    print(f"\nGraph Statistics:")
    print(f"  Nodes: {stats['num_nodes']}")
    print(f"  Edges: {stats['num_edges']}")
    print(f"  Directed: {stats['is_directed']}")
    print(f"  Density: {stats['density']:.4f}")
    print(f"  Vertices: {', '.join(stats['nodes'])}")

    # Single query
    print(f"\n{'─' * 60}")
    print("Single Query: A -> G")
    print(f"{'─' * 60}")
    result = finder.dijkstra("A", "G")
    path = result["target_path"]
    distance = result["target_distance"]
    print(f"  Shortest distance: {distance}")
    print(f"  Path: {' -> '.join(path) if path else 'No path found'}")

    # All distances from A
    print(f"\n{'─' * 60}")
    print("All Distances from A:")
    print(f"{'─' * 60}")
    all_result = finder.dijkstra("A")
    for node, dist in sorted(all_result["distances"].items()):
        path = finder._reconstruct_path(all_result["predecessors"], "A", node)
        path_str = " -> ".join(path) if path else "N/A"
        print(f"  A -> {node}: distance = {dist:>5}, path = {path_str}")

    # Batch queries
    print(f"\n{'─' * 60}")
    print("Batch Queries:")
    print(f"{'─' * 60}")
    queries = [
        {"source": "A", "target": "G"},
        {"source": "A", "target": "F"},
        {"source": "B", "target": "G"},
        {"source": "C", "target": "F"},
        {"source": "D", "target": "A"},
        {"source": "G", "target": "A"},
    ]

    batch_results = finder.batch_shortest_paths(queries)
    for r in batch_results:
        path_str = " -> ".join(r["path"]) if r["path"] else "No path"
        dist_str = f"{r['distance']}" if r["distance"] != float("inf") else "unreachable"
        print(f"  {r['source']} -> {r['target']}: {dist_str:>12}  |  {path_str}")

    # Visualize with the A->G shortest path highlighted
    print(f"\n{'─' * 60}")
    print("Generating visualization...")
    finder.visualize(
        highlight_path=path,
        output_file="dijkstra_shortest_path.png",
        title="Dijkstra: Shortest Path from A to G",
    )

    print("\nDone!")


if __name__ == "__main__":
    main()

networkx==3.2.1
matplotlib==3.8.2