KiCad - the KiCad Bridge

#!/usr/bin/env python3
"""Apply a routing plan OFFLINE into a copy of the board and gate it with the shared KiCad DRC.

    python3 demo/routing/esc/write_copper.py BOARD-planes.kicad_pcb PLAN.json [--out FILE]
                                             [--no-fill-planes] [--drc] [--baseline BOARD.kicad_pcb]
                                             [--report drc-report.json]

The plan is the list of `kicad_route_net` calls the router wrote (see ai_router.py): each entry
is {"net", "width", "points": [...]} or {"net", "width", "paths": [[...], ...]} plus optional
"viaSize" / "viaDrill". Waypoints are "REF.PAD" (the pad centre), [x, y], or {"x", "y",
"layer": "B.Cu"} / {"pad": "REF.PAD", "layer": ...}; a layer marker means a through via AT that
point and every following segment on the new layer, exactly as rust/crates/kicad-core/src/pcb.rs
`plan` reads it. Segments and vias are written as KiCad 10 s-expressions with the board's net
reference, a fresh uuid each, and spliced in before the closing paren.

Plane zones (GND on In1.Cu, +3V3 on In2.Cu from add_planes.py) have no fill in the file and the
shared `service-kicad pcb drc` does not refill zones, so by default this script also writes a
raster fill for every zone: overlapping 0.1 mm strips covering the outline (0.5 mm edge clearance)
minus every other-net through pad, via and hole expanded by the 0.2 mm clearance and the grid
slop. KiCad's own fill on the live board replaces it (thermal reliefs and all); here it lets the
DRC count each plane-net via as connected. --no-fill-planes leaves the zones unfilled.

--drc runs `service-kicad pcb drc` on the written file and prints errors, warnings, unconnected
items and the violation types. With --baseline, errors already present on the unrouted board
(same type, description and item positions) are listed separately as inherited, the way the
bridge's route_net compares before/after; the gate is zero NEW errors and zero unconnected.
"""
import argparse
import json
import math
import os
import subprocess
import sys
import tempfile
import uuid
from collections import Counter
from pathlib import Path

import numpy as np

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from escboard import Board, Grid, fmt  # noqa: E402

CLEARANCE = 0.2
EDGE_CLEARANCE = 0.5
SLOP = 0.1


# ----------------------------------------------------------------------------- plan -> copper

def _point(spec, board):
    """(x, y, pad, explicit_layer) for one waypoint, matching pcb.rs resolve_point."""
    layer = None
    pad = None
    if isinstance(spec, str):
        pad = board.pad_by_key(spec)
        if pad is None:
            raise ValueError("unknown pad %r" % spec)
        return pad.x, pad.y, pad, None
    if isinstance(spec, dict):
        layer = spec.get("layer")
        if spec.get("pad"):
            pad = board.pad_by_key(str(spec["pad"]))
            if pad is None:
                raise ValueError("unknown pad %r" % spec["pad"])
            return pad.x, pad.y, pad, layer
        return float(spec["x"]), float(spec["y"]), None, layer
    if isinstance(spec, (list, tuple)) and len(spec) == 2:
        return float(spec[0]), float(spec[1]), None, None
    raise ValueError("bad waypoint %r" % (spec,))


def plan_to_copper(board, entry):
    """One plan entry -> (segments, vias): segments (x0,y0,x1,y1,layer,width), vias (x,y,size,drill)."""
    net = entry["net"]
    if net not in board.nets:
        raise ValueError("unknown net %r" % net)
    width = float(entry.get("width", 0.25))
    size = float(entry.get("viaSize", 0.6))
    drill = float(entry.get("viaDrill", 0.3))
    paths = entry.get("paths") or [entry["points"]]
    segments, vias = [], []
    for path in paths:
        if not 2 <= len(path) <= 512:
            raise ValueError("path needs 2..512 waypoints (%s)" % net)
        resolved = [_point(p, board) for p in path]
        for x, y, pad, _ in resolved:
            if pad is not None and pad.net != net:
                raise ValueError("pad %s is on %s, not %s" % (pad.key, pad.net, net))
        layer = entry.get("layer") or resolved[0][3] or "F.Cu"
        for i, (x, y, pad, explicit) in enumerate(resolved):
            nxt = explicit or layer
            if i > 0:
                px, py = resolved[i - 1][0], resolved[i - 1][1]
                if (px, py) != (x, y):
                    segments.append((px, py, x, y, layer, width))
            if nxt != layer:
                vias.append((x, y, size, drill))
                layer = nxt
            if pad is not None and layer not in pad.layers:
                raise ValueError("pad %s has no copper on %s" % (pad.key, layer))
    return segments, vias


