AI Flow

#!/usr/bin/env python3
"""Land a pours plan on the live board through kicad_add_zone, one undo step per group, then measure.

Usage: pours_live.py --target ConfRoomROG --board "C:/.../board.kicad_pcb" --pours pours.json
                     [--thread kicad-bridge-esc-routing] [--log pours-log.json] [--dry-run] [--pause 1.0]

The JSON is the tools/add_pours.py shape ({"zones": [...]}); "outline": true zones get the board
outline (inset by "inset" mm) resolved here from the local copy of the board given with --local,
because the verb takes explicit polygons. Zones are grouped by net into one kicad_add_zone call
each (one native undo step per net), lowest priority first. kicad_zone_state {refill:true} is read
before and after, so the log carries the copper kept per layer, the number this round is about.
"""
import argparse, json, subprocess, sys, time, os

ap = argparse.ArgumentParser()
ap.add_argument("--target", required=True); ap.add_argument("--board", required=True); ap.add_argument("--pours", required=True)
ap.add_argument("--local", help="local copy of the board, for outline: true zones")
ap.add_argument("--thread", default="kicad-bridge-esc-routing"); ap.add_argument("--log", default="pours-log.json")
ap.add_argument("--dry-run", action="store_true"); ap.add_argument("--pause", type=float, default=1.0)
a = ap.parse_args()
T = ["adom-bridge", "--ai-thread", a.thread, "--target", a.target]


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


spec = json.load(open(a.pours, encoding="utf-8"))
zones = spec["zones"] if isinstance(spec, dict) else spec
outline = None
if any(z.get("outline") for z in zones):
    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
    
    from escboard import Board, polygon_area
    from pours import inset_polygon
    b = Board(a.local)
    outline = list(b.outline)
    if polygon_area(outline) < 0:
        outline.reverse()
groups = {}
for z in zones:
    zz = {k: v for k, v in z.items() if k not in ("outline", "inset")}
    if z.get("outline"):
        zz["polygon"] = [[round(x, 3), round(y, 3)] for x, y in inset_polygon(outline, float(z.get("inset", 0.5)))]
    groups.setdefault(z["net"], []).append(zz)
order = sorted(groups, key=lambda n: min(z.get("priority", 0) for z in groups[n]))

before = call("kicad_zone_state", {"filePath": a.board, "refill": True})
print("before:", before.get("filledAreaByLayerMm2"), before.get("filledCoverageByLayerPercent"))
st = call("kicad_routing_state", {"filePath": a.board})
rev = st.get("revision")
if not rev:
    print("routing_state failed", json.dumps(st)[:600]); sys.exit(1)
log = {"before": {k: before.get(k) for k in ("filledAreaByLayerMm2", "filledCoverageByLayerPercent", "zoneCount")}, "calls": []}
t_start = time.time()
for net in order:
    args = {"filePath": a.board, "expectedRevision": rev, "zones": groups[net]}
    if a.dry_run:
        args["dryRun"] = True
    r = call("kicad_add_zone", args)
    ok = r.get("success") is True
    print("%-10s %d zone(s): %s %s ids %s refilled %s %.1fs %s" % (net, len(groups[net]), "ok" if ok else "FAILED", r.get("errorCode") or "", len(r.get("itemIds") or []), r.get("refilled"), r["_dt"], (r.get("postCommitError") or "")[:80]))
    log["calls"].append({"net": net, "zones": groups[net], "reply": {k: v for k, v in r.items() if k not in ("_hint",)}})
    if not ok:
        print(json.dumps(r)[:1500]); break
    rev = r.get("revision") or rev
    time.sleep(a.pause)
after = call("kicad_zone_state", {"filePath": a.board, "refill": True})
print("after:", after.get("filledAreaByLayerMm2"), after.get("filledCoverageByLayerPercent"))
v = call("kicad_routing_validate", {"filePath": a.board}) if not a.dry_run else {}
print("validate:", {k: v.get(k) for k in ("errors", "warnings", "unconnected", "clean", "zonesRefilled")})
log["after"] = {k: after.get(k) for k in ("filledAreaByLayerMm2", "filledCoverageByLayerPercent", "zoneCount", "zones")}
log["validate"] = v; log["wall_seconds"] = round(time.time() - t_start, 1)
json.dump(log, open(a.log, "w"), indent=1)