shibosoftwaredev/f1c100s-linux-nema17-stepper-controller

scripts/relocate-tight-vias.py

"""Relocate trapped drills using short, real per-layer connection stubs.

The old drill is removed, not hidden or shrunk. Every old copper junction is
retained and joined to the new full-through 0.30/0.45 mm via. Emits a candidate
only; all emitted-copper checks must pass before adoption.
"""
from route_geometry import *
from collections import defaultdict
from shapely.ops import unary_union
from shapely import contains_xy, prepare
from shapely.strtree import STRtree
import numpy as np
import heapq,time,sys

c=load_circuit(os.environ.get("VIA_CIRCUIT_INPUT","dist/index/circuit.json"));root=connectivity(c)
outward="--outward" in sys.argv
remove_smd="--smd" in sys.argv
ordinary_only="--ordinary" in sys.argv
thermal_relocation="--thermal-land" in sys.argv
search_radius=next((float(a.split("=",1)[1]) for a in sys.argv if a.startswith("--radius=")),None)
outward_launches={(4.199894,1.60001):(4.199894,1.00001),
                  (5.400044,10.39999):(5.400044,10.99999)}
sources={e["source_trace_id"]:e for e in c if e["type"]=="source_trace"}
ports={e["pcb_port_id"]:e for e in c if e["type"]=="pcb_port"}
components={e["pcb_component_id"]:e for e in c if e["type"]=="pcb_component"}
source_parts={e["source_component_id"]:e for e in c if e["type"]=="source_component"}
source_ports={e["source_port_id"]:e for e in c if e["type"]=="source_port"}
def net(t):
    s=sources[t["source_trace_id"]];ids=s.get("connected_source_port_ids",[])+s.get("connected_source_net_ids",[])
    return root(ids[0])
def key(x,y):return round(x,6),round(y,6)
routes=json.loads(Path(os.environ.get("VIA_REPLAY_INPUT","routing/complete-routes.json")).read_text());metal=[]
seen=set()
for t in routes:
    n=net(t)
    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
        aa,bb=key(a["x"],a["y"]),key(b["x"],b["y"])
        if math.dist(aa,bb)<1e-7:continue
        signature=n,a["layer"],tuple(sorted([aa,bb]))
        if signature in seen:continue
        seen.add(signature)
        metal.append(dict(net=n,layers=[a["layer"]],shape=LineString([aa,bb]).buffer(max(a["width"],b["width"])/2,quad_segs=16),
          pad=False,points=[aa,bb],width=max(a["width"],b["width"])))
for e in c:
    if e["type"] not in ["pcb_smtpad","pcb_plated_hole"]:continue
    p=ports.get(e.get("pcb_port_id"));n=root(p["source_port_id"]) if p else "NC:"+e[e["type"]+"_id"]
    component=components[e["pcb_component_id"]]
    bare_probe=component.get("do_not_place",False) and source_parts[component["source_component_id"]].get("ftype")=="simple_test_point"
    pin=source_ports.get(p.get("source_port_id"),{}).get("pin_number") if p else None
    metal.append(dict(net=n,layers=e.get("layers",[e.get("layer")]),
      shape=geometry(e,"outer_" if e["type"]=="pcb_plated_hole" else ""),pad=True,bare_probe=bare_probe,
      soc_thermal_land=source_parts[component["source_component_id"]]["name"]=="U1" and pin==89))
trace_by_id={e["pcb_trace_id"]:e for e in c if e["type"]=="pcb_trace"}
vias=[dict(x=e["x"],y=e["y"],net=net({"source_trace_id":e.get("source_trace_id",trace_by_id.get(e.get("pcb_trace_id"),{}).get("source_trace_id"))}))
      for e in c if e["type"]=="pcb_via"]
fixed=json.loads(Path("routing/native-features.json").read_text())
fixed_xy={key(v["x"],v["y"]) for v in fixed["thermal_vias"]+fixed.get("auxiliary_vias",[])}
ordinary_xy=set()
if ordinary_only:
    # Thermal-pad via arrays are a separate stencil/process review. Do not
    # disturb them while repairing ordinary populated component solder lands.
    pads_by_component={}
    source_ports={e["source_port_id"]:e for e in c if e["type"]=="source_port"}
    for e in c:
        if e["type"]!="pcb_smtpad":continue
        component=source_parts[components[e["pcb_component_id"]]["source_component_id"]]
        p=source_ports.get(ports.get(e.get("pcb_port_id"),{}).get("source_port_id"),{})
        if (component["name"],p.get("pin_number")) in {("U1",89),("U_MOTOR",29),("U_PD",17)}:continue
        if component.get("ftype")=="simple_test_point" or components[e["pcb_component_id"]].get("do_not_place"):continue
        for v in vias:
            if Point(v["x"],v["y"]).buffer(.15).intersection(geometry(e)).area>1e-7:
                ordinary_xy.add(key(v["x"],v["y"]))
