shibosoftwaredev/f1c100s-linux-nema17-stepper-controller

scripts/prepare-usb-top.py

"""Prepare a bounded, physically checked USB-face change; never relax DRC."""
from route_geometry import *
from collections import defaultdict
from shapely.strtree import STRtree

baseline=Path("../../checkpoints/nema17-before-usb-top-v052-UkdpHy/restore-verified")
c=load_circuit("dist/index/circuit.json");old=load_circuit(str(baseline/"dist/index/circuit.json"))
assert not [e for e in c if e["type"].endswith("_error")],"Resolve native placement errors before routing"
root=connectivity(c);mapping=json.loads(Path("routing/net-map.json").read_text())["roots"]
sources={e["source_trace_id"]:e for e in c if e["type"]=="source_trace"}
named={e.get("name"):e for e in sources.values()}
parts={e["source_component_id"]:e["name"] for e in c if e["type"]=="source_component"}
pcnames={e["pcb_component_id"]:parts[e["source_component_id"]] for e in c if e["type"]=="pcb_component"}
ports={e["pcb_port_id"]:e for e in c if e["type"]=="pcb_port"}
by_source={p["source_port_id"]:p for p in ports.values()}
moved=set(json.loads(Path("usb-top-placement.json").read_text()))
newpads=[]
for e in c:
    if e["type"] in ["pcb_smtpad","pcb_plated_hole"] and pcnames[e["pcb_component_id"]] in moved:
        p=ports.get(e.get("pcb_port_id"))
        n=mapping[root(p["source_port_id"])] if p else "NC:"+e[e["type"]+"_id"]
        newpads.append((geometry(e,"outer_" if e["type"]=="pcb_plated_hole" else ""),e.get("layers",[e.get("layer")]),n))
def net(t):
    s=sources[t["source_trace_id"]]
    return mapping[root((s.get("connected_source_port_ids",[])+s.get("connected_source_net_ids",[]))[0])]
original=json.loads((baseline/"routing/complete-routes.json").read_text())
bad_vias=set()
for t in original:
    for p in t["route"]:
        if p["route_type"]=="via" and any(g.distance(Point(p["x"],p["y"]))<.351 for g,ls,n in newpads):
            bad_vias.add((round(p["x"],6),round(p["y"],6)))
retained=[];removed=[];seen=set()
for t in original:
    n=net(t)
    for p in t["route"]:
        if p["route_type"]!="via":continue
        key=round(p["x"],6),round(p["y"],6)
        if key in bad_vias:continue
        k=(n,"via",key)
        if k in seen:continue
        seen.add(k)
        retained.append({**t,"route":[p]})
    for a,b in zip(t["route"],t["route"][1:]):
        if a["route_type"]!=b["route_type"] or a["route_type"]!="wire" or a["layer"]!=b["layer"]:continue
        if math.dist((a["x"],a["y"]),(b["x"],b["y"]))<1e-8:continue
        width=max(a["width"],b["width"]);g=LineString([(a["x"],a["y"]),(b["x"],b["y"])]).buffer(width/2)
        key=n,a["layer"],tuple(sorted([(a["x"],a["y"]),(b["x"],b["y"])])),width
        if key in seen:continue
        seen.add(key)
        if any(nn!=n and a["layer"] in ls and g.distance(h)<.1001 for h,ls,nn in newpads):
            removed.append(dict(source=t["source_trace_id"],net=n,layer=a["layer"],width=width,points=[(a["x"],a["y"]),(b["x"],b["y"])]))
            continue
        route=[dict(a),dict(b)]
        for p in route:
            p["width"]=width
            for field in ["start_pcb_port_id","end_pcb_port_id"]:
                q=ports.get(p.get(field))
                if q and (p["layer"] not in q["layers"] or math.dist((p["x"],p["y"]),(q["x"],q["y"]))>1e-6):p.pop(field)
        retained.append({**t,"route":route})
