Compare commits

...
Author SHA1 Message Date
rsp2k f1d0f8df70 Pin access: --access-cost, vias in pads inside wire keepouts, exact stub checks
--access-cost C makes a route pay C to pass where it would close another
pin's way out (its end cells, the via it may need there, and the
clearance around both) while that pin still has connections to route.

A pin inside a wire keepout can only leave by a via; its stub to the
cell centre was track inside the keepout, which the exact check refused.
The via now goes on the pin itself. Pin stubs are also checked exactly
against wire keepouts, as the exact engine will check them.

bench/diagnose.py reports what each refused connection collided with.
2026-09-28 09:15:11 -06:00
rsp2k c44ff272ae 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.
2026-09-28 00:30:05 -06:00
rsp2k 05f31fc828 Merge feat/promote-failures: reroute with failed connections first; --orders 2026-09-27 22:53:47 -06:00
rsp2k 01ca108831 --promote-failures: reroute with the connections that failed moved first
Each round routes every connection that has failed so far ahead of the
rest, which keep their order. The promoted set only grows, so two groups
cannot take turns failing; the best round is kept.
2026-09-27 22:34:06 -06:00
rsp2k 0f5dacbd88 --orders: route several connection orders in parallel, keep the best
The default order is always one of the candidates, so the result is
never worse than a plain run; ties go to the earlier candidate.
2026-09-27 22:14:59 -06:00
rsp2k e67ccc04a0 Keep benchmark board lists and results out of git 2026-09-27 15:07:13 -06:00
rsp2k ed2e4c3fac bench: compare routing variants over several connection orders
Runs every board with every variant under the same seeds and compares
them seed by seed, so a change is judged against the routing-order
noise rather than one lucky or unlucky run.
2026-09-27 15:06:30 -06:00
rsp2k 76c3af94b8 --seed: shuffle the order connections are routed in
The router is deterministic, but its result depends heavily on which
connection routes first. A seed shuffles that order reproducibly, so a
change can be judged over several orders instead of the one it happened
to get.
2026-09-27 15:04:35 -06:00
rsp2k cf8b7ece60 uv.lock: record the CalVer version from pyproject 2026-09-27 15:04:35 -06:00
rsp2k f8804512ac Merge feat/plane-layers: route to inner plane layers with a via 2026-09-27 15:01:38 -06:00
12 changed files with 888 additions and 22 deletions
+4
View File
@@ -20,3 +20,7 @@ htmlcov/
# Editor / OS
.DS_Store
*.swp
# Benchmark inputs and results: board lists name local, often private, designs
/bench/boards*.txt
/bench/results/
+206
View File
@@ -0,0 +1,206 @@
"""Route boards under several connection orders and compare variants fairly.
One routing run says little: which connection routes first moves a result by
ten connections either way on a crowded board. This runs every board with
every variant under the same seeds (``--seed``), so variant A and variant B
face the same order on seed 3, and compares them seed by seed.
uv run python bench/bench.py BOARDS -v base= -v hist1="--history-cost 1" \\
--seeds 6 --jobs 4 --out bench/results/hist.jsonl
BOARDS is a text file, one board per line: a name, a DSN path and any flags
that board always needs, e.g.
sense /path/to/charge-sense.dsn --edge-clearance 0.5
# comments and blank lines are ignored
A variant is ``NAME=FLAGS``, or ``NAME@SRC=FLAGS`` to run the code under SRC
(a worktree's ``src`` directory) rather than the installed package. The first
variant is the baseline. Results are appended to --out as JSON lines, and runs
already in it are skipped, so an interrupted benchmark picks up where it left
off. Exits non-zero if any run crashed or timed out.
"""
from __future__ import annotations
import argparse
from concurrent.futures import ThreadPoolExecutor, as_completed
import json
import math
import os
from pathlib import Path
import re
import shlex
import subprocess
import sys
import tempfile
import threading
import time
NOT_ROUTED = re.compile(r"freeroute: (\d+) connection\(s\) not routed")
TO_ZONE = re.compile(r"freeroute: (\d+) more connection\(s\) left to a zone fill")
def parse_boards(path: Path) -> list[tuple[str, Path, list[str]]]:
boards = []
for line in path.read_text().splitlines():
words = shlex.split(line, comments=True)
if not words:
continue
if len(words) < 2:
raise SystemExit(f"{path}: want 'name path [flags]', got {line!r}")
dsn = Path(words[1]).expanduser()
if not dsn.is_file():
raise SystemExit(f"{path}: no such DSN {dsn}")
boards.append((words[0], dsn, words[2:]))
return boards
def parse_variant(spec: str) -> tuple[str, str | None, list[str]]:
head, sep, flags = spec.partition("=")
if not sep or not head:
raise SystemExit(f"variant {spec!r}: want NAME=FLAGS or NAME@SRC=FLAGS")
name, _, src = head.partition("@")
if src and not (Path(src) / "freeroute").is_dir():
raise SystemExit(f"variant {name}: {src} has no freeroute package")
return name, src or None, shlex.split(flags)
def run_one(board, variant, seed: int, timeout: float) -> dict:
name, dsn, board_flags = board
vname, src, flags = variant
env = dict(os.environ)
if src:
env["PYTHONPATH"] = src + os.pathsep + env.get("PYTHONPATH", "")
with tempfile.TemporaryDirectory() as tmp:
cmd = [
sys.executable, "-m", "freeroute.cli", str(dsn), "-do", f"{tmp}/out.ses",
*board_flags, *flags, "--seed", str(seed),
]
start = time.monotonic()
proc = subprocess.Popen(
cmd, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True
)
expired = threading.Event()
def expire() -> None:
expired.set()
proc.kill()
timer = threading.Timer(timeout, expire)
timer.start()
err = proc.stderr.read()
# reap it here rather than through Popen, to keep its peak memory
_, wstatus, usage = os.wait4(proc.pid, 0)
timer.cancel()
proc.returncode = os.waitstatus_to_exitcode(wstatus)
proc.stderr.close()
status = "timeout" if expired.is_set() else (
"ok" if proc.returncode == 0 else f"exit {proc.returncode}"
)
rss_mb = round(usage.ru_maxrss / 1024)
secs = round(time.monotonic() - start, 1)
missing = NOT_ROUTED.search(err)
zone = TO_ZONE.search(err)
return {
"board": name, "variant": vname, "seed": seed, "status": status,
"unrouted": int(missing.group(1)) if missing else 0,
"to_zone": int(zone.group(1)) if zone else 0,
"secs": secs, "rss_mb": rss_mb,
"tail": err.strip().splitlines()[-1] if status != "ok" and err.strip() else "",
}
def mean_sd(xs: list[float]) -> tuple[float, float]:
m = sum(xs) / len(xs)
sd = math.sqrt(sum((x - m) ** 2 for x in xs) / (len(xs) - 1)) if len(xs) > 1 else 0.0
return m, sd
def report(rows: list[dict], boards, variants) -> None:
ok = {(r["board"], r["variant"], r["seed"]): r for r in rows if r["status"] == "ok"}
base = variants[0][0]
print(f"\nunrouted per board (lower is better), baseline {base!r}\n")
print(f"{'board':14} {'variant':14} {'n':>3} {'mean':>6} {'sd':>5} {'min':>4} {'max':>4}"
f" {'secs':>6} vs baseline, seed by seed")
for bname, _, _ in boards:
for vname, _, _ in variants:
runs = sorted((k[2], r) for k, r in ok.items() if k[:2] == (bname, vname))
if not runs:
print(f"{bname:14} {vname:14} no finished runs")
continue
xs = [r["unrouted"] for _, r in runs]
m, sd = mean_sd(xs)
t, _ = mean_sd([r["secs"] for _, r in runs])
line = (f"{bname:14} {vname:14} {len(xs):>3} {m:>6.1f} {sd:>5.1f} {min(xs):>4}"
f" {max(xs):>4} {t:>6.0f}")
if vname != base:
diffs = [r["unrouted"] - ok[(bname, base, s)]["unrouted"]
for s, r in runs if (bname, base, s) in ok]
if diffs:
dm, dsd = mean_sd(diffs)
se = dsd / math.sqrt(len(diffs)) if len(diffs) > 1 else float("inf")
better = sum(d < 0 for d in diffs)
worse = sum(d > 0 for d in diffs)
# a difference inside two standard errors is not a finding
verdict = ("better" if dm < 0 else "worse") if abs(dm) > 2 * se else "no clear change"
line += (f" {dm:+.1f} ± {se:.1f} ({better} better, {worse} worse,"
f" {len(diffs) - better - worse} same) {verdict}")
print(line)
failed = [r for r in rows if r["status"] != "ok"]
for r in failed:
print(f"FAILED {r['board']} {r['variant']} seed {r['seed']}: {r['status']} {r['tail']}")
def main(argv: list[str] | None = None) -> int:
ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
ap.add_argument("boards", type=Path, help="board list: name, DSN path, flags")
ap.add_argument("-v", "--variant", action="append", required=True,
help="NAME=FLAGS or NAME@SRC=FLAGS; the first is the baseline")
ap.add_argument("--seeds", type=int, default=6, help="orders per board and variant")
ap.add_argument("--first-seed", type=int, default=0)
ap.add_argument("--jobs", type=int, default=4, help="runs at once")
ap.add_argument("--timeout", type=float, default=1800, help="seconds per run")
ap.add_argument("--out", type=Path, required=True, help="JSON lines, appended")
ap.add_argument("--report-only", action="store_true", help="summarise --out, run nothing")
args = ap.parse_args(argv)
boards = parse_boards(args.boards)
variants = [parse_variant(v) for v in args.variant]
if len({v[0] for v in variants}) != len(variants):
raise SystemExit("variant names must differ")
seeds = range(args.first_seed, args.first_seed + args.seeds)
rows: list[dict] = []
if args.out.exists():
rows = [json.loads(line) for line in args.out.read_text().splitlines() if line.strip()]
done = {(r["board"], r["variant"], r["seed"]) for r in rows if r["status"] == "ok"}
todo = [(b, v, s) for b in boards for v in variants for s in seeds
if (b[0], v[0], s) not in done]
if not args.report_only and todo:
args.out.parent.mkdir(parents=True, exist_ok=True)
print(f"{len(todo)} runs, {args.jobs} at a time", file=sys.stderr)
# retried runs replace their earlier failure
rows = [r for r in rows if (r["board"], r["variant"], r["seed"]) in done]
with ThreadPoolExecutor(args.jobs) as pool, args.out.open("w") as out:
for r in rows:
out.write(json.dumps(r) + "\n")
futures = [pool.submit(run_one, b, v, s, args.timeout) for b, v, s in todo]
for n, fut in enumerate(as_completed(futures), 1):
r = fut.result()
rows.append(r)
out.write(json.dumps(r) + "\n")
out.flush()
print(f"[{n}/{len(todo)}] {r['board']} {r['variant']} seed {r['seed']}:"
f" {r['status']} unrouted {r['unrouted']} ({r['secs']}s,"
f" {r['rss_mb']} MB)", file=sys.stderr)
wanted = {(b[0], v[0], s) for b in boards for v in variants for s in seeds}
rows = [r for r in rows if (r["board"], r["variant"], r["seed"]) in wanted]
report(rows, boards, variants)
return 1 if any(r["status"] != "ok" for r in rows) or len(rows) < len(wanted) else 0
if __name__ == "__main__":
raise SystemExit(main())
+253
View File
@@ -0,0 +1,253 @@
"""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.board.search_tree import ShapeSearchTree
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 _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:
expanded = box.offset(clearance)
other = next(
(s for s in self.overlapping(expanded)
if s.net_no != net_no and s.on_layer(layer)
and (s.blocks == "all" or s.blocks == kind + "s")),
None,
)
log.append((net_no, kind, layer, box, clearance, other))
return hit
ShapeSearchTree.clearance_conflict = conflict
return log
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 _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,
access_cost: float = 0.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,
)
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)
# 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"
else:
what = ("trace/via of " if other.routed else "pad of ") + names.get(other.net_no, "?")
short = (clearance - _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": [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("--access-cost", type=float, default=0.0)
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)
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())
+89 -19
View File
@@ -19,6 +19,7 @@ import argparse
import sys
from freeroute.dsn.reader import parse_dsn
from freeroute.route.orders import best_of_orders, promote_failures
from freeroute.route.pipeline import (
build_exact_routing_result,
build_rooms_routing_result,
@@ -29,8 +30,8 @@ from freeroute.ses import write_ses
#: which optional flags each engine actually implements. Anything not listed for
#: an engine is rejected by :func:`_validate` rather than quietly ignored.
ENGINE_OPTIONS: dict[str, frozenset[str]] = {
"grid": frozenset({"rip_up", "max_passes"}),
"exact": frozenset({"shove", "diagonal", "rip_up", "max_passes"}),
"grid": frozenset({"rip_up", "max_passes", "seed", "orders", "promote", "access"}),
"exact": frozenset({"shove", "diagonal", "rip_up", "max_passes", "seed", "orders", "promote", "access"}),
"room": frozenset({"shove", "pack"}),
}
@@ -41,6 +42,10 @@ _FLAG_NAMES = {
"diagonal": "--diagonal",
"rip_up": "--no-rip-up",
"max_passes": "-mp/--max-passes",
"seed": "--seed",
"orders": "--orders",
"promote": "--promote-failures",
"access": "--access-cost",
}
@@ -133,6 +138,40 @@ def build_parser() -> argparse.ArgumentParser:
help="copper to board-edge clearance in mm, e.g. KiCad's min_copper_edge_clearance "
"(the DSN does not carry it); default 0: copper only has to stay inside the outline",
)
parser.add_argument(
"--seed",
type=int,
metavar="N",
help="grid and exact engines: shuffle the order connections are routed in; "
"the same seed gives the same result (default: fewer-pin nets first)",
)
parser.add_argument(
"--orders",
type=int,
metavar="N",
help="grid and exact engines: route N connection orders in parallel and keep the "
"one leaving the fewest connections unrouted (the default order is one of them)",
)
parser.add_argument(
"--promote-failures",
type=int,
metavar="R",
help="grid and exact engines: up to R rounds, each routing the connections that "
"failed so far first; keeps the best round",
)
parser.add_argument(
"--access-cost",
type=float,
metavar="C",
help="grid and exact engines: extra cost for a route to pass where it would close "
"another pin's way out while that pin still has connections to route (default 0: off)",
)
parser.add_argument(
"--jobs",
type=int,
metavar="J",
help="processes for --orders (default: one per order, up to the CPU count)",
)
return parser
@@ -152,6 +191,10 @@ def main(argv: list[str] | None = None) -> int:
("diagonal", args.diagonal),
("rip_up", args.no_rip_up),
("max_passes", args.max_passes is not None),
("seed", args.seed is not None),
("orders", args.orders is not None),
("promote", args.promote_failures is not None),
("access", args.access_cost is not None),
)
if asked
]
@@ -173,29 +216,56 @@ def main(argv: list[str] | None = None) -> int:
rip_up = not args.no_rip_up
max_passes = args.max_passes if args.max_passes is not None else 10
if args.orders is not None and args.orders < 1:
parser.error("--orders must be at least 1")
if args.engine == "exact":
result = build_exact_routing_result(
dsn,
layers=layers,
shove=args.shove,
diagonal=args.diagonal,
rip_up=rip_up,
max_passes=max_passes,
edge_clearance_mm=args.edge_clearance,
resolution=2 if args.grid_resolution is None else args.grid_resolution,
heuristic_weight=args.heuristic_weight,
)
elif args.engine == "room":
options = {
"layers": layers,
"shove": args.shove,
"diagonal": args.diagonal,
"rip_up": rip_up,
"max_passes": max_passes,
"edge_clearance_mm": args.edge_clearance,
"resolution": 2 if args.grid_resolution is None else args.grid_resolution,
"heuristic_weight": args.heuristic_weight,
"access_cost": args.access_cost or 0.0,
}
elif args.engine == "grid":
options = {
"layers": layers,
"rip_up": rip_up,
"max_passes": max_passes,
"edge_clearance_mm": args.edge_clearance,
"resolution": 1 if args.grid_resolution is None else args.grid_resolution,
"access_cost": args.access_cost or 0.0,
}
if args.engine == "room":
result = build_rooms_routing_result(
dsn, layers=layers, shove=args.shove, pack=args.pack,
edge_clearance_mm=args.edge_clearance,
)
else:
result = build_routing_result(
dsn, layers=layers, rip_up=rip_up, max_passes=max_passes,
edge_clearance_mm=args.edge_clearance,
resolution=1 if args.grid_resolution is None else args.grid_resolution,
elif args.promote_failures and args.promote_failures > 1:
if args.orders:
parser.error("--orders and --promote-failures cannot be combined")
result, rounds = promote_failures(
dsn_text, args.engine, options, rounds=args.promote_failures, seed=args.seed
)
counts = ", ".join(str(m) for m, _z in rounds)
print(f"freeroute: {len(rounds)} rounds, unrouted {counts}; kept the best", file=sys.stderr)
elif args.orders and args.orders > 1:
result, kept, tried = best_of_orders(
dsn_text, args.engine, options, orders=args.orders, seed=args.seed, jobs=args.jobs
)
counts = ", ".join(f"{'default' if s is None else s}: {m}" for s, (m, _z) in tried)
print(
f"freeroute: tried {len(tried)} orders, kept "
f"{'the default order' if kept is None else f'--seed {kept}'} (unrouted {counts})",
file=sys.stderr,
)
elif args.engine == "exact":
result = build_exact_routing_result(dsn, seed=args.seed, **options)
else:
result = build_routing_result(dsn, seed=args.seed, **options)
ses_text = write_ses(dsn, result)
missing = [u for u in result.unrouted if not u[3]]
+6
View File
@@ -94,6 +94,9 @@ def route_board_exact(
edge_clearance: int | None = None,
resolution: int = 1,
heuristic_weight: float = 1.0,
seed: int | None = None,
route_first: frozenset | None = None,
access_cost: float = 0.0,
) -> ExactRouteResult:
"""Route ``board`` orthogonally and verify the result with exact geometry.
@@ -122,6 +125,9 @@ def route_board_exact(
edge_clearance=edge_clearance,
resolution=resolution,
heuristic_weight=heuristic_weight,
seed=seed,
route_first=route_first,
access_cost=access_cost,
)
edge = 0 if edge_clearance is None else edge_clearance
outline = board.outline
+112 -1
View File
@@ -28,6 +28,7 @@ from __future__ import annotations
import heapq
import math
import random
from freeroute.board.board import BasicBoard
from freeroute.board.items import ConductionArea, ObstacleArea, Pin
@@ -154,8 +155,23 @@ class GridRouter:
edge_clearance: int | None = None,
resolution: int = 1,
heuristic_weight: float = 1.0,
seed: int | None = None,
route_first: frozenset | None = None,
access_cost: float = 0.0,
) -> None:
self.board = board
# Pin access: while a pin still has connections to route, the cells
# its route would leave by (its end cells, the via it may need there,
# and the clearance around both) cost other nets this much extra, so
# they route round a fine-pitch pin's way out instead of sealing it.
self.access_cost = max(float(access_cost), 0.0)
self._doorways: dict[Node, list[tuple[int, int]]] = {}
self._pending: dict[int, int] = {}
# None keeps the deterministic order; a seed shuffles which connection
# routes first, to measure how much a result owes to the order
self.seed = seed
# connections, as (net_no, one end, other end), to route before the rest
self.route_first = frozenset(route_first or ())
# how far (in cells) an escape stub may run to leave a walled-in pin
self._escape_cells = 12
self._crowded_cache: dict = {}
@@ -281,6 +297,8 @@ class GridRouter:
margin = self.clearance + self.half_width
outline = self.board.outline
self._via_edge_block: set[Cell] = set()
# wire keepouts as boxes, to check a pin stub the way the exact engine will
self._keepout_boxes: dict[int, list[IntBox]] = {}
if outline is not None:
# cells too close to the real outline (rounded corners included):
# for a trace on every layer, and separately for a via's wider copper
@@ -317,6 +335,8 @@ class GridRouter:
if box.is_empty():
continue
# the same distance the exact engine verifies against
if item.blocks_wires:
self._keepout_boxes.setdefault(item.layer, []).append(box)
for cell in self._cells_in_box(box, margin):
if item.blocks_wires:
self._keepout.add((*cell, item.layer))
@@ -466,12 +486,16 @@ class GridRouter:
enter = self._enter_cost(cell, cl, net_no, allow_ripup, conns)
if enter is None:
continue
if self._doorways:
enter += self._access(nxt, net_no)
step = 1.0 if dx == 0 or dy == 0 else math.sqrt(2)
self._relax(
current, nxt, base + step + enter, goal_cell, g_score, came_from, open_heap
)
if len(self.layers) > 1:
g_via = base + self.via_cost
if self._doorways:
g_via += self._access(current, net_no)
for layer in self.layers:
if layer == cl:
continue
@@ -505,6 +529,18 @@ class GridRouter:
]
def _stub_clear(self, pin: Pin, cell: Cell, layer: int, net_no: int, conns) -> bool:
keepouts = self._keepout_boxes.get(layer)
if keepouts:
# the stub is off the grid, so check it exactly, as the exact
# engine will: its copper grown by clearance must miss every keepout
from freeroute.geometry.polyline import Polyline, PolylineShape
end = self._cell_center(cell)
if end != pin.location:
for tile in PolylineShape(Polyline([pin.location, end]), self.half_width).tiles():
grown = tile.offset(self.clearance)
if any(k.overlaps(grown) for k in keepouts):
return False
for c in self._stub_cells(pin.location, cell):
owner = self._cell_owner.get((*c, layer))
if owner is not None and conns[owner].net_no != net_no:
@@ -757,8 +793,19 @@ class GridRouter:
if node not in self._cell_owner:
self._cell_owner[node] = conn.index
conn.trace_cells.append((c, layer))
# A pin inside a wire keepout can only leave by a via. Its stub to the
# cell centre would be track inside the keepout, so the via goes on
# the pin itself (a via in its own pad) and the stub is left out.
in_pad: dict[int, IntPoint] = {}
if len(wire_segs) > 1:
for idx, pin in ((0, conn.a_end), (len(wire_segs) - 1, conn.b_end)):
layer, cells = wire_segs[idx]
if isinstance(pin, Pin) and len(cells) == 1 and (*cells[0], layer) in self._keepout:
in_pad[idx] = pin.location
conn.wires = []
for idx, (layer, cells) in enumerate(wire_segs):
if idx in in_pad:
continue
points = [self._cell_center(c) for c in cells]
if idx == 0 and isinstance(conn.a_end, Pin):
points.insert(0, conn.a_end.location)
@@ -768,7 +815,11 @@ class GridRouter:
if len(points) >= 2:
conn.wires.append((layer, points))
conn.vias = [self._cell_center(c) for c in conn.via_cells]
for idx, location in in_pad.items():
# via k joins wire segments k and k + 1
conn.vias[0 if idx == 0 else idx - 1] = location
conn.routed = True
self._settle(conn, -1)
def _rip(self, conn: _Connection) -> None:
for cell, layer in conn.trace_cells:
@@ -791,6 +842,7 @@ class GridRouter:
conn.vias = []
conn.routed = False
conn.rip_count += 1
self._settle(conn, +1)
def _crossed_owners(self, path: list[Node], net_no: int, conns: list[_Connection]) -> set[int]:
owners: set[int] = set()
@@ -806,6 +858,50 @@ class GridRouter:
owners.add(i)
return owners
# --- pin access ---------------------------------------------------------
@staticmethod
def _group_pins(conn: _Connection):
return [*conn.a_group[0], *conn.b_group[0]]
def _build_doorways(self, conns: list[_Connection]) -> None:
"""Map every cell that would close a pin's way out to that pin."""
spread = [(0, 0), *self._halo_offsets]
done: set[int] = set()
for conn in conns:
for pin in self._group_pins(conn):
self._pending[id(pin)] = self._pending.get(id(pin), 0) + 1
if id(pin) in done:
continue
done.add(id(pin))
claim = (id(pin), conn.net_no)
exits: set[Node] = set()
for x, y, layer in self._end_nodes(pin, conn.net_no, conns):
exits.add((x, y, layer))
for c in self._via_cells((x, y)):
exits.update((*c, other) for other in self.layers)
cells: set[Node] = set()
for x, y, layer in exits:
for dx, dy in spread:
if 0 <= x + dx < self.cols and 0 <= y + dy < self.rows:
cells.add((x + dx, y + dy, layer))
for node in cells:
self._doorways.setdefault(node, []).append(claim)
def _settle(self, conn: _Connection, delta: int) -> None:
if not self._pending:
return
for pin in self._group_pins(conn):
if id(pin) in self._pending:
self._pending[id(pin)] += delta
def _access(self, node: Node, net_no: int) -> float:
"""Extra cost for ``net_no`` to use a cell another pin still needs."""
for pin_id, owner_net in self._doorways.get(node, ()):
if owner_net != net_no and self._pending.get(pin_id, 0) > 0:
return self.access_cost
return 0.0
# --- passes -------------------------------------------------------------
def route(self) -> RouteResult:
@@ -814,14 +910,23 @@ class GridRouter:
self._conns = conns
if not conns:
return RouteResult(self.half_width)
if self.access_cost:
self._build_doorways(conns)
passes = self.max_passes if self.rip_up else 1
best_count = -1
best_snapshot: list[tuple[int, list, list]] | None = None
best_unrouted = len(conns)
# conns[i].index == i is what rip-up relies on, so shuffle a view
order = list(conns)
if self.seed is not None:
random.Random(self.seed).shuffle(order)
if self.route_first:
# stable: the promoted keep their order among themselves, as do the rest
order.sort(key=lambda c: (c.net_no, *_ends_of(c)) not in self.route_first)
for _ in range(passes):
changed = False
for conn in conns: # deterministic: connection creation order
for conn in order:
if conn.routed:
continue
if self._try_route(conn, conns):
@@ -1050,6 +1155,9 @@ def route_board(
edge_clearance: int | None = None,
resolution: int = 1,
heuristic_weight: float = 1.0,
seed: int | None = None,
route_first: frozenset | None = None,
access_cost: float = 0.0,
) -> RouteResult:
"""Convenience wrapper: build a :class:`GridRouter` and route the board."""
router = GridRouter(
@@ -1065,5 +1173,8 @@ def route_board(
edge_clearance=edge_clearance,
resolution=resolution,
heuristic_weight=heuristic_weight,
seed=seed,
route_first=route_first,
access_cost=access_cost,
)
return router.route()
+98
View File
@@ -0,0 +1,98 @@
"""Route one board in several connection orders at once and keep the best.
Which connection routes first moves the result on a crowded board by several
connections either way, and no single order wins everywhere. Routing a few
orders in parallel processes and keeping the one that leaves the fewest
connections unrouted costs cores, not wall time.
Without a seed the candidates are the default order and seeds 0 to N-2, so the
result is never worse than a plain run. With seed S they are seeds S to S+N-1.
Ties go to the earlier candidate, so the choice is deterministic.
"""
from __future__ import annotations
from concurrent.futures import ProcessPoolExecutor
import os
from freeroute.dsn import parse_dsn
from freeroute.ses import RoutingResult
from .pipeline import build_exact_routing_result, build_routing_result
def candidate_seeds(orders: int, seed: int | None = None) -> list[int | None]:
orders = max(int(orders), 1)
if seed is None:
return [None, *range(orders - 1)]
return list(range(seed, seed + orders))
def score(result: RoutingResult) -> tuple[int, int]:
"""Connections left open, then those left to a zone fill: lower is better."""
missing = sum(1 for u in result.unrouted if not u[3])
return missing, len(result.unrouted) - missing
def route_once(dsn_text: str, engine: str, options: dict, seed: int | None) -> RoutingResult:
dsn = parse_dsn(dsn_text)
if engine == "exact":
return build_exact_routing_result(dsn, seed=seed, **options)
if engine == "grid":
return build_routing_result(dsn, seed=seed, **options)
raise ValueError(f"routing orders are not supported by the {engine} engine")
def best_of_orders(
dsn_text: str,
engine: str,
options: dict,
*,
orders: int,
seed: int | None = None,
jobs: int | None = None,
) -> tuple[RoutingResult, int | None, list[tuple[int | None, tuple[int, int]]]]:
"""Route ``orders`` orders and return the best result, the seed that made
it (None for the default order), and every candidate's seed and score."""
seeds = candidate_seeds(orders, seed)
if len(seeds) == 1:
result = route_once(dsn_text, engine, options, seeds[0])
return result, seeds[0], [(seeds[0], score(result))]
jobs = max(1, min(jobs or os.cpu_count() or 1, len(seeds)))
with ProcessPoolExecutor(jobs) as pool:
futures = [pool.submit(route_once, dsn_text, engine, options, s) for s in seeds]
results = [f.result() for f in futures]
scores = [score(r) for r in results]
best = min(range(len(seeds)), key=lambda i: (scores[i], i))
return results[best], seeds[best], list(zip(seeds, scores, strict=True))
def promote_failures(
dsn_text: str,
engine: str,
options: dict,
*,
rounds: int,
seed: int | None = None,
) -> tuple[RoutingResult, list[tuple[int, int]]]:
"""Route, then route again with every connection that has failed so far
moved to the front, keeping the rest in their order. Returns the best
round's result and every round's score.
The promoted set only grows, so a connection that failed once keeps its
early place and two groups cannot take turns failing. Stops early when a
round routes everything or fails nothing new."""
numbers = {n.name: i + 1 for i, n in enumerate(parse_dsn(dsn_text).nets)}
first: frozenset = frozenset()
best: RoutingResult | None = None
scores: list[tuple[int, int]] = []
for _ in range(max(int(rounds), 1)):
result = route_once(dsn_text, engine, {**options, "route_first": first}, seed)
scores.append(score(result))
if best is None or scores[-1] < score(best):
best = result
failed = {(numbers.get(name), a, b) for name, a, b, zone in result.unrouted if not zone}
if not failed or failed <= first:
break
first = first | failed
return best, scores
+24 -1
View File
@@ -80,6 +80,9 @@ def route_dsn_board(
layers: list[int] | None = None,
rip_up: bool = True,
max_passes: int = 10,
seed: int | None = None,
route_first: frozenset | None = None,
access_cost: float = 0.0,
) -> tuple[RouteResult, int, list[str]]:
"""Route a parsed :class:`DsnBoard`; return the board-unit result, the scale,
and the layer names. ``layers`` restricts routing to those signal-layer
@@ -100,6 +103,9 @@ def route_dsn_board(
layers=layers,
rip_up=rip_up,
max_passes=max_passes,
seed=seed,
route_first=route_first,
access_cost=access_cost,
)
return result, scale, [layer.name for layer in dsn.layers]
@@ -140,11 +146,16 @@ def build_routing_result(
rip_up: bool = True,
max_passes: int = 10,
edge_clearance_mm: float | None = None,
seed: int | None = None,
route_first: frozenset | None = None,
access_cost: float = 0.0,
) -> RoutingResult:
"""Route ``dsn`` (grid track) and convert to a DSN-unit RoutingResult."""
route, scale, layer_names = route_dsn_board(
dsn, layers=layers, rip_up=rip_up, max_passes=max_passes,
edge_clearance_mm=edge_clearance_mm, resolution=resolution,
edge_clearance_mm=edge_clearance_mm, resolution=resolution, seed=seed,
route_first=route_first,
access_cost=access_cost,
)
return _to_routing_result(route, dsn, scale, layer_names)
@@ -160,6 +171,9 @@ def route_dsn_board_exact(
rip_up: bool = True,
shove: bool = False,
diagonal: bool = False,
seed: int | None = None,
route_first: frozenset | None = None,
access_cost: float = 0.0,
) -> tuple[ExactRouteResult, int, list[str]]:
"""Route a parsed :class:`DsnBoard` with the exact-geometry (orthogonal,
DRC-verified) track; return the exact result, the scale, and layer names.
@@ -184,6 +198,9 @@ def route_dsn_board_exact(
rip_up=rip_up,
shove=shove,
diagonal=diagonal,
seed=seed,
route_first=route_first,
access_cost=access_cost,
)
return exact, scale, [layer.name for layer in dsn.layers]
@@ -199,6 +216,9 @@ def build_exact_routing_result(
rip_up: bool = True,
max_passes: int = 10,
edge_clearance_mm: float | None = None,
seed: int | None = None,
route_first: frozenset | None = None,
access_cost: float = 0.0,
) -> RoutingResult:
"""Route ``dsn`` (exact track) and convert to a DSN-unit RoutingResult."""
exact, scale, layer_names = route_dsn_board_exact(
@@ -211,6 +231,9 @@ def build_exact_routing_result(
edge_clearance_mm=edge_clearance_mm,
resolution=resolution,
heuristic_weight=heuristic_weight,
seed=seed,
route_first=route_first,
access_cost=access_cost,
)
return _to_routing_result(exact.result, dsn, scale, layer_names)
+78
View File
@@ -0,0 +1,78 @@
"""--orders: route several connection orders and keep the best."""
from __future__ import annotations
from pathlib import Path
from freeroute.cli import main
from freeroute.route.orders import (
best_of_orders,
candidate_seeds,
promote_failures,
route_once,
score,
)
FIXTURES = Path(__file__).resolve().parent.parent / "dsn" / "fixtures"
# Greedy (no rip-up), the default order strands a net here and seeds 1 to 3 do not.
RIPUP = (FIXTURES / "ripup_needed.dsn").read_text()
GREEDY = {"rip_up": False}
def test_the_default_order_is_always_a_candidate_without_a_seed():
assert candidate_seeds(1) == [None]
assert candidate_seeds(4) == [None, 0, 1, 2]
assert candidate_seeds(3, seed=7) == [7, 8, 9]
def test_keeps_the_order_that_leaves_the_fewest_unrouted():
assert score(route_once(RIPUP, "grid", GREEDY, None)) == (1, 0)
result, kept, tried = best_of_orders(RIPUP, "grid", GREEDY, orders=3, jobs=2)
assert kept == 1
assert score(result) == (0, 0)
assert tried == [(None, (1, 0)), (0, (1, 0)), (1, (0, 0))]
def test_ties_go_to_the_earlier_candidate():
# every order routes this board fully, so the default order is kept
_, kept, tried = best_of_orders(RIPUP, "grid", {}, orders=3, jobs=3)
assert kept is None
assert all(s == (0, 0) for _, s in tried)
def test_cli_orders_writes_the_best_session(tmp_path, capsys):
out = tmp_path / "best.ses"
rc = main([str(FIXTURES / "ripup_needed.dsn"), "--no-rip-up", "--orders", "3", "-do", str(out)])
assert rc == 0
err = capsys.readouterr().err
assert "tried 3 orders, kept --seed 1" in err
assert "not routed" not in err
assert out.read_text().startswith("(session")
def test_promoting_the_failed_connection_routes_it():
best, rounds = promote_failures(RIPUP, "grid", GREEDY, rounds=3)
assert rounds == [(1, 0), (0, 0)] # stops once nothing fails
assert score(best) == (0, 0)
def test_nothing_promoted_is_the_plain_order():
plain = route_once(RIPUP, "grid", {}, None)
empty = route_once(RIPUP, "grid", {"route_first": frozenset()}, None)
assert plain == empty
def test_cli_promote_failures(tmp_path, capsys):
out = tmp_path / "p.ses"
rc = main(
[
str(FIXTURES / "ripup_needed.dsn"),
"--no-rip-up",
"--promote-failures",
"3",
"-do",
str(out),
]
)
assert rc == 0
assert "2 rounds, unrouted 1, 0" in capsys.readouterr().err
+13
View File
@@ -264,6 +264,19 @@ def test_larger_board_still_connects_its_multipin_nets():
assert len(result.routed_net_numbers) == multi_pin == 4
def test_a_seed_is_repeatable_and_does_change_the_order():
# The benchmark samples seeds to measure how much a result owes to the
# routing order. A seed that changed nothing would make every sample the
# same run and report no spread at all.
text = (FIXTURES / "kicad_routable.dsn").read_text()
default = _wire_key(route_dsn_board(parse_dsn(text))[0])
runs = {s: route_dsn_board(parse_dsn(text), seed=s)[0] for s in range(4)}
again = route_dsn_board(parse_dsn(text), seed=1)[0]
assert _wire_key(again) == _wire_key(runs[1])
assert any(_wire_key(r) != default for r in runs.values())
assert all(r.unrouted == 0 for r in runs.values())
# --- oracle parity ----------------------------------------------------------
+4
View File
@@ -154,6 +154,10 @@ def test_cli_supported_combinations_are_accepted(tmp_path):
"diagonal": ["--diagonal"],
"rip_up": ["--no-rip-up"],
"max_passes": ["-mp", "4"],
"seed": ["--seed", "3"],
"orders": ["--orders", "2"],
"promote": ["--promote-failures", "2"],
"access": ["--access-cost", "5"],
}
for engine, options in ENGINE_OPTIONS.items():
for option in options:
Generated
+1 -1
View File
@@ -128,7 +128,7 @@ wheels = [
[[package]]
name = "freeroute"
version = "0.1.0"
version = "2026.7.14"
source = { editable = "." }
[package.dev-dependencies]