"""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 rpmroute.board import build_board from rpmroute.board.search_tree import ShapeSearchTree from rpmroute.dsn import parse_dsn from rpmroute.geometry.octagon import overlaps_2d from rpmroute.route import grid_router from rpmroute.route.exact_router import route_board_exact from rpmroute.route.grid_router import GridRouter, _ends_of from rpmroute.route.pipeline import ( _mm_to_board, _net_widths_board, _rule_clearance_dsn, _rule_width_dsn, _via_drill_board, _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 _capture_conflicts() -> list: """Record each exact-check refusal: the copper box, its kind, and the shape it came too close to. The check stops at a connection's first conflict, so refusals and rejected connections line up in order.""" log: list = [] original = ShapeSearchTree.clearance_conflict def conflict(self, box, net_no, layer, clearance, kind="wire"): hit = original(self, box, net_no, layer, clearance, kind) if hit: # the shape that tripped the check, found the way the check does other = None expanded = box.offset(clearance) for s in self.overlapping(expanded): if s.net_no == net_no or not s.on_layer(layer): continue if s.blocks != "all" and s.blocks != kind + "s": continue if not s.clearance: if s.tile.overlaps(box): other = s break continue if (s.exact is not None and overlaps_2d(s.exact, expanded)) or ( s.exact is None and s.tile.overlaps(expanded) ): other = s break log.append((net_no, kind, layer, box, clearance, other)) return hit ShapeSearchTree.clearance_conflict = conflict return log def _wall_name(wall, names) -> str: if isinstance(wall, str): if wall.startswith("fixed copper of net "): net = int(wall.rsplit(" ", 1)[1]) return f"fixed copper of {names.get(net, net)}" return wall return names.get(wall, "pads/keepouts") def _gap(a, b) -> int: dx = max(b.ll.x - a.ur.x, a.ll.x - b.ur.x, 0) dy = max(b.ll.y - a.ur.y, a.ll.y - b.ur.y, 0) return max(dx, dy) def _fresh(seen) -> GridRouter: router = grid_router.GridRouter(*seen["args"], **seen["kwargs"]) router._conns = router._build_connections() return router def _static_wall(router: GridRouter, node) -> str: """What static copper closes a node: a keepout, copper already on the board (a fixed via or track, by net), or a pad.""" if node in router._keepout: return "keepout" fixed = getattr(router, "_diag_fixed", None) if fixed is None: fixed = router._diag_fixed = {} margin = router.clearance + router.half_width for net, layers, box in router.board.get_fixed_copper(): for cell in router._cells_in_box(box, margin): for layer in layers: fixed[(*cell, layer)] = net net = fixed.get(node) if net is not None: return f"fixed copper of net {net}" return "pads" 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[_static_wall(router, n)] += 1 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, access_cost: float = 0.0, hole_to_hole_mm: float = 0.25, seed: int | None = None, negotiate: int = 0, ) -> 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() conflicts = _capture_conflicts() 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, access_cost=access_cost, seed=seed, negotiate=negotiate, # the product's configuration: net-class widths and hole spacing net_widths=_net_widths_board(dsn, scale), via_drill=_via_drill_board(dsn, scale), hole_to_hole=_mm_to_board(dsn, scale, hole_to_hole_mm) or 0, ) 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) # pins on turned pads leave along their pad's axis: replay with the same exits state._axis_exits = dict(routed._axis_exits) 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) # refusals in the order the check made them, one per rejected connection per_net: dict[int, list] = {} for net_no, kind, layer, box, clearance, other in conflicts: if other is None: what = "?" elif other.net_no == -1: what = "keepout" if other.clearance else "rule area" else: what = ("trace/via of " if other.routed else "pad of ") + names.get(other.net_no, "?") need = clearance if other is None or other.clearance else 0 short = (need - _gap(box, other.tile)) / scale / 1000 if other is not None else 0 per_net.setdefault(net_no, []).append( {"copper": kind, "layer": layer, "against": what, "short_mm": round(short, 4)} ) 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" if per_net.get(net_no): row["conflict"] = per_net[net_no].pop(0) 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": [ _wall_name(n, names) 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("--access-cost", type=float, default=0.0) ap.add_argument("--seed", type=int, help="route in this shuffled order, as bench.py does") ap.add_argument("--negotiate", type=int, default=0, metavar="N", help="as the router's --negotiate") ap.add_argument("--hole-to-hole", type=float, default=0.25, metavar="MM") 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, args.access_cost, args.hole_to_hole, args.seed, args.negotiate) 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'])}]" if "conflict" in row: c = row["conflict"] extra += (f" [{c['copper']} on layer {c['layer']} {c['short_mm']} mm too close" f" to {c['against']}]") 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())