# Remove only obsolete, unanchored leaf copper left by the face changes.
# Stop at a real pad, through-via, junction or another touching same-net wire.
# This opens the new VBUS escape without deleting a working rail/ground tree.
allpads=[]
for e in c:
    if e["type"] in ["pcb_smtpad","pcb_plated_hole"]:
        p=ports.get(e.get("pcb_port_id"));nn=mapping[root(p["source_port_id"])] if p else None
        allpads.append((nn,e.get("layers",[e.get("layer")]),geometry(e,"outer_" if e["type"]=="pcb_plated_hole" else "")))
pruned=0
for iteration in range(100):
    wire_rows=[(i,t) for i,t in enumerate(retained) if len(t["route"])==2]
    degree=defaultdict(list);shapes=[]
    for i,t in wire_rows:
        a,b=t["route"];nn=net(t);l=a["layer"]
        shapes.append(LineString([(a["x"],a["y"]),(b["x"],b["y"])]).buffer(a["width"]/2))
        for p in [a,b]:degree[(nn,l,round(p["x"],6),round(p["y"],6))].append(i)
    tree=STRtree(shapes);vias={(net(t),round(p["x"],6),round(p["y"],6)) for t in retained for p in t["route"] if p["route_type"]=="via"}
    remove=set()
    for (nn,l,x,y),ids in degree.items():
        if len(ids)!=1 or (nn,x,y) in vias:continue
        i=ids[0];t=retained[i];point=Point(x,y);allow=t["route"][0]["width"]/2+1e-5
        if any(n==nn and l in ls and g.distance(point)<=allow for n,ls,g in allpads):continue
        if any(wire_rows[int(j)][0]!=i and net(wire_rows[int(j)][1])==nn and wire_rows[int(j)][1]["route"][0]["layer"]==l and shapes[int(j)].distance(point)<1e-6 for j in tree.query(point)):continue
        remove.add(i)
    if not remove:break
    retained=[t for i,t in enumerate(retained) if i not in remove];pruned+=len(remove)
assert iteration<99
# Authored power fanout: short 0.35 mm connector escapes reach 0.30/0.45 mm
# through-vias, then rejoin the original 0.80 mm outer-layer VBUS trunks.
# No thin monitor branch is used as a substitute for those power trunks.
def authored(sid,layer,width,points,via=False):
    route=[dict(route_type="wire",x=x,y=y,layer=layer,width=width) for x,y in points]
    if via:route.append(dict(route_type="via",x=points[-1][0],y=points[-1][1],from_layer=layer,to_layer="bottom" if layer=="top" else "top",via_diameter=.45,via_hole_diameter=.3))
    retained.append(dict(type="pcb_trace",source_trace_id=sid,route=route))
for t in original:
    if net(t)!="N99":continue
    for a,b in zip(t["route"],t["route"][1:]):
        if a["route_type"]==b["route_type"]=="wire" and a["layer"]==b["layer"] and a["width"]>=.8-1e-8 and b["width"]>=.8-1e-8:
            retained.append({**t,"route":[a,b]})
authored("source_trace_83","top",.35,[(4.5998397,13.07585565),(4.7,13.8),(5.1,14.2),(5.8,14.9),(5.8,15.7),(6.1,16),(6.1,17.1)])
authored("source_trace_83","top",.8,[(6.1,17.1),(9.05,17.1)])
authored("source_trace_84","top",.35,[(9.3999571,13.07602075),(9.25,13.8),(9.05,14.2),(9.05,17.1)])
for i,t in enumerate(retained):t["pcb_trace_id"]=f"usb_top_retained_{i}"
def endpoint(p):return {k:p[k] for k in ["x","y","layers","pcb_port_id"]}
jobs=[]
def job(source,start,target=None,constraints=None):
    jobs.append(dict(source=source,start=endpoint(start),**({"target":{**endpoint(target),"connected_tree":True}} if target else {}),constraints=constraints or {}))