def segment_sexpr(board, s, net):
    x0, y0, x1, y1, layer, w = s
    return ("\t(segment\n\t\t(start %s %s)\n\t\t(end %s %s)\n\t\t(width %s)\n\t\t(layer \"%s\")\n%s\t\t(uuid \"%s\")\n\t)"
            % (fmt(x0), fmt(y0), fmt(x1), fmt(y1), fmt(w), layer, board.net_ref_sexpr(net), uuid.uuid4()))


def via_sexpr(board, v, net):
    x, y, size, drill = v
    return ("\t(via\n\t\t(at %s %s)\n\t\t(size %s)\n\t\t(drill %s)\n\t\t(layers \"F.Cu\" \"B.Cu\")\n%s\t\t(uuid \"%s\")\n\t)"
            % (fmt(x), fmt(y), fmt(size), fmt(drill), board.net_ref_sexpr(net), uuid.uuid4()))


def splice(text, blocks):
    trimmed = text.rstrip()
    idx = trimmed.rfind(")")
    body = text[:idx].rstrip()
    nl = "\r\n" if "\r\n" in text[:2000] else "\n"
    return body + nl + nl.join(b.replace("\n", nl) for b in blocks) + nl + ")" + nl


# ----------------------------------------------------------------------------- plane raster fill

def zone_fill_polygons(board, zone_net, zone_layer, all_vias_by_net, grid=None):
    """ONE fractured polygon per connected region of the plane (KiCad's own fill format: the outer
    contour with every hole spliced in through a zero-width slit). KiCad does not connect
    overlapping sub-polygons of one zone to each other, so a fill made of strips would be a set
    of islands; a single polygon is one island the way a real fill is.

    all_vias_by_net: {net: [(x, y, size, drill), ...]} for every via the plan adds."""
    grid = grid or Grid(board.bbox)
    free = grid.outline_inside(board.outline, EDGE_CLEARANCE + SLOP)
    block = np.zeros((grid.H, grid.W), dtype=bool)
    by_clearance = {}   # pads with a local clearance override are dilated by their own value
    for p in board.pads:
        if not p.thru or p.net == zone_net:
            continue
        target = block if p.clearance <= CLEARANCE else by_clearance.setdefault(p.clearance, np.zeros((grid.H, grid.W), dtype=bool))
        for poly in p.polys:
            grid.poly_mask(poly, target)
        if p.drill:
            jj, ii = grid.disk_cells((p.x, p.y), p.drill / 2.0)
            target[jj, ii] = True
    for net, vias in all_vias_by_net.items():
        if net == zone_net:
            continue
        for x, y, size, drill in vias:
            jj, ii = grid.disk_cells((x, y), size / 2.0)
            block[jj, ii] = True
    block = grid.dilate(block, (CLEARANCE + SLOP) / grid.res)
    for clr, mask in by_clearance.items():
        block |= grid.dilate(mask, (clr + SLOP) / grid.res)
    free &= ~block
    loops = _trace_boundaries(free)
    outers = [lp for lp in loops if _signed_area(lp) > 0]
    holes = [lp for lp in loops if _signed_area(lp) < 0]
    polys = []
    for outer in outers:
        ob = _bbox(outer)
        mine = []
        for h in holes:
            hb = _bbox(h)
            if hb[0] < ob[0] or hb[1] < ob[1] or hb[2] > ob[2] or hb[3] > ob[3]:
                continue
            if _point_in_poly(h[0][0] + 1e-6, h[0][1] + 1e-6, outer):
                mine.append(h)
        poly = _fracture(outer, mine)
        polys.append([(grid.x0 + x * grid.res, grid.y0 + y * grid.res) for x, y in poly])
    return polys