holes=[geometry(e,"hole_") for e in c if e["type"]=="pcb_hole"]
tree=STRtree([m["shape"] for m in metal]);began=time.monotonic();accepted=[];failures=[]
def bad(v):
    p=Point(v["x"],v["y"])
    if remove_smd:
        return any(metal[int(i)]["pad"] and not metal[int(i)].get("bare_probe") and p.distance(metal[int(i)]["shape"])<.15-.0002
                   for i in tree.query(p.buffer(.15)))
    return any(metal[int(i)]["net"]!=v["net"] and p.distance(metal[int(i)]["shape"])<.35-.0002
               for i in tree.query(p.buffer(.35)))
for v in vias:
    if time.monotonic()-began>50:break
    if key(v["x"],v["y"]) in fixed_xy or not bad(v):continue
    if ordinary_only and key(v["x"],v["y"]) not in ordinary_xy:continue
    if outward and key(v["x"],v["y"]) not in outward_launches:continue
    if not outward and len(sys.argv)>2 and not sys.argv[1].startswith("--") and math.dist((v["x"],v["y"]),tuple(map(float,sys.argv[1:3])))>.001:continue
    old=key(v["x"],v["y"]);p=Point(old)
    refs=[(t,j,q) for t in routes if net(t)==v["net"] for j,q in enumerate(t["route"])
          if q["route_type"]=="via" and key(q["x"],q["y"])==old]
    if not refs:continue
    layers=set()
    for t,j,q in refs:layers.update([q["from_layer"],q["to_layer"]])
    own=[metal[int(i)] for i in tree.query(p.buffer(.225)) if metal[int(i)]["net"]==v["net"]
         and p.distance(metal[int(i)]["shape"])<.225]
    layers.update(l for m in own for l in m["layers"])
    # Do not lose an incidental annulus-edge contact which has no center stub.
    if any(not any(m["net"]==v["net"] and l in m["layers"] and m["shape"].buffer(.0001).covers(p) for m in own) for l in layers):
        failures.append(dict(old=old,reason="noncentral existing copper contact"));continue
    width=max(.12,max(q.get("width",.12) for t,j,z in refs for q in [t["route"][max(0,j-1)],t["route"][min(len(t["route"])-1,j+1)]]))
    layer_widths={l:max([.12]+[p["width"] for t,j,z in refs
        for p in [t["route"][max(0,j-1)],t["route"][min(len(t["route"])-1,j+1)]]
        if p["route_type"]=="wire" and p["layer"]==l]) for l in layers}
    pitch=.01;radius=search_radius if search_radius is not None else 3.5 if outward else 2.5;half=int(radius/pitch);size=2*half+1
    axisx=v["x"]+np.arange(-half,half+1)*pitch;axisy=v["y"]+np.arange(-half,half+1)*pitch
    xx,yy=np.meshgrid(axisx,axisy);region=p.buffer(radius*1.5)
    nearby=[metal[int(i)] for i in tree.query(region)]
    forbidden_vias=[];blocked={l:[] for l in layers}
    for m in nearby:
        if m["net"]!=v["net"]:
            forbidden_vias.append(m["shape"].buffer(.354))
            for l in set(m["layers"])&layers:blocked[l].append(m["shape"].buffer(.1002+layer_widths[l]/2))
        # Bare same-net factory probes are already via-in-pad features with
        # no fitted body or paste. This does not permit vias under SMT parts.
        # An existing SoC thermal stitch may move WITHIN its assigned thermal
        # land. No ordinary solder land or opposite-face component is exempt.
        assigned_thermal=thermal_relocation and m.get("soc_thermal_land") and m["net"]==v["net"] and m["shape"].covers(p)
        if m["pad"] and not(m["net"]==v["net"] and m.get("bare_probe")) and not assigned_thermal:
            forbidden_vias.append(m["shape"].buffer(.33 if remove_smd else .163))
    for q in vias:
        if q is v:continue
        g=Point(q["x"],q["y"]);forbidden_vias.append(g.buffer(.554))
        if q["net"]!=v["net"]:
            for l in layers:blocked[l].append(g.buffer(.354+layer_widths[l]/2))
    for h in holes:
        forbidden_vias.append(h.buffer(.429))
        for l in layers:blocked[l].append(h.buffer(.203+layer_widths[l]/2))
    for x in [-15.5,15.5]:
        for y in [-15.5,15.5]:
            for l in layers:
                r=2.85 if l=="top" else 2.6 if l=="bottom" else 0
                if r:blocked[l].append(Point(x,y).buffer(r+layer_widths[l]/2))
    def raster(items):
        g=unary_union(items);prepare(g);return contains_xy(g,xx,yy)
    via_ok=~raster(forbidden_vias)
    if thermal_relocation:
        assigned=[m["shape"].buffer(-.225) for m in metal if m.get("soc_thermal_land") and m["net"]==v["net"] and m["shape"].covers(p)]
        assert assigned,"thermal relocation must start in its original SoC thermal land"
        via_ok&=raster(assigned)
    via_ok&=(np.abs(xx)<20.72)&(np.abs(yy)<20.72)
    maps={l:raster(items)|((np.abs(xx)>20.95-layer_widths[l]/2)|(np.abs(yy)>20.95-layer_widths[l]/2)) for l,items in blocked.items()}
    distances={};previous={};launches={};directions=[(di,dj,math.hypot(di,dj)) for di in [-1,0,1] for dj in [-1,0,1] if di or dj]
    for l,m in maps.items():
        launch=outward_launches[old] if outward and l=="top" else old
        launch_point=Point(launch)
        # Dense existing 0.12 mm corridors may have no grid-aligned center.
        # Reuse their real, unchanged centerlines to reach an open launch
        # point rather than widening a corridor or deleting its obstacles.
        graph=defaultdict(list)
        for item in nearby:
            if item["net"]!=v["net"] or l not in item["layers"] or item["pad"] or item.get("width",0)<layer_widths[l]-1e-7:continue
            a,b=item["points"]
            graph[a].append((b,math.dist(a,b)));graph[b].append((a,math.dist(a,b)))
            if LineString([a,b]).distance(launch_point)<1e-6:
                for q in [a,b]:graph[launch].append((q,math.dist(launch,q)));graph[q].append((launch,math.dist(launch,q)))
        graph_dist={launch:0};graph_paths={launch:[launch]};queue=[(0,launch)]
        while queue:
            cost,point=heapq.heappop(queue)
            if cost>graph_dist[point]+1e-9:continue
            for q,d in graph[point]:
                new=cost+d
                if new<graph_dist.get(q,math.inf):
                    graph_dist[q]=new;graph_paths[q]=graph_paths[point]+[q];heapq.heappush(queue,(new,q))
        blocked_geometry=unary_union(blocked[l])
        candidates=[(point,path) for point,path in graph_paths.items()]
        for a,path in list(graph_paths.items()):
            for b,d in graph[a]:
                if d<.05:continue
                for t in np.arange(.05,d,.05):
                    q=(a[0]+(b[0]-a[0])*t/d,a[1]+(b[1]-a[1])*t/d)
                    candidates.append((q,path+[q]))
        seeds={}
        for point,path in candidates:
            if math.dist(point,old)>radius:continue
            ii=int(round((point[0]-v["x"])/pitch))+half;jj=int(round((point[1]-v["y"])/pitch))+half
            for i in range(max(0,ii-1),min(size,ii+2)):
                for j in range(max(0,jj-1),min(size,jj+2)):
                    if m[j,i]:continue
                    target_point=(float(axisx[i]),float(axisy[j]))
                    connector=LineString([point,target_point])
                    if connector.intersects(blocked_geometry):continue
                    prefix=path+[target_point];cost=sum(math.dist(a,b) for a,b in zip(prefix,prefix[1:]))/pitch
                    if (i,j) not in seeds or cost<seeds[i,j][0]:seeds[i,j]=(cost,prefix)
        launches[l]=seeds
        dist=np.full((size,size),np.inf);prev={};heap=[]
        for (i,j),(cost,prefix) in seeds.items():dist[j,i]=cost;heapq.heappush(heap,(cost,i,j))
        while heap:
            cost,i,j=heapq.heappop(heap)
            if cost>dist[j,i]+1e-8:continue
            for di,dj,step in directions:
                ii,jj=i+di,j+dj
                if not(0<=ii<size and 0<=jj<size) or m[jj,ii]:continue
                if di and dj and (m[j,ii] or m[jj,i]):continue
                new=cost+step
                if new<dist[jj,ii]:dist[jj,ii]=new;prev[ii,jj]=(i,j);heapq.heappush(heap,(new,ii,jj))
        distances[l]=dist;previous[l]=prev
    score=sum(distances.values())
    score[~via_ok]=np.inf
    # Don't replace a drill with an almost coincident near-duplicate.
    score[np.hypot(xx-v["x"],yy-v["y"])<.075]=np.inf
    j,i=np.unravel_index(np.argmin(score),score.shape)
    if not np.isfinite(score[j,i]):failures.append(dict(old=old,reason="no legal local via/stubs"));continue
    target=round(float(axisx[i]),6),round(float(axisy[j]),6);paths={}
    for l in layers:
        path=[(int(i),int(j))]
        while path[-1] in previous[l]:path.append(previous[l][path[-1]])
        prefix=launches[l][path[-1]][1]
        path.reverse();reduced=[path[0]]
        for a,b,d in zip(path,path[1:],path[2:]):
            if (b[0]-a[0],b[1]-a[1])!=(d[0]-b[0],d[1]-b[1]):reduced.append(b)
        reduced.append(path[-1]);paths[l]=prefix[:-1]+[(round(float(axisx[ii]),6),round(float(axisy[jj]),6)) for ii,jj in reduced]
    # Replace each actual old via occurrence with its checked round-trip stubs.
    for t,j,q in sorted(refs,key=lambda x:x[1],reverse=True):
        f,to=q["from_layer"],q["to_layer"]
        w=lambda points,layer:[dict(route_type="wire",layer=layer,width=layer_widths[layer],x=x,y=y) for x,y in points]
        beg,end=j,j+1
        if outward and f=="top":
            beg=max(i for i,z in enumerate(t["route"][:j]) if z["route_type"]=="wire" and z["layer"]=="top" and key(z["x"],z["y"])==outward_launches[old])
        if outward and to=="top":
            end=1+next(i for i,z in enumerate(t["route"]) if i>j and z["route_type"]=="wire" and z["layer"]=="top" and key(z["x"],z["y"])==outward_launches[old])
        replacement=w(paths[f],f)+[{**q,"x":target[0],"y":target[1]}]+w(list(reversed(paths[to])),to)
        # Circuit JSON consumers use the maximum endpoint width for a segment.
        # Keep width transitions at zero-length junctions, otherwise adding a
        # wide stub can silently widen the untouched preceding/following wire.
        previous=t["route"][beg-1] if beg else None
        following=t["route"][end] if end<len(t["route"]) else None
        if previous and previous["route_type"]=="wire" and previous["layer"]==f:
            replacement.insert(0,{**replacement[0],"width":previous["width"]})
        if following and following["route_type"]=="wire" and following["layer"]==to:
            replacement.append({**replacement[-1],"width":following["width"]})
        for field in ["start_pcb_port_id","end_pcb_port_id"]:
            if field in t["route"][beg]:replacement[0][field]=t["route"][beg][field]
            if field in t["route"][end-1]:replacement[-1][field]=t["route"][end-1][field]
        t["route"][beg:end]=replacement
    # Preserve contacts on layers that were incidental to the drilled barrel.
    explicit={l for t,j,q in refs for l in [q["from_layer"],q["to_layer"]]}
    for l in layers-explicit:
        owner=refs[0][0]
        routes.append({**{k:z for k,z in owner.items() if k!="route"},
          "pcb_trace_id":f"local_via_stub_{len(routes)}",
          "route":[dict(route_type="wire",layer=l,width=layer_widths[l],x=x,y=y) for x,y in paths[l]]})
    for l,points in paths.items():
        for a,b in zip(points,points[1:]):
            if math.dist(a,b)>1e-7:metal.append(dict(net=v["net"],layers=[l],shape=LineString([a,b]).buffer(layer_widths[l]/2),pad=False,points=[key(*a),key(*b)],width=layer_widths[l]))
    accepted.append(dict(old=old,new=target,width=width,layerWidths=layer_widths,paths=paths))
    v["x"],v["y"]=target;tree=STRtree([m["shape"] for m in metal])
Path("routing/relocated-vias-candidate.json").write_text(json.dumps(routes,indent=2)+"\n")
report=dict(seconds=round(time.monotonic()-began,2),accepted=accepted,failures=failures,notInstalled=True,
            offendingViasRemaining=sum(bad(v) for v in vias))
Path("reports/relocated-vias-candidate.json").write_text(json.dumps(report,indent=2)+"\n")
print(json.dumps({k:v for k,v in report.items() if k not in ["accepted","failures"]}|{"accepted":len(accepted),"rejected":len(failures)}))