shibosoftwaredev/f1c100s-linux-nema17-stepper-controller

scripts/finish-review-copper.py

"""Explicit clearance-polygon fanouts for this board's remaining signal links.

Retains all existing power trunks. Does not change the replay guard or router.
Full-stack stitches and each line segment are subsequently audited and imported
against the original source graph; this file cannot approve fabrication.
"""
from route_geometry import *
from shapely.ops import unary_union
from shapely.strtree import STRtree
from collections import defaultdict,deque
import subprocess,heapq
c=load_circuit('routing/review-placement.circuit.json');root=connectivity(c)
mapping=json.loads(Path('routing/net-map.json').read_text())['roots']
ports={e['pcb_port_id']:e for e in c if e['type']=='pcb_port'}
sources={e['source_trace_id']:e for e in c if e['type']=='source_trace'}
def net(source):
    s=sources[source]
    return mapping[root((s.get('connected_source_port_ids',[])+s.get('connected_source_net_ids',[]))[0])]
focused=bool(os.environ.get('FOCUSED_REVIEW'))
physical=json.loads(Path(os.environ.get('FINISH_PHYSICAL','routing/bulk-top-closure-partial-physical-candidate.json')).read_text())
restore=dict(removeVias=[],removeWires=[],wires=[],vias=[]) if focused else json.loads(Path('routing/review-restored-status.json').read_text())
for v in restore['removeVias']:
    matches=[p for p in physical['vias'] if p['net']==v['net'] and math.dist((p['x'],p['y']),(v['x'],v['y']))<1e-7]
    assert len(matches)==1;physical['vias'].remove(matches[0])
for w in restore['removeWires']:
    matches=[p for p in physical['wires'] if p['net']==w['net'] and p['layer']==w['layer'] and p['points']==w['points']]
    assert len(matches)==1;physical['wires'].remove(matches[0])
physical['wires']+=restore['wires'];physical['vias']+=restore['vias']
Path('routing/review-latest-seed.json').write_text(json.dumps(physical,indent=2)+'\n')
plan=json.loads(Path(os.environ.get('FINISH_PLAN','routing/bulk-top-final-plan.json')).read_text())
attempt=dict(routes=[]) if focused else json.loads(Path('reports/bulk-top-closure-attempt.json').read_text())
jobs=plan['jobs'][len(attempt['routes']):]
authored=dict(wires=[],vias=[]);completed=[]
def polygon_path(polygon,endpoints):
    a,b=endpoints
    if math.dist(a,b)<1e-8:return [a]
    if polygon.covers(LineString([a,b])):return [a,b]
    raw=subprocess.run(['/opt/homebrew/bin/bun','scripts/triangulate-review-polygon.mjs'],input=json.dumps(dict(
        outer=list(polygon.exterior.coords),holes=[list(h.coords) for h in polygon.interiors])),text=True,capture_output=True,check=True)
    triangles=[Polygon(xy) for xy in json.loads(raw.stdout)]
    tree=STRtree(triangles)
    start=next(int(i) for i in tree.query(Point(a)) if triangles[int(i)].buffer(1e-9).covers(Point(a)))
    goals={int(i) for i in tree.query(Point(b)) if triangles[int(i)].buffer(1e-9).covers(Point(b))}
    edges={};adj=defaultdict(list)
    for i,t in enumerate(triangles):
        xy=list(t.exterior.coords)
        for u,v in zip(xy,xy[1:]):
            key=tuple(sorted([tuple(u),tuple(v)]))
            if key in edges:
                j=edges[key];p=((u[0]+v[0])/2,(u[1]+v[1])/2)
                adj[i].append((j,p));adj[j].append((i,p))
            else:edges[key]=i
    queue=[(math.dist(a,b),0,start,a)];previous={start:None};cost={start:0};found=None
    while queue:
        _,distance,i,position=heapq.heappop(queue)
        if distance>cost[i]+1e-9:continue
        if i in goals:found=i;break
        for j,p in adj[i]:
            total=distance+math.dist(position,p)
            if total<cost.get(j,math.inf):
                cost[j]=total;previous[j]=(i,p);heapq.heappush(queue,(total+math.dist(p,b),total,j,p))
    assert found is not None,('No exact triangle path',a,b)
    points=[b]
    while found!=start:found,p=previous[found];points.append(p)
    points.append(a);points.reverse()
    simple=[a];i=0
    allowed=polygon.buffer(1e-10)
    while i<len(points)-1:
        j=next(j for j in range(len(points)-1,i,-1) if allowed.covers(LineString([points[i],points[j]])))
        simple.append(points[j]);i=j
    assert polygon.buffer(1e-9).covers(LineString(simple))
    return simple