def _trace_boundaries(free):
    """Boundary loops of a binary raster as lattice polygons (cell-corner coordinates, in cells).
    Region on the right of the direction of travel; outer loops and holes get opposite signed areas."""
    H, W = free.shape
    padded = np.zeros((H + 2, W + 2), dtype=bool)
    padded[1:-1, 1:-1] = free
    f = padded
    # directed edges around free cells whose neighbour across that side is not free
    edges = []
    jj, ii = np.nonzero(f[1:-1, 1:-1] & ~f[0:-2, 1:-1])   # top side: TL -> TR
    edges += [((i - 0.5, j - 0.5), (i + 0.5, j - 0.5)) for j, i in zip(jj.tolist(), ii.tolist())]
    jj, ii = np.nonzero(f[1:-1, 1:-1] & ~f[1:-1, 2:])     # right side: TR -> BR
    edges += [((i + 0.5, j - 0.5), (i + 0.5, j + 0.5)) for j, i in zip(jj.tolist(), ii.tolist())]
    jj, ii = np.nonzero(f[1:-1, 1:-1] & ~f[2:, 1:-1])     # bottom side: BR -> BL
    edges += [((i + 0.5, j + 0.5), (i - 0.5, j + 0.5)) for j, i in zip(jj.tolist(), ii.tolist())]
    jj, ii = np.nonzero(f[1:-1, 1:-1] & ~f[1:-1, 0:-2])   # left side: BL -> TL
    edges += [((i - 0.5, j + 0.5), (i - 0.5, j - 0.5)) for j, i in zip(jj.tolist(), ii.tolist())]
    out = {}
    for a, b in edges:
        out.setdefault(a, []).append(b)
    loops = []
    while out:
        a = next(iter(out))
        b = out[a].pop()
        if not out[a]:
            del out[a]
        loop = [a, b]
        prev = (b[0] - a[0], b[1] - a[1])
        while True:
            cur = loop[-1]
            cands = out.get(cur)
            if not cands:
                break
            if len(cands) == 1:
                nxt = cands.pop()
            else:
                # a pinch point: prefer the right turn (toward the region), then straight, then left
                right = (-prev[1], prev[0])
                pick = None
                for want in (right, prev, (prev[1], -prev[0])):
                    for c in cands:
                        if (c[0] - cur[0], c[1] - cur[1]) == want:
                            pick = c
                            break
                    if pick is not None:
                        break
                nxt = pick if pick is not None else cands[-1]
                cands.remove(nxt)
            if not cands:
                del out[cur]
            prev = (nxt[0] - cur[0], nxt[1] - cur[1])
            if nxt == loop[0]:
                break
            loop.append(nxt)
        # drop collinear vertices
        simp = []
        n = len(loop)
        for k in range(n):
            p0, p1, p2 = loop[k - 1], loop[k], loop[(k + 1) % n]
            if (p1[0] - p0[0]) * (p2[1] - p1[1]) - (p1[1] - p0[1]) * (p2[0] - p1[0]) != 0:
                simp.append(p1)
        if len(simp) >= 4:
            loops.append(simp)
    return loops


def _signed_area(poly):
    a = 0.0
    for k in range(len(poly)):
        x0, y0 = poly[k]
        x1, y1 = poly[(k + 1) % len(poly)]
        a += x0 * y1 - x1 * y0
    return a / 2.0


def _bbox(poly):
    xs = [p[0] for p in poly]
    ys = [p[1] for p in poly]
    return min(xs), min(ys), max(xs), max(ys)


def _point_in_poly(x, y, poly):
    inside = False
    n = len(poly)
    for k in range(n):
        ax, ay = poly[k]
        bx, by = poly[(k + 1) % n]
        if (ay > y) != (by > y):
            if x < ax + (y - ay) * (bx - ax) / (by - ay):
                inside = not inside
    return inside


def _fracture(outer, holes):
    """Splice every hole into the outer contour through a horizontal zero-width slit to the left,
    holes in order of their left edge so a slit only ever meets contour that is already part of
    the polygon. The slit leaves the hole from the middle of its leftmost vertical edge, so it never
    hits a lattice vertex."""
    poly = list(outer)

    def leftmost_edge(h):
        best = None
        n = len(h)
        for k in range(n):
            a, b = h[k], h[(k + 1) % n]
            if a[0] == b[0] and (best is None or a[0] < best[0]):
                best = (a[0], k)
        return best

    order = sorted(range(len(holes)), key=lambda k: leftmost_edge(holes[k])[0])
    for hk in order:
        h = holes[hk]
        x_h, k = leftmost_edge(h)
        a, b = h[k], h[(k + 1) % len(h)]
        ym = (a[1] + b[1]) / 2.0
        if abs(b[1] - a[1]) >= 2:
            ym = a[1] + (0.5 if b[1] > a[1] else -0.5)  # a whole cell from the vertex: never a lattice vertex
        start = (x_h, ym)
        # hole contour starting at the slit point
        hole = [start] + h[k + 1:] + h[:k + 1] + [start]
        # nearest vertical edge of the polygon to the left of the slit point at height ym
        hit = None
        for e in range(len(poly)):
            p0, p1 = poly[e], poly[(e + 1) % len(poly)]
            if p0[0] == p1[0] and p0[0] < x_h and min(p0[1], p1[1]) < ym < max(p0[1], p1[1]):
                if hit is None or p0[0] > hit[0]:
                    hit = (p0[0], e)
        if hit is None:
            continue  # nothing to the left inside this outer: leave the hole out (cannot happen for a hole)
        xe, e = hit
        joint = (xe, ym)
        poly = poly[:e + 1] + [joint] + hole + [joint] + poly[e + 1:]
    return poly


