KiCad - the KiCad Bridge

#!/usr/bin/env python3
"""Replay a routing plan on the live board through the bridge, one kicad_route_net per entry.

Usage: route_live.py --target arav-rog --board "C:/.../esc-g431-fable-placed-planes.kicad_pcb" \\
                     --plan fable-esc-routing-plan.json [--pause 0.6] [--thread "kicad-bridge-esc-routing"] \\
                     [--log route-log.json] [--dry-run] [--start N] [--save]

Mirrors demo/placement/fable/place_live.py: kicad_routing_state gives the revision, every
plan entry becomes one kicad_route_net call (one native Undo step, DRC-checked by the bridge
against a snapshot) with expectedRevision from the previous reply, and the reply's revision
feeds the next call. Every reply is logged with timestamps. A drc_rejected or stale_board
reply stops the run and is printed in full; kicad_routing_validate closes the run and the
whole log is written as JSON. --dry-run sends every call with dryRun:true (nothing lands, the
revision never changes). --start N skips the first N entries (resume after a stop).
"""
import argparse
import json
import subprocess
import sys
import time

ap = argparse.ArgumentParser()
ap.add_argument("--target", required=True); ap.add_argument("--board", required=True); ap.add_argument("--plan", required=True)
ap.add_argument("--pause", type=float, default=0.6)
ap.add_argument("--thread", default="kicad-bridge-esc-routing"); ap.add_argument("--log", default="route-log.json")
ap.add_argument("--dry-run", action="store_true")
ap.add_argument("--start", type=int, default=0, help="skip the first N plan entries")
ap.add_argument("--save", action="store_true", help="ask the bridge to save the board after the last entry")
a = ap.parse_args()
T = ["adom-bridge", "--ai-thread", a.thread, "--target", a.target]


def call(verb, args, timeout=240):
    t0 = time.time()
    out = subprocess.run(T + [verb, json.dumps(args)], capture_output=True, text=True, timeout=timeout).stdout
    try:
        r = json.loads(out)
    except Exception:
        r = {"status": "parse_error", "raw": out[:400]}
    r["_t0"] = t0; r["_dt"] = round(time.time() - t0, 2); r["_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(t0))
    return r


plan = json.load(open(a.plan, encoding="utf-8"))
entries = plan["nets"] if isinstance(plan, dict) else plan
entries = entries[a.start:]

st = call("kicad_routing_state", {"filePath": a.board})
if st.get("success") is False or st.get("status") == "error" or not st.get("revision"):
    print("routing_state failed", json.dumps(st)[:800]); sys.exit(1)
rev = st["revision"]
print(f"start: revision {rev[:8]}, {len(st.get('pads', []))} pads, {len(st.get('segments', []))} segments, {len(st.get('vias', []))} vias on the board, "
      f"{len(st.get('netsRemaining', []))} nets remaining; {len(entries)} plan entries{' (dry run)' if a.dry_run else ''}")
log = {"start": {k: v for k, v in st.items() if k not in ("pads",)}, "plan": a.plan, "board": a.board, "target": a.target,
       "dryRun": a.dry_run, "entries": []}
t_start = time.time()
ok_count = 0
for i, e in enumerate(entries, 1):
    args = {"filePath": a.board, "expectedRevision": rev, "net": e["net"], "width": e["width"]}
    for k in ("viaSize", "viaDrill", "layer"):
        if k in e:
            args[k] = e[k]
    if "paths" in e:
        args["paths"] = e["paths"]
    else:
        args["points"] = e["points"]
    if a.dry_run:
        args["dryRun"] = True
    if a.save and i == len(entries) and not a.dry_run:
        args["save"] = True
    r = call("kicad_route_net", args)
    ok = r.get("success", True) is not False and r.get("status") != "error" and r.get("status") != "parse_error"
    code = r.get("errorCode") or ""
    drc = r.get("drc") or {}
    n_items = len(r.get("itemIds") or []) if not a.dry_run else len(r.get("segments") or []) + len(r.get("vias") or [])
    paths = len(e.get("paths") or [1])
    print(f"{i}/{len(entries)} {e['net']:<26} w{e['width']:<5} {paths} path(s): {'ok' if ok else 'FAIL'} {code} "
          f"items {n_items} drc errors {drc.get('errors')} unconnected {drc.get('unconnected')} rev {str(r.get('revision', ''))[:8]} {r['_dt']}s")
    log["entries"].append({"index": i + a.start, "net": e["net"], "width": e["width"], "reply": {k: v for k, v in r.items() if k not in ("_hint", "_timeoutHint")}})
    if not ok:
        print(json.dumps(r)[:2000])
        if code in ("drc_rejected", "stale_board") or not a.dry_run:
            break
    else:
        ok_count += 1
    if not a.dry_run:
        rev = r.get("revision") or rev
    time.sleep(a.pause)
v = call("kicad_routing_validate", {"filePath": a.board}, timeout=600)
log["validate"] = {k: val for k, val in v.items() if k not in ("_hint",)}
log["wall_seconds"] = round(time.time() - t_start, 1)
log["ok_entries"] = ok_count
print("validate:", {k: v.get(k) for k in ("status", "success", "errors", "warnings", "unconnected", "clean", "stale", "zonesRefilled")}, "wall", log["wall_seconds"], "s")
json.dump(log, open(a.log, "w"), indent=1)
print("log:", a.log)