KiCad - the KiCad Bridge

#!/usr/bin/env python3
"""Land a placement plan on the live board through the bridge, batch by batch.

Usage: place_live.py --target arav-rog --board "C:/.../esc-g431-unplaced.kicad_pcb" --plan esc-placement-plan.json
       [--batch 6] [--pause 1.2] [--thread "kicad-bridge-esc-demo"] [--log place-log.json]

Each batch is one kicad_move_footprint call (one native undo step) with the
expected revision; the reply's revision feeds the next call. Order: the macro
parts first (bridge, driver, MCU, rails), then each anchor's passives, so the
board grows outward from the power stage on camera. Every reply is logged with
timestamps; a stale_board or courtyard_overlap stops the run and is reported.
"""
import argparse, json, subprocess, sys, 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("--batch", type=int, default=6); ap.add_argument("--pause", type=float, default=1.2)
ap.add_argument("--thread", default="kicad-bridge-esc-demo"); ap.add_argument("--log", default="place-log.json")
ap.add_argument("--dry-run", action="store_true")
a = ap.parse_args()
T = ["adom-bridge", "--ai-thread", a.thread, "--target", a.target]

def call(verb, args, timeout=200):
    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)
    return r

plan = json.load(open(a.plan))["plan"]
# order: macro parts (no anchor) first, then passives grouped by anchor
macro = [r for r, p in plan.items() if not p.get("anchor")]
by_anchor = {}
for r, p in plan.items():
    if p.get("anchor"):
        by_anchor.setdefault(p["anchor"], []).append(r)
order = macro[:]
for m in macro:
    order += sorted(by_anchor.get(m, []))
for m, refs in by_anchor.items():
    if m not in macro:
        order += sorted(refs)
assert len(order) == len(plan), (len(order), len(plan))

st = call("kicad_placement_state", {"filePath": a.board})
if st.get("status") != "ok":
    print("placement_state failed", json.dumps(st)[:600]); sys.exit(1)
rev = st["revision"]
print(f"start: {len(st.get('footprints', []))} footprints, revision {rev[:8]}, ratsnest {st.get('ratsnest', {}).get('totalMm')} mm")
log = {"start": st, "batches": []}
batches = [order[i:i + a.batch] for i in range(0, len(order), a.batch)]
t_start = time.time()
for i, refs in enumerate(batches, 1):
    moves = [{"ref": r, "x": plan[r]["x"], "y": plan[r]["y"], "rotation": plan[r]["rotation"], "side": plan[r].get("side", "F.Cu")} for r in refs]
    args = {"filePath": a.board, "expectedRevision": rev, "refs": moves}
    if a.dry_run: args["dryRun"] = True
    r = call("kicad_move_footprint", args)
    ok = r.get("status") == "ok"
    print(f"batch {i}/{len(batches)} {refs}: {r.get('status')} {r.get('errorCode') or ''} rev {str(r.get('revision', ''))[:8]} {r['_dt']}s")
    log["batches"].append({"refs": refs, "reply": {k: v for k, v in r.items() if k not in ('_hint', '_timeoutHint')}})
    if not ok and not a.dry_run:
        print(json.dumps(r)[:1200]); break
    rev = r.get("revision") or rev
    time.sleep(a.pause)
v = call("kicad_placement_validate", {"filePath": a.board})
log["validate"] = v; log["wall_seconds"] = round(time.time() - t_start, 1)
print("validate:", {k: v.get(k) for k in ("status", "overlaps", "outside", "ratsnestMm", "errors", "warnings")}, "wall", log["wall_seconds"], "s")
json.dump(log, open(a.log, "w"), indent=1)