def zones_of(board):
    """[(node, net, layer)] for every top-level zone."""
    out = []
    for z in board.root.find_all("zone"):
        n = z.find("net")
        net = n.value() if n is not None else ""
        if board.net_format == "number" and n is not None:
            try:
                net = board.net_table.get(int(n.atom(1)), net)
            except (TypeError, ValueError):
                pass
        nn = z.find("net_name")
        if nn is not None:
            net = nn.value()
        layer = z.find("layer").value() if z.find("layer") is not None else None
        out.append((z, net, layer))
    return out


def fill_zones(board, text, vias_by_net):
    """Return the board text with a filled_polygon list appended inside every zone."""
    grid = Grid(board.bbox)
    edits = []
    stats = {}
    for node, net, layer in zones_of(board):
        if not net or not layer:
            continue
        polys = zone_fill_polygons(board, net, layer, vias_by_net, grid)
        blocks = []
        for poly in polys:
            pts = " ".join("(xy %s %s)" % (fmt(x), fmt(y)) for x, y in poly)
            blocks.append("\t\t(filled_polygon\n\t\t\t(layer \"%s\")\n\t\t\t(pts\n\t\t\t\t%s\n\t\t\t)\n\t\t)\n" % (layer, pts))
        # insert before the zone's closing paren
        edits.append((node.end - 1, node.end - 1, "".join(blocks) + "\t"))
        # a raster fill has no thermal spokes: connect pads solidly in this gate copy so the DRC
        # does not report starved thermals on the through pads inside the plane
        cp = node.find("connect_pads")
        if cp is not None and cp.atom(1) not in ("yes", "no", "thru_hole_only"):
            edits.append((cp.start, cp.start + len("(connect_pads"), "(connect_pads yes"))
        stats[net] = len(polys)
    edits.sort()
    out = []
    pos = 0
    for at, end, ins in edits:
        out.append(text[pos:at])
        out.append(ins)
        pos = end
    out.append(text[pos:])
    return "".join(out), stats


# ----------------------------------------------------------------------------- DRC

def run_drc(path):
    with tempfile.TemporaryDirectory() as tmp:
        report = Path(tmp) / "drc.json"
        proc = subprocess.run(["service-kicad", "pcb", "drc", "--format", "json", "--out", str(report), str(path)],
                              capture_output=True, text=True, timeout=900)
        if proc.returncode or not report.is_file():
            raise RuntimeError("service-kicad drc failed: %s %s" % (proc.stdout[-600:], proc.stderr[-600:]))
        return json.loads(report.read_text(encoding="utf-8"))


def fingerprint(v):
    return json.dumps([v.get("type"), v.get("severity"), v.get("description"),
                       sorted(json.dumps(i.get("pos"), sort_keys=True) for i in v.get("items", []))], sort_keys=True)