for job in jobs:
    n=net(job['source']);width=job['constraints']['min_trace_thickness'];a=job['start'];b=job['target']
    for p in [a,b]:
        if p['pcb_port_id'] in ports:
            real=ports[p['pcb_port_id']];p.update(x=real['x'],y=real['y'],layers=real['layers'])
    assert width<.299,('High-current trunks must remain explicitly engineered on outer copper',job)
    blocked={l:[] for l in LAYERS};via_bad=[]
    for e in c:
        if e['type']=='pcb_hole':
            g=geometry(e,'hole_');via_bad.append(g.buffer(.429))
            for l in LAYERS:blocked[l].append(g.buffer(.2005+width/2))
        if e['type'] not in ['pcb_smtpad','pcb_plated_hole']:continue
        p=ports.get(e.get('pcb_port_id'));nn=mapping[root(p['source_port_id'])] if p else 'NC'
        g=geometry(e,'outer_' if e['type']=='pcb_plated_hole' else '')
        via_bad.append(g.buffer(.354))
        if nn!=n:
            for l in e.get('layers',[e.get('layer')]):blocked[l].append(g.buffer(.1005+width/2))
        if e['type']=='pcb_plated_hole':
            h=geometry(e,'hole_');via_bad.append(h.buffer(.429))
            if nn!=n:
                for l in LAYERS:blocked[l].append(h.buffer((.3005 if l.startswith('inner') else .2805)+width/2))
    for w in physical['wires']:
        if w['net']==n:continue
        g=LineString(w['points']).buffer(w['width']/2)
        blocked[w['layer']].append(g.buffer(.1005+width/2));via_bad.append(g.buffer(.354))
    for v in physical['vias']:
        p=Point(v['x'],v['y']);via_bad.append(p.buffer(.554))
        if v['net']!=n:
            for l in LAYERS:blocked[l].append(p.buffer(.354+width/2))
    for x in [-15.5,15.5]:
        for y in [-15.5,15.5]:
            p=Point(x,y);via_bad.append(p.buffer(3.1))
            blocked['top'].append(p.buffer(2.8505+width/2));blocked['bottom'].append(p.buffer(2.6005+width/2))
    edge=20.9495-width/2
    regions=[]
    for l in job['constraints'].get('allowed_layers',LAYERS):
        free=box(-edge,-edge,edge,edge).difference(unary_union(blocked[l]))
        regions += [(l,g) for g in (list(free.geoms) if hasattr(free,'geoms') else [free]) if g.area>1e-12]
    tree=STRtree([g for l,g in regions]);adj=defaultdict(list)
    available=box(-20.7,-20.7,20.7,20.7).difference(unary_union(via_bad))
    spaces=list(available.geoms) if hasattr(available,'geoms') else [available]
    spaces.sort(key=lambda g:g.distance(Point(a['x'],a['y']))+g.distance(Point(b['x'],b['y'])))
    for g in spaces:
        if g.is_empty:continue
        q=g.representative_point();xy=(q.x,q.y)
        touching=[int(i) for i in tree.query(q) if regions[int(i)][1].covers(q)]
        for i in touching:
            for j in touching:
                if i!=j:adj[i].append((j,xy))
    aa=Point(a['x'],a['y']);bb=Point(b['x'],b['y'])
    own=[]
    for w in physical['wires']:
        if w['net']==n:
            line=LineString(w['points']);own.append(dict(shape=line.buffer(w['width']/2),center=line,layers=[w['layer']]))
    for v in physical['vias']:
        if v['net']==n:
            p=Point(v['x'],v['y']);own.append(dict(shape=p.buffer(.225),center=p,layers=LAYERS))
    for e in c:
        if e['type'] not in ['pcb_smtpad','pcb_plated_hole'] or e.get('pcb_port_id') not in ports:continue
        p=ports[e['pcb_port_id']]
        if mapping[root(p['source_port_id'])]!=n:continue
        g=geometry(e,'outer_' if e['type']=='pcb_plated_hole' else '')
        own.append(dict(shape=g,center=g.buffer(-width/2-.01),layers=e.get('layers',[e.get('layer')])))
    own_tree=STRtree([o['shape'] for o in own]);todo=[int(i) for i in own_tree.query(bb) if own[int(i)]['shape'].buffer(1e-8).covers(bb) and set(own[int(i)]['layers'])&set(b['layers'])];target_indices=set()
    while todo:
        i=todo.pop()
        if i in target_indices:continue
        target_indices.add(i);o=own[i]
        for j in own_tree.query(o['shape'].buffer(1e-8)):
            j=int(j);p=own[j]
            if j not in target_indices and set(o['layers'])&set(p['layers']) and o['shape'].distance(p['shape'])<1e-8:todo.append(j)
    assert target_indices,('Landing must belong to actual retained copper',job)
    if any(own[i]['shape'].buffer(1e-8).covers(aa) and set(own[i]['layers'])&set(a['layers']) for i in target_indices):
        completed.append(dict(source=job['source'],net=n,alreadyConnectedByActualCopper=True))
        print(json.dumps(completed[-1]),flush=True)
        continue
    todo=[int(i) for i in own_tree.query(aa) if own[int(i)]['shape'].buffer(1e-8).covers(aa) and set(own[int(i)]['layers'])&set(a['layers'])];source_indices=set()
    while todo:
        i=todo.pop()
        if i in source_indices:continue
        source_indices.add(i);o=own[i]
        for j in own_tree.query(o['shape'].buffer(1e-8)):
            j=int(j);p=own[j]
            if j not in source_indices and set(o['layers'])&set(p['layers']) and o['shape'].distance(p['shape'])<1e-8:todo.append(j)
    assert source_indices,('Start must belong to actual retained copper',job)
    starts={}
    for j in source_indices:
        o=own[j]
        for i in tree.query(o['center']):
            i=int(i);l,g=regions[i]
            if l not in o['layers']:continue
            intersection=g.intersection(o['center'])
            if intersection.is_empty:continue
            q=intersection.representative_point();xy=(q.x,q.y)
            if i not in starts or math.dist((b['x'],b['y']),xy)<math.dist((b['x'],b['y']),starts[i]):starts[i]=xy
    goals={}
    for j in target_indices:
        o=own[j]
        for i in tree.query(o['center']):
            i=int(i);l,g=regions[i]
            if l not in o['layers']:continue
            intersection=g.intersection(o['center'])
            if intersection.is_empty:continue
            q=intersection.representative_point();xy=(q.x,q.y)
            if i not in goals or math.dist((a['x'],a['y']),xy)<math.dist((a['x'],a['y']),goals[i]):goals[i]=xy
    queue=deque(starts);previous={i:None for i in starts};found=None
    while queue:
        i=queue.popleft()
        if i in goals:found=i;break
        for j,xy in adj[i]:
            if j not in previous:previous[j]=(i,xy);queue.append(j)
    assert found is not None,('No measured layer path',job,starts,goals)
    landing=goals[found]
    transitions=[];end=found
    while previous[found] is not None:
        old,xy=previous[found];transitions.append((old,found,xy));found=old
    transitions.reverse();position=starts[found];drawn=[]
    for old,new,xy in transitions+[(end,None,landing)]:
        path=polygon_path(regions[old][1],(position,xy));layer=regions[old][0]
        if len(path)>1:
            w=dict(net=n,layer=layer,width=width,points=path);physical['wires'].append(w);authored['wires'].append(w);drawn.append(w)
        if new is not None:
            v=dict(net=n,x=xy[0],y=xy[1]);physical['vias'].append(v);authored['vias'].append(v)
        position=xy
    completed.append(dict(source=job['source'],net=n,segments=sum(len(w['points'])-1 for w in drawn),lengthMm=sum(LineString(w['points']).length for w in drawn),vias=len(transitions)))
    print(json.dumps(completed[-1]),flush=True)
    if os.environ.get('SWAP_PROGRESS'):
        Path('routing/swap-progress-physical.json').write_text(json.dumps(physical,indent=2)+'\n')
        Path('reports/swap-progress.json').write_text(json.dumps(dict(complete=False,notInstalled=True,routes=completed),indent=2)+'\n')
Path('routing/review-finished-copper.json').write_text(json.dumps(authored,indent=2)+'\n')
Path('routing/bulk-top-closure-physical-candidate.json').write_text(json.dumps(physical,indent=2)+'\n')
num=lambda x:str(round(float(x)*10000,4))
rows=['(session local_rework (routes (resolution um 10) (network_out']
for n in sorted({w['net'] for w in physical['wires']}|{v['net'] for v in physical['vias']}):
    rows.append(f'(net {n}')
    for w in physical['wires']:
        if w['net']==n:rows.append(f"(wire (path {w['layer']} {num(w['width'])} "+' '.join(num(v) for p in w['points'] for v in p)+'))')
    for v in physical['vias']:
        if v['net']==n:rows.append(f"(via VIA {num(v['x'])} {num(v['y'])})")
    rows.append(')')
rows.append(')))')
Path('routing/bulk-top-closure-candidate.ses').write_text('\n'.join(rows)+'\n')
Path('reports/bulk-top-closure-closure.json').write_text(json.dumps(dict(complete=True,notInstalled=True,routes=completed,method='explicit clearance-polygon fanouts; independent import required'),indent=2)+'\n')
print(json.dumps(dict(complete=True,wires=len(physical['wires']),vias=len(physical['vias']))))