# Every relocated bypass/reference terminal must reach the original SoC net,
# not just a newly floating nearby wire. The final emitted-only continuity
# audit independently checks every source connection across the entire board.
for name,oldp in [(pcnames[p["pcb_component_id"]],p) for p in ports.values() if pcnames[p["pcb_component_id"]] in moved and pcnames[p["pcb_component_id"]]!="J_USB_PWR"]:
    s=next(s for s in sources.values() if oldp["source_port_id"] in s.get("connected_source_port_ids",[]))
    peers=[by_source[x] for x in s["connected_source_port_ids"] if x in by_source and parts.get(next(e["source_component_id"] for e in c if e["type"]=="pcb_component" and e["pcb_component_id"]==by_source[x]["pcb_component_id"]))=="U1"]
    assert peers,(name,s["name"])
    job(s["source_trace_id"],oldp,peers[0])
for name in ["USB_VBUS_A","USB_VBUS_B","PD_CC1_TO_SINK_CC2","PD_CC2_TO_SINK_CC1","USB_PWR_SIGNAL_GND","USB_PWR_SHELL_GND","USB_PWR_BOARD_GND"]:
    s=named[name]
    constraints={"min_trace_thickness":.35,"allowed_layers":["top","bottom"]} if name.startswith("USB_VBUS") else {}
    pp=[by_source[x] for x in s["connected_source_port_ids"]]
    if name=="USB_VBUS_B":
        constraints={"min_trace_thickness":.8,"allowed_layers":["top"]}
        jobs.append(dict(source=s["source_trace_id"],start=dict(x=9.05,y=17.1,layers=["top"],pcb_port_id="authored_VBUS_bus"),target={**endpoint(pp[1]),"connected_tree":True},constraints=constraints))
        continue
    jobs.append(dict(source=s["source_trace_id"],start=endpoint(pp[0]),target={**endpoint(pp[1]),"connected_tree":True},constraints=constraints))
# Reconnect any other real terminals whose original segment crossed a new
# land. Whole-board emitted-only import remains the authority for completion.
for sid in sorted({w["source"] for w in removed}):
    s=sources[sid]
    pp=[by_source[x] for x in s["connected_source_port_ids"] if x in by_source]
    soc=[p for p in pp if pcnames[p["pcb_component_id"]]=="U1"]
    if not soc:continue
    for p in pp:
        if p==soc[0] or pcnames[p["pcb_component_id"]] in moved:continue
        if -4<=p["x"]<=20.9 and -2<=p["y"]<=20.9:job(sid,p,soc[0])
# Preserve a legal escape from the fine-pitch SD data pin before new USB
# fanouts claim that channel. Its old long segment crosses the new USB bank.
s=named["N_SDMMC0_D0"]
job(s["source_trace_id"],by_source[s["connected_source_port_ids"][0]],by_source[s["connected_source_port_ids"][1]])
s=named["N_VRA1"]
job(s["source_trace_id"],by_source[s["connected_source_port_ids"][0]],by_source[s["connected_source_port_ids"][1]])
# Power and CC fanouts first, before the low-current bypass leaves.
jobs.sort(key=lambda j:-3 if sources[j["source"]].get("name")=="N_VRA1" and pcnames[ports[j["start"]["pcb_port_id"]]["pcb_component_id"]]=="U1" else -2 if sources[j["source"]].get("name")=="N_SDMMC0_D0" and pcnames[ports[j["start"]["pcb_port_id"]]["pcb_component_id"]]=="U1" else 0 if sources[j["source"]].get("name","").startswith("PD_CC") else 1 if sources[j["source"]].get("name","").startswith("USB_VBUS") else 2)
def write(p,v):Path(p).write_text(json.dumps(v,indent=2)+"\n")
write("routing/usb-top-placement.circuit.json",c)
write("routing/usb-top-retained.json",retained)
write("routing/usb-top-plan.json",dict(prefix="usb-top",placement="routing/usb-top-placement.circuit.json",retained="routing/usb-top-retained.json",extent=[-4,-2,20.9,20.9],jobs=jobs))
write("reports/usb-top-routing-preparation.json",dict(moved=sorted(moved),removedSegments=removed,prunedObsoleteLeafSegments=pruned,removedVias=sorted(bad_vias),retainedSegments=len(retained),jobs=jobs))
print(json.dumps(dict(retained=len(retained),removedSegments=len(removed),prunedLeaves=pruned,removedVias=len(bad_vias),jobs=len(jobs))))