Dijkstra Shortest Path Finder (python, written by Codex)

Codex 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 (Python)

## Requirements
- Ubuntu 22.04
- Python 3.10+

## Install
```bash
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```

## Run
```bash
python src/main.py graph.csv --source A --destination D
python src/main.py graph.json --source A --all --directed
python src/main.py graph.csv --queries pairs.txt --format json --output shortest_path.json
```

If no input is provided, a sample weighted graph is generated and demonstrated.

# No external dependencies required

#!/usr/bin/env python3
import argparse
import csv
import json
import math
import heapq
from pathlib import Path


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description="Dijkstra shortest path finder")
    p.add_argument("file", nargs="?")
    p.add_argument("--source")
    p.add_argument("--destination")
    p.add_argument("--directed", action="store_true")
    p.add_argument("--all", action="store_true")
    p.add_argument("--format", choices=["text", "json", "dot"], default="text")
    p.add_argument("--queries")
    p.add_argument("--output", default="shortest_path.json")
    return p.parse_args()


def load_graph(path: Path, directed: bool):
    g: dict[str, list[tuple[str, float]]] = {}

    def add(u: str, v: str, w: float):
        g.setdefault(u, []).append((v, w))

    if path.suffix.lower() == ".json":
        obj = json.loads(path.read_text(encoding="utf-8"))
        for u, nbrs in obj.items():
            for v, w in nbrs.items():
                add(u, v, float(w))
                if not directed:
                    add(v, u, float(w))
    else:
        with path.open("r", encoding="utf-8") as f:
            r = csv.DictReader(f)
            for row in r:
                u = row.get("source") or row.get("u") or row.get("from") or row.get("src")
                v = row.get("destination") or row.get("v") or row.get("to") or row.get("dst")
                w = float(row.get("weight") or row.get("w") or 1)
                if u is None or v is None:
                    continue
                add(u, v, w)
                if not directed:
                    add(v, u, w)

    for u, edges in list(g.items()):
        for v, _ in edges:
            g.setdefault(v, [])
    return g


def graph_stats(g, directed):
    n = len(g)
    edges = sum(len(v) for v in g.values())
    neg = sum(1 for es in g.values() for _, w in es if w < 0)
    if not directed:
        edges //= 2

    # connected components via undirected traversal over provided adjacency
    visited = set()
    comps = 0
    for s in g:
        if s in visited:
            continue
        comps += 1
        stack = [s]
        visited.add(s)
        while stack:
            u = stack.pop()
            for v, _ in g.get(u, []):
                if v not in visited:
                    visited.add(v)
                    stack.append(v)

    density = 0.0
    if n > 1:
        density = edges / (n * (n - 1)) if directed else (2 * edges) / (n * (n - 1))

    return {
        "totalNodes": n,
        "totalEdges": edges,
        "averageDegree": (sum(len(v) for v in g.values()) / n) if n else 0,
        "connectedComponents": comps,
        "density": density,
        "negativeEdges": neg,
    }


def dijkstra(g, src):
    dist = {u: math.inf for u in g}
    prev: dict[str, str] = {}
    dist[src] = 0.0
    pq = [(0.0, src)]

    while pq:
        d, u = heapq.heappop(pq)
        if d != dist[u]:
            continue
        for v, w in g[u]:
            nd = d + w
            if nd < dist[v]:
                dist[v] = nd
                prev[v] = u
                heapq.heappush(pq, (nd, v))
    return dist, prev


def build_path(prev, src, dst):
    if src == dst:
        return [src]
    path = []
    cur = dst
    while cur in prev and cur != src:
        path.append(cur)
        cur = prev[cur]
    if cur != src:
        return None
    path.append(src)
    path.reverse()
    return path


def query_one(g, src, dst):
    dist, prev = dijkstra(g, src)
    d = dist.get(dst, math.inf)
    if not math.isfinite(d):
        return {"source": src, "destination": dst, "exists": False, "message": "No path"}
    path = build_path(prev, src, dst)
    return {
        "source": src,
        "destination": dst,
        "exists": True,
        "totalDistance": d,
        "nodeSequence": path,
        "edges": max(0, len(path) - 1) if path else 0,
    }


def format_result(r, mode):
    if mode == "json":
        return json.dumps(r, indent=2)
    if mode == "dot":
        lines = ["digraph G {"]
        if r.get("nodeSequence"):
            for u, v in zip(r["nodeSequence"], r["nodeSequence"][1:]):
                lines.append(f'  "{u}" -> "{v}";')
        lines.append("}")
        return "\n".join(lines)
    if not r.get("exists"):
        return f"No path from {r['source']} to {r['destination']}"
    return f"Distance: {r['totalDistance']}\nPath: {' -> '.join(r['nodeSequence'])}\nEdges: {r['edges']}"


def parse_queries(path: Path):
    out = []
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line:
            continue
        s, d = [x.strip() for x in line.split(",", 1)]
        out.append((s, d))
    return out


def write_sample(path: Path):
    nodes = [f"N{i}" for i in range(1, 16)]
    edges = []
    for i in range(14):
        edges.append((nodes[i], nodes[i + 1], 1 + (i % 5)))
    edges += [
        ("N1", "N5", 4), ("N2", "N8", 3), ("N3", "N10", 6), ("N4", "N12", 2),
        ("N6", "N11", 5), ("N7", "N13", 2), ("N8", "N14", 7), ("N9", "N15", 1),
        ("N5", "N9", 3), ("N10", "N14", 2), ("N11", "N15", 4),
    ]
    with path.open("w", encoding="utf-8", newline="") as f:
        w = csv.writer(f)
        w.writerow(["source", "destination", "weight"])
        w.writerows(edges)


def main() -> int:
    args = parse_args()
    file_path = Path(args.file) if args.file else Path("sample_graph.csv")

    if not args.file:
        write_sample(file_path)
        args.source = args.source or "N1"
        args.destination = args.destination or "N15"

    g = load_graph(file_path, args.directed)
    stats = graph_stats(g, args.directed)
    if stats["negativeEdges"] > 0:
        print("Warning: negative edge weights found; Bellman-Ford is recommended.")

    results = []
    if args.queries:
        for s, d in parse_queries(Path(args.queries)):
            results.append(query_one(g, s, d))
    elif args.all and args.source:
        dist, prev = dijkstra(g, args.source)
        for n in g:
            if math.isfinite(dist[n]):
                p = build_path(prev, args.source, n)
                results.append({"source": args.source, "destination": n, "exists": True, "totalDistance": dist[n], "nodeSequence": p, "edges": len(p) - 1 if p else 0})
            else:
                results.append({"source": args.source, "destination": n, "exists": False, "message": "No path"})
    else:
        if not args.source or not args.destination:
            print("Error: --source and --destination are required unless --all or --queries is used", file=sys.stderr)
            return 1
        results.append(query_one(g, args.source, args.destination))

    for r in results:
        print(format_result(r, args.format))
        if args.format != "json":
            print()

    report = {
        "config": vars(args),
        "graphStats": stats,
        "results": results,
    }
    Path(args.output).write_text(json.dumps(report, indent=2), encoding="utf-8")
    return 0


if __name__ == "__main__":
    import sys
    raise SystemExit(main())