OCuLink Riser for the Supermicro AIOM Slot

#!/usr/bin/env python3
"""Land design/silk_plan.json on the live board through the KiCad bridge, in small groups (one undo step each), so
the silkscreen step's clip shows the labels going on. Writes design/silk_events.json (time, group, ids, reason)."""
import json, os, subprocess, sys, time

ROOT = os.path.join(os.path.dirname(__file__), "..")
BOARD = "C:/Users/john/Documents/adom-aiflow/ocp-oculink-riser/ocp-oculink-riser.kicad_pcb"
THREAD = "OCP OCuLink riser"
plan = json.load(open(os.path.join(ROOT, "design", "silk_plan.json")))
events = []


def bridge(verb, args):
    args = dict(args, reason=args.get("reason", "silkscreen step: land the reviewed label plan"))
    out = subprocess.run(["adom-bridge", "--ai-thread", THREAD, verb, json.dumps(args)], capture_output=True, text=True, timeout=300)
    try:
        d = json.loads(out.stdout)
    except Exception:
        raise SystemExit(f"{verb}: unreadable reply {out.stdout[:300]} {out.stderr[:300]}")
    r = d.get("result", d)
    if isinstance(r, dict) and "output" in r and isinstance(r["output"], str):
        try:
            r = json.loads(r["output"]).get("data", r)
        except Exception:
            pass
    return d, r


def revision():
    _, r = bridge("kicad_placement_state", {"filePath": BOARD})
    return r.get("revision")


def group_of(item):
    if item.get("field"):
        ref = item["ref"]
        y = item["y"]
        return f"refs {'faceplate' if y < 20 else 'logic' if y < 75 else 'edge'}"
    t = item["text"]
    if item["layer"] == "B.SilkS":
        return "bottom face"
    if t.startswith(("OCP_B", "B")) and item["y"] > 105:
        return "finger pins"
    if t.startswith("J") or "PORT" in t:
        return "ports"
    return "labels"


ORDER = ["finger pins", "ports", "refs faceplate", "refs logic", "labels", "refs edge", "bottom face"]
# KiCad 10.0.5's IPC server cannot update footprint fields: references/values go in by a disclosed saved-file edit
# (gen/silk.py apply on the saved board); this script lands the standalone texts live. ONLY limits a rerun.
ONLY = sys.argv[1].split(",") if len(sys.argv) > 1 else ORDER
groups = {g: [] for g in ORDER}
for f in plan["fields"]:
    groups[group_of(f)].append(("field", f))
for t in plan["texts"]:
    groups[group_of(t)].append(("text", t))

rev = revision()
for g in ORDER:
    if g not in ONLY:
        continue
    items = [it for it in groups[g] if it[0] == "text"]
    # small batches: a ref/value pair or a few labels at a time, in reading order
    items.sort(key=lambda it: (round(it[1]["y"] / 6), it[1]["x"]))
    for k in range(0, len(items), 4):
        chunk = items[k:k + 4]
        fields = [dict(id=it["id"], x=it["x"], y=it["y"], rotation=it["rotation"], layer=it["layer"], height=it["height"],
                       width=it["height"], stroke=it["stroke"], visible=True) for kind, it in chunk if kind == "field"]
        texts = [dict(text=it["text"], layer=it["layer"], x=it["x"], y=it["y"], height=it["height"], width=it["height"],
                      stroke=it["stroke"], rotation=it["rotation"], align=it["align"]) for kind, it in chunk if kind == "text"]
        t0 = time.time()
        if fields:
            d, r = bridge("kicad_silk_fields_batch", {"filePath": BOARD, "expectedRevision": rev, "update": fields})
            if d.get("status") != "ok" and not r.get("revision"):
                print("fields batch refused:", json.dumps(d)[:600]); sys.exit(1)
            rev = r.get("revision") or revision()
        if texts:
            d, r = bridge("kicad_silk_text_batch", {"filePath": BOARD, "expectedRevision": rev, "create": texts})
            if d.get("status") != "ok" and not r.get("revision"):
                print("text batch refused:", json.dumps(d)[:600]); sys.exit(1)
            rev = r.get("revision") or revision()
        events.append(dict(t=round(t0, 1), group=g, op="place",
                           items=[(it.get("ref") or "") + ("." + it["field"] if kind == "field" else "") + (it.get("text", "") if kind == "text" else "")
                                  for kind, it in chunk],
                           reason=[it.get("why") or f"{it['ref']} {it['field'].lower()} beside its part" for kind, it in chunk]))
        print(f"{g}: {', '.join(events[-1]['items'])}", flush=True)
        time.sleep(1.2)
_ev = os.path.join(ROOT, "design", "silk_events.json")
prev = json.load(open(_ev)) if os.path.exists(_ev) and len(sys.argv) > 1 else []
json.dump(prev + events, open(_ev, "w"), indent=1)
print("landed", sum(len(e["items"]) for e in events), "labels in", len(events), "steps; revision", rev)