def summarize(data, baseline=None):
    viol = data.get("violations", [])
    errors = [v for v in viol if v.get("severity") == "error"]
    warnings = [v for v in viol if v.get("severity") == "warning"]
    inherited = []
    new = errors
    if baseline is not None:
        existing = Counter(fingerprint(v) for v in baseline.get("violations", []) if v.get("severity") == "error")
        new, inherited = [], []
        for v in errors:
            k = fingerprint(v)
            if existing.get(k, 0) > 0:
                existing[k] -= 1
                inherited.append(v)
            else:
                new.append(v)
    unconnected = data.get("unconnected_items", [])
    # isolated_copper warnings on a raster-filled plane are the strips themselves (each filled
    # polygon that touches no pad is an "island" to KiCad); the live fill has none of them
    fill_artifacts = [v for v in warnings if v.get("type") == "isolated_copper"]
    warnings = [v for v in warnings if v.get("type") != "isolated_copper"]
    return {
        "errors": len(errors), "newErrors": len(new), "inheritedErrors": len(inherited),
        "warnings": len(warnings), "unconnected": len(unconnected),
        "rasterFillIslandWarnings": len(fill_artifacts),
        "errorTypes": dict(Counter(v.get("type") for v in errors)),
        "newErrorTypes": dict(Counter(v.get("type") for v in new)),
        "warningTypes": dict(Counter(v.get("type") for v in warnings)),
        "newErrorList": [{"type": v.get("type"), "description": v.get("description"),
                          "items": [i.get("description") for i in v.get("items", [])],
                          "pos": [i.get("pos") for i in v.get("items", [])]} for v in new],
        "unconnectedList": [[i.get("description") for i in u.get("items", [])] for u in unconnected],
        "kicadVersion": data.get("kicad_version"),
    }


# ----------------------------------------------------------------------------- main

def apply(board, plan, fill_planes=True):
    """Return (text, stats) for the board text with the plan's copper (and optional fills)."""
    entries = plan["nets"] if isinstance(plan, dict) else plan
    blocks = []
    vias_by_net = {}
    n_seg = n_via = 0
    length = 0.0
    widths = Counter()
    for entry in entries:
        segs, vias = plan_to_copper(board, entry)
        for s in segs:
            blocks.append(segment_sexpr(board, s, entry["net"]))
            length += math.hypot(s[2] - s[0], s[3] - s[1])
            widths[s[5]] += 1
        for v in vias:
            blocks.append(via_sexpr(board, v, entry["net"]))
            vias_by_net.setdefault(entry["net"], []).append(v)
        n_seg += len(segs)
        n_via += len(vias)
    text = splice(board.text, blocks)
    stats = {"entries": len(entries), "nets": len({e["net"] for e in entries}), "segments": n_seg, "vias": n_via,
             "copperLengthMm": round(length, 2), "segmentsByWidth": {str(k): v for k, v in sorted(widths.items())}}
    if fill_planes:
        text, fills = fill_zones(board, text, vias_by_net)
        stats["zoneFillPolygons"] = fills
    return text, stats


def main(argv=None):
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    ap.add_argument("board", help="the board WITH planes (add_planes.py output)")
    ap.add_argument("plan", help="ai_router.py plan JSON")
    ap.add_argument("--out", help="output .kicad_pcb (default: <plan>-routed.kicad_pcb beside the plan)")
    ap.add_argument("--no-fill-planes", action="store_true")
    ap.add_argument("--drc", action="store_true")
    ap.add_argument("--baseline", help="unrouted board whose DRC errors are inherited (default: the input board)")
    ap.add_argument("--report", help="write the DRC summary JSON here")
    a = ap.parse_args(argv)
    board = Board(a.board)
    plan = json.load(open(a.plan, encoding="utf-8"))
    text, stats = apply(board, plan, fill_planes=not a.no_fill_planes)
    out = Path(a.out) if a.out else Path(a.plan).with_name(Path(a.plan).stem + "-routed.kicad_pcb")
    out.write_text(text, encoding="utf-8", newline="")
    Board(str(out))
    print("%s: %s" % (out, json.dumps(stats)))
    if a.drc:
        base = run_drc(a.baseline or a.board)
        data = run_drc(out)
        s = summarize(data, base)
        s["copper"] = stats
        s["file"] = str(out)
        print("drc: %d errors (%d new, %d inherited), %d warnings (+%d raster-fill island warnings), %d unconnected; new error types %s; warning types %s; KiCad %s" % (
            s["errors"], s["newErrors"], s["inheritedErrors"], s["warnings"], s["rasterFillIslandWarnings"], s["unconnected"],
            json.dumps(s["newErrorTypes"], sort_keys=True), json.dumps(s["warningTypes"], sort_keys=True), s["kicadVersion"]))
        for v in s["newErrorList"][:40]:
            print("  NEW %s: %s %s" % (v["type"], v["description"], v["items"]))
        for u in s["unconnectedList"][:40]:
            print("  UNCONNECTED %s" % (u,))
        if a.report:
            Path(a.report).write_text(json.dumps(s, indent=1) + "\n", encoding="utf-8")
        return 0 if (s["newErrors"] == 0 and s["unconnected"] == 0) else 2
    return 0


if __name__ == "__main__":
    sys.exit(main())