#!/usr/bin/python3 """Bounded, board-specific two-layer draft router. Requires pcbnew and numpy. Dry-run by default. --route --overwrite explicitly permits saving routing; existing copper additionally requires --replace-routes. Never edits the project, schematic, footprint libraries or generator. Review actual KiCad DRC afterwards. This is not a production autorouter or a fabrication qualification. """ import argparse from collections import defaultdict import hashlib import heapq import json import math from pathlib import Path import time import numpy as np import pcbnew as p ROOT = Path(__file__).resolve().parent PCB = ROOT / 'serial-carrier.kicad_pcb' STEP = 0.05 NX, NY = int(82 / STEP) + 1, int(80 / STEP) + 1 N = NX * NY CLEARANCE = 0.25 MARGIN = 0.02 LAYERS = (p.F_Cu, p.B_Cu) X = 100 + np.arange(NX)[None, :] * STEP Y = 100 + np.arange(NY)[:, None] * STEP def mm(v): return p.FromMM(v) def point(xy): return p.VECTOR2I(mm(xy[0]), mm(xy[1])) def pos(item): v = item.GetPosition() return (p.ToMM(v.x), p.ToMM(v.y)) def identity(board): """Immutable geometry/linkage snapshot, independent of serialization order.""" footprints = [] for f in board.GetFootprints(): pads = sorted((a.m_Uuid.AsString(), a.GetNumber(), a.GetNetname(), pos(a), (a.GetSize().x, a.GetSize().y), (a.GetDrillSize().x, a.GetDrillSize().y), a.GetShape(), a.GetAttribute(), a.GetLayerSet().FmtHex(), a.GetOrientationDegrees(), a.GetPinFunction(), a.GetPinType()) for a in f.Pads()) footprints.append((f.m_Uuid.AsString(), f.GetReference(), f.GetValue(), f.GetFPIDAsString(), f.GetPath().AsString(), pos(f), f.GetOrientationDegrees(), f.GetLayer(), pads)) edges = sorted((a.m_Uuid.AsString(), a.GetShape(), a.GetStart().x, a.GetStart().y, a.GetEnd().x, a.GetEnd().y, a.GetWidth()) for a in board.GetDrawings() if a.GetLayer() == p.Edge_Cuts) zones = sorted((z.m_Uuid.AsString(), z.GetLayerSet().FmtHex(), z.GetIsRuleArea(), z.GetDoNotAllowTracks(), z.GetDoNotAllowVias(), z.GetDoNotAllowZoneFills(), tuple((z.Outline().COutline(0).CPoint(i).x, z.Outline().COutline(0).CPoint(i).y) for i in range(z.Outline().COutline(0).PointCount()))) for z in board.Zones()) nets = sorted((n.GetNetCode(), n.GetNetname()) for n in board.GetNetsByNetcode().values()) return (board.GetCopperLayerCount(), sorted(footprints), edges, zones, nets) def window(x0, y0, x1, y1): ix0 = max(0, int(math.floor((x0 - 100) / STEP))) iy0 = max(0, int(math.floor((y0 - 100) / STEP))) ix1 = min(NX, int(math.ceil((x1 - 100) / STEP)) + 1) iy1 = min(NY, int(math.ceil((y1 - 100) / STEP)) + 1) return slice(iy0, iy1), slice(ix0, ix1) def capsule(mask, a, b, radius): sy, sx = window(min(a[0], b[0]) - radius, min(a[1], b[1]) - radius, max(a[0], b[0]) + radius, max(a[1], b[1]) + radius) xx, yy = X[:, sx], Y[sy, :] dx, dy = b[0] - a[0], b[1] - a[1] length2 = dx * dx + dy * dy t = np.clip(((xx - a[0]) * dx + (yy - a[1]) * dy) / length2, 0, 1) if length2 else 0 mask[sy, sx] |= (xx - a[0] - t * dx)**2 + (yy - a[1] - t * dy)**2 <= radius**2 def pad_obstacle(mask, pad, extra, hole_only=False): xy = pos(pad) size = pad.GetDrillSize() if hole_only else pad.GetSize() rx, ry = p.ToMM(size.x) / 2, p.ToMM(size.y) / 2 if not rx or not ry: return if abs(rx - ry) < 1e-6 and (hole_only or pad.GetShape() == p.PAD_SHAPE_CIRCLE): capsule(mask, xy, xy, rx + extra) else: # Bounding rectangle deliberately overestimates non-circular pad shapes. box = pad.GetBoundingBox() x0, y0 = p.ToMM(box.GetX()) - extra, p.ToMM(box.GetY()) - extra x1, y1 = p.ToMM(box.GetRight()) + extra, p.ToMM(box.GetBottom()) + extra sy, sx = window(x0, y0, x1, y1) mask[sy, sx] |= (X[:, sx] >= x0) & (X[:, sx] <= x1) & (Y[sy, :] >= y0) & (Y[sy, :] <= y1) def boundary(radius): edge = 0.5 + radius + MARGIN mask = np.broadcast_to((X < 100 + edge) | (X > 182 - edge) | (Y < 100 + edge) | (Y > 180 - edge), (NY, NX)).copy() mask |= (X < 136 + edge) & (Y > 161 - edge) # Expand the RF reservation by copper radius plus numerical safety margin. r = radius + MARGIN mask |= (X >= 110 - r) & (X <= 131 + r) & (Y <= 113 + r) return mask def masks(board, pads, net, width): blocked = np.stack([boundary(width / 2)] * 2) via = boundary(0.35) for pad in pads: if pad.GetNetCode() != net or not pad.GetNumber(): for z, layer in enumerate(LAYERS): if pad.IsOnLayer(layer): pad_obstacle(blocked[z], pad, CLEARANCE + width / 2 + MARGIN) pad_obstacle(via, pad, CLEARANCE + 0.35 + MARGIN) # Even same-net holes must not receive a drilled via. pad_obstacle(via, pad, CLEARANCE + 0.35 + MARGIN, hole_only=True) for track in board.GetTracks(): if track.GetNetCode() == net: if isinstance(track, p.PCB_VIA): capsule(via, pos(track), pos(track), 0.7 + CLEARANCE + MARGIN) continue a = (p.ToMM(track.GetStart().x), p.ToMM(track.GetStart().y)) b = (p.ToMM(track.GetEnd().x), p.ToMM(track.GetEnd().y)) radius = p.ToMM(track.GetWidth(p.F_Cu) if isinstance(track, p.PCB_VIA) else track.GetWidth()) / 2 for z, layer in enumerate(LAYERS): if isinstance(track, p.PCB_VIA) or track.GetLayer() == layer: capsule(blocked[z], a, b, radius + CLEARANCE + width / 2 + MARGIN) capsule(via, a, b, radius + CLEARANCE + 0.35 + MARGIN) return blocked.reshape(-1), via.reshape(-1) def node(xy, layer=0): x, y = (int(round((v - 100) / STEP)) for v in xy) return layer * N + y * NX + x def decode(i): z, q = divmod(i, N) y, x = divmod(q, NX) return (100 + x * STEP, 100 + y * STEP), z def astar(blocked, via, start, target, deadline, max_expansions): tx = int(round((target[0] - 100) / STEP)) ty = int(round((target[1] - 100) / STEP)) targetq = ty * NX + tx def h(q): y, x = divmod(q, NX) dx, dy = abs(x - tx), abs(y - ty) return 10 * max(dx, dy) + 4 * min(dx, dy) dist = np.full(2 * N, 2147483647, dtype=np.int32) parent = np.full(2 * N, -1, dtype=np.int32) heap = [] for z in (0, 1): i = node(start, z) if not blocked[i]: dist[i] = 0 heapq.heappush(heap, (h(i % N), 0, i)) expanded = 0 moves = ((1, 0, 10), (-1, 0, 10), (0, 1, 10), (0, -1, 10), (1, 1, 14), (1, -1, 14), (-1, 1, 14), (-1, -1, 14)) while heap: _, cost, i = heapq.heappop(heap) if cost != dist[i]: continue z, q = divmod(i, N) if q == targetq: path = [i] while parent[path[-1]] >= 0: path.append(int(parent[path[-1]])) return path[::-1], expanded expanded += 1 if expanded >= max_expansions or (expanded % 1024 == 0 and time.monotonic() >= deadline): return None, expanded y, x = divmod(q, NX) for dx, dy, stepcost in moves: if not (0 <= x + dx < NX and 0 <= y + dy < NY): continue j = i + dy * NX + dx if blocked[j] or (dx and dy and (blocked[i + dx] or blocked[i + dy * NX])): continue # Mild layer preference prevents needless coincident layer choices. penalty = int((z == 0 and dx == 0) or (z == 1 and dy == 0)) nc = cost + stepcost + penalty if nc < dist[j]: dist[j], parent[j] = nc, i heapq.heappush(heap, (nc + h(j % N), nc, j)) j = (1 - z) * N + q if not via[q] and not blocked[j]: nc = cost + int(18 / STEP) # 1.8 mm equivalent cost for a layer change. if nc < dist[j]: dist[j], parent[j] = nc, i heapq.heappush(heap, (nc + h(q), nc, j)) return None, expanded def add_path(board, path, start, target, net, width): points = [decode(i) for i in path] points = [(start, points[0][1])] + points + [(target, points[-1][1])] # Only merge exactly collinear grid runs; never shortcut obstacle checks. simple = [] for item in points: if simple and item == simple[-1]: continue while len(simple) >= 2 and item[1] == simple[-1][1] == simple[-2][1]: a, b, c = simple[-2][0], simple[-1][0], item[0] cross = (b[0]-a[0])*(c[1]-b[1]) - (b[1]-a[1])*(c[0]-b[0]) dot = (b[0]-a[0])*(c[0]-b[0]) + (b[1]-a[1])*(c[1]-b[1]) if abs(cross) > 1e-8 or dot < 0: break simple.pop() simple.append(item) for (a, z), (b, zz) in zip(simple, simple[1:]): if z != zz: assert a == b item = p.PCB_VIA(board) item.SetPosition(point(a)) item.SetWidth(mm(0.7)) item.SetDrill(mm(0.3)) item.SetViaType(p.VIATYPE_THROUGH) item.SetLayerPair(p.F_Cu, p.B_Cu) else: if a == b: continue item = p.PCB_TRACK(board) item.SetStart(point(a)) item.SetEnd(point(b)) item.SetWidth(mm(width)) item.SetLayer(LAYERS[z]) item.SetNetCode(net) board.Add(item) def pairs_for_net(pads): # Minimum spanning tree includes duplicate switch contacts explicitly. connected = {0} todo = set(range(1, len(pads))) pairs = [] while todo: length, a, b = min((math.dist(pos(pads[a]), pos(pads[b])), a, b) for a in connected for b in todo) pairs.append((length, pads[a], pads[b])) connected.add(b) todo.remove(b) return pairs def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--route', action='store_true') parser.add_argument('--overwrite', action='store_true') parser.add_argument('--replace-routes', action='store_true') parser.add_argument('--seconds', type=float, default=240) parser.add_argument('--max-expansions', type=int, default=1200000) args = parser.parse_args() if args.route and not args.overwrite: parser.error('--route requires --overwrite; back up manual routing first') if not 0 < args.seconds <= 900 or not 0 < args.max_expansions <= 1500000: parser.error('bounds: seconds (0,900], max-expansions (0,1500000]') if ROOT.joinpath('~serial-carrier.kicad_pcb.lck').exists(): parser.error('PCB editor lock exists; close the PCB before routing') original_bytes = PCB.read_bytes() board = p.LoadBoard(str(PCB)) before = identity(board) if board.GetCopperLayerCount() != 2 or len(board.GetFootprints()) != 13: parser.error('unexpected board structure; review router assumptions') if len(board.Zones()) != 1 or not board.Zones()[0].GetIsRuleArea(): parser.error('unexpected zones; router supports only the original RF rule area') outline = [(100, 100), (182, 100), (182, 180), (136, 180), (136, 161), (100, 161)] expected_edges = {frozenset((a, b)) for a, b in zip(outline, outline[1:] + outline[:1])} actual_edges = {frozenset(((p.ToMM(e.GetStart().x), p.ToMM(e.GetStart().y)), (p.ToMM(e.GetEnd().x), p.ToMM(e.GetEnd().y)))) for e in board.GetDrawings() if e.GetLayer() == p.Edge_Cuts} if actual_edges != expected_edges: parser.error('outline differs from the board-specific routing envelope') reserve = board.Zones()[0] contour = reserve.Outline().COutline(0) rf_points = {(p.ToMM(contour.CPoint(i).x), p.ToMM(contour.CPoint(i).y)) for i in range(contour.PointCount())} if (rf_points != {(110, 100), (131, 100), (131, 113), (110, 113)} or not all(reserve.IsOnLayer(layer) for layer in LAYERS) or not all((reserve.GetDoNotAllowTracks(), reserve.GetDoNotAllowVias(), reserve.GetDoNotAllowZoneFills()))): parser.error('RF rule area differs from the board-specific reservation') if board.GetTracks() and not args.replace_routes: parser.error('existing routing protected; --replace-routes required even for preview') for track in list(board.GetTracks()): board.Delete(track) pads = [a for f in board.GetFootprints() for a in f.Pads()] if any(a.GetOrientationDegrees() % 90 or not all(a.IsOnLayer(l) for l in LAYERS) for a in pads): parser.error('router requires axis-aligned through-hole pads') nets = defaultdict(list) for pad in pads: if pad.GetNumber() and pad.GetNetCode(): nets[pad.GetNetCode()].append(pad) work = [] for net, netpads in nets.items(): if len(netpads) > 1: name = netpads[0].GetNetname() for length, a, b in pairs_for_net(netpads): # Short local links first, then the long signal fanout; ground last. priority = (name == '/GND', length) work.append((priority, net, name, a, b)) work.sort(key=lambda w: (w[0], w[2])) start_time = time.monotonic() deadline = start_time + args.seconds results = [] for _, net, name, a, b in work: width = 0.5 if name in ('/+3V3', '/GND') else 0.25 path, expanded = None, 0 if time.monotonic() < deadline: blocked, via = masks(board, pads, net, width) path, expanded = astar(blocked, via, pos(a), pos(b), deadline, args.max_expansions) fallback = False if path is None and width == 0.5 and time.monotonic() < deadline: width, fallback = 0.25, True blocked, via = masks(board, pads, net, width) path, more = astar(blocked, via, pos(a), pos(b), deadline, args.max_expansions) expanded += more label = lambda pad: pad.GetParentFootprint().GetReference() + '.' + pad.GetNumber() result = dict(net=name, start=label(a), end=label(b), start_mm=pos(a), end_mm=pos(b), routed=path is not None, width_mm=width, power_fallback=fallback, expansions=expanded) results.append(result) if path is not None: add_path(board, path, pos(a), pos(b), net, width) print(json.dumps(result), flush=True) assert identity(board) == before, 'immutable board data changed' report = dict(draft_only=True, saved=args.route, grid_mm=STEP, clearance_mm=CLEARANCE, safety_margin_mm=MARGIN, via_diameter_mm=0.7, via_drill_mm=0.3, elapsed_seconds=time.monotonic()-start_time, connections=results, routed_tree_edges=sum(r['routed'] for r in results), total_tree_edges=len(results), tracks=sum(not isinstance(t, p.PCB_VIA) for t in board.GetTracks()), vias=sum(isinstance(t, p.PCB_VIA) for t in board.GetTracks()), immutable_identity_sha256=hashlib.sha256(repr(before).encode()).hexdigest(), input_sha256=hashlib.sha256(original_bytes).hexdigest(), validation='KiCad DRC must be run separately; tree-edge counts are not DRC connectivity') if args.route: if PCB.read_bytes() != original_bytes or ROOT.joinpath('~serial-carrier.kicad_pcb.lck').exists(): raise RuntimeError('PCB changed or editor opened during routing; refusing overwrite') # Check a serialized candidate before replacing the user's board. candidate = ROOT / 'validation/pcb-routing-candidate.kicad_pcb' p.SaveBoard(str(candidate), board) assert identity(p.LoadBoard(str(candidate))) == before, 'saved identity mismatch' candidate.replace(PCB) ROOT.joinpath('validation/pcb-routing-run.json').write_text(json.dumps(report, indent=2)+'\n') print(json.dumps({k: v for k, v in report.items() if k != 'connections'}, indent=2)) if __name__ == '__main__': main()