bench/diagnose.py: sort an exact-engine run's failures by cause

rejected by the exact check, unroutable even alone, walled in (a pin's
free region is closed off by other copper, with the nets that close it),
or congestion. The board state is the best pass's, rebuilt by replaying
its paths.

bench/diagnose.py

@@ -0,0 +1,198 @@
+"""Why did each connection fail? Sort the failures of one exact-engine run.
+
+    uv run python bench/diagnose.py board.dsn --edge-clearance 0.5 [--json out.json]
+
+Every connection the exact engine leaves unrouted lands in one bucket:
+
+rejected    the grid routed it, the exact geometry check refused its copper
+alone       it cannot be routed even on a board with no other routes: the
+            pads, keepouts or the grid itself are the limit, not congestion
+walled-in   in the routed board one of its pins has no way out: a flood from
+            the pin's end cells through free cells stops short (pin access)
+congestion  both pins can get out, but no free path joins them
+
+The board state is the one the result came from (the best pass), rebuilt by
+replaying that pass's paths. Diagnostic only: it reaches into the router.
+"""
+
+from __future__ import annotations
+
+import argparse
+from collections import Counter
+import json
+from pathlib import Path
+import sys
+
+from freeroute.board import build_board
+from freeroute.dsn import parse_dsn
+from freeroute.route import grid_router
+from freeroute.route.exact_router import route_board_exact
+from freeroute.route.grid_router import GridRouter, _ends_of
+from freeroute.route.pipeline import (
+    _mm_to_board,
+    _rule_clearance_dsn,
+    _rule_width_dsn,
+    _via_half_board,
+)
+
+# a pin whose free neighbourhood holds fewer nodes than this is walled in
+FLOOD_LIMIT = 200
+
+
+def _capture_router():
+    """Patch GridRouter so the instance route_board_exact builds, and the paths
+    of its best pass, can be read back afterwards."""
+    seen: dict = {}
+    original_commit = GridRouter._commit
+    original_snapshot = GridRouter._snapshot
+
+    def commit(self, conn, path):
+        original_commit(self, conn, path)
+        self._diag_paths[conn.index] = (list(path), conn.a_end, conn.b_end)
+
+    def snapshot(self, conns):
+        self._diag_best = {c.index: self._diag_paths[c.index] for c in conns if c.routed}
+        seen["router"] = self
+        return original_snapshot(self, conns)
+
+    def init(self, *args, **kwargs):
+        original_init(self, *args, **kwargs)
+        self._diag_paths = {}
+        self._diag_best = {}
+        seen["kwargs"] = kwargs
+        seen["args"] = args
+
+    original_init = GridRouter.__init__
+    GridRouter.__init__ = init
+    GridRouter._commit = commit
+    GridRouter._snapshot = snapshot
+    return seen
+
+
+def _fresh(seen) -> GridRouter:
+    router = grid_router.GridRouter(*seen["args"], **seen["kwargs"])
+    router._conns = router._build_connections()
+    return router
+
+
+def _flood(router: GridRouter, pin, net_no: int, conns) -> tuple[int, Counter]:
+    """How many nodes a pin can reach through free cells (up to FLOOD_LIMIT),
+    and which other nets' copper bounds the region."""
+    walls: Counter = Counter()
+    # the search starts on every end node unconditionally (it may leave one
+    # inside a wire keepout by a via), so the flood does too
+    start = list(router._end_nodes(pin, net_no, conns))
+    seen = set(start)
+    frontier = list(start)
+    while frontier and len(seen) < FLOOD_LIMIT:
+        x, y, layer = frontier.pop()
+        steps = [(x + dx, y + dy, layer) for dx, dy in router._neighbours]
+        if router._via_placeable((x, y), net_no, conns):
+            steps += [(x, y, other) for other in router.layers if other != layer]
+        elif len(seen) == 1:
+            # the pin's own exit is a via that other copper leaves no room for
+            via_nodes = [(*c, lay) for c in router._via_cells((x, y)) for lay in router.layers]
+            for owner in router._crossed_owners(via_nodes, net_no, conns):
+                walls[conns[owner].net_no] += 1
+        for n in steps:
+            if n in seen or not (0 <= n[0] < router.cols and 0 <= n[1] < router.rows):
+                continue
+            if router._enter_cost(n[:2], n[2], net_no, False, conns) is None:
+                # trace, via footprint or clearance halo, as rip-up sees them
+                owners = router._crossed_owners([n], net_no, conns)
+                for owner in owners:
+                    walls[conns[owner].net_no] += 1
+                if not owners:
+                    walls[0] += 1  # a pad or keepout
+                continue
+            seen.add(n)
+            frontier.append(n)
+    return len(seen), walls
+
+
+def diagnose(
+    dsn_text: str, edge_clearance_mm: float | None, resolution: int = 2, shove: bool = False
+) -> dict:
+    dsn = parse_dsn(dsn_text)
+    names = {i + 1: n.name for i, n in enumerate(dsn.nets)}
+    scale = max(dsn.resolution.value, 1)
+    seen = _capture_router()
+    exact = route_board_exact(
+        build_board(dsn),
+        trace_width=round(_rule_width_dsn(dsn) * scale),
+        clearance=round(_rule_clearance_dsn(dsn) * scale),
+        via_half=_via_half_board(dsn, scale),
+        edge_clearance=_mm_to_board(dsn, scale, edge_clearance_mm),
+        resolution=resolution,
+        shove=shove,
+    )
+    routed = seen["router"]
+    best = routed._diag_best
+    grid_failed = {(c.net_no, *_ends_of(c)) for c in routed._conns if c.index not in best}
+
+    # the board as the best pass left it
+    state = _fresh(seen)
+    conns = state._conns
+    for idx, (path, a_end, b_end) in best.items():
+        conns[idx].a_end, conns[idx].b_end = a_end, b_end
+        state._commit(conns[idx], path)
+    empty = _fresh(seen)
+    by_key = {(c.net_no, *_ends_of(c)): c for c in conns}
+
+    rows = []
+    for failure in exact.result.failed:
+        net_no, a, b = failure[0], failure[1], failure[2]
+        zone = failure[3] if len(failure) > 3 else ""
+        if zone:
+            continue  # left to a zone fill, not a failure
+        key = (net_no, a, b)
+        row = {"net": names.get(net_no, str(net_no)), "a": a, "b": b}
+        conn = by_key.get(key)
+        if key not in grid_failed:
+            row["kind"] = "rejected"
+        elif conn is None:
+            row["kind"] = "unknown"
+        elif empty._search(empty._conns[conn.index], allow_ripup=False, conns=empty._conns) is None:
+            row["kind"] = "alone"
+        else:
+            walled = []
+            for end, group in ((a, conn.a_group), (b, conn.b_group)):
+                for pin in group[0][:1]:  # a group's first pin is its route end
+                    reach, walls = _flood(state, pin, net_no, conns)
+                    if reach < FLOOD_LIMIT:
+                        walled.append({
+                            "pin": end, "reach": reach,
+                            "walls": [names.get(n, "pads/keepouts") for n, _ in walls.most_common(4)],
+                        })
+            row["kind"] = "walled-in" if walled else "congestion"
+            if walled:
+                row["walled"] = walled
+        rows.append(row)
+    return {"failures": rows, "counts": dict(Counter(r["kind"] for r in rows))}
+
+
+def main(argv: list[str] | None = None) -> int:
+    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
+    ap.add_argument("dsn", type=Path)
+    ap.add_argument("--edge-clearance", type=float, metavar="MM")
+    ap.add_argument("--grid-resolution", type=int, default=2, metavar="K")
+    ap.add_argument("--shove", action="store_true", help="as mckicad routes")
+    ap.add_argument("--json", type=Path, help="also write the full table here")
+    args = ap.parse_args(argv)
+
+    report = diagnose(args.dsn.read_text(), args.edge_clearance, args.grid_resolution, args.shove)
+    for row in report["failures"]:
+        extra = ""
+        for w in row.get("walled", []):
+            extra += f"  [{w['pin']} reaches {w['reach']} cells, walled by {', '.join(w['walls'])}]"
+        print(f"{row['kind']:11} {row['net']}: {row['a']} to {row['b']}{extra}")
+    total = sum(report["counts"].values())
+    print(f"\n{total} failed: " + ", ".join(f"{k} {v}" for k, v in sorted(report["counts"].items())),
+          file=sys.stderr)
+    if args.json:
+        args.json.write_text(json.dumps(report, indent=1))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())