shibosoftwaredev/f1c100s-linux-nema17-stepper-controller

scripts/prepare-bulk-top-closure.py

"""Connect actual copper islands reported by the independent source-path audit.

Every virtual start is an existing through-via in the start pad's physical
component, never an assumed net connection. The resulting copper is audited
and reimported against every original source connection before installation.
"""
from route_geometry import *
from shapely.strtree import STRtree
from collections import defaultdict
c=load_circuit('routing/review-placement.circuit.json');root=connectivity(c)
m=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'}
ports={e['pcb_port_id']:e for e in c if e['type']=='pcb_port'}
def sn(s):return m[root((s.get('connected_source_port_ids',[])+s.get('connected_source_net_ids',[]))[0])]
net_sources=defaultdict(list)
for s in sources.values():net_sources[sn(s)].append(s)
snapshot=os.environ.get('CLOSURE_SNAPSHOT')
physical=json.loads(Path(snapshot or 'routing/bulk-top-physical-candidate.json').read_text())
patches=json.loads(Path(os.environ.get('CLOSURE_PATCHES','routing/bulk-top-island-patches.json')).read_text())
previous_removed=(json.loads(Path('reports/bulk-top-physical-islands.json').read_text())['removedForManualEscapes'] if snapshot else [])
for replacement in patches.get('replaceWires',[]):
    matches=[w for w in physical['wires'] if w['net']==replacement['net'] and w['layer']==replacement['layer']
             and len(w['points'])==2 and all(any(math.dist(a,b)<1e-7 for b in replacement['points']) for a in w['points'])]
    assert len(matches)==1,('Resolve exact old segment before reshaping it',replacement,matches)
    physical['wires']=[w for w in physical['wires'] if w is not matches[0]]
for v in patches.get('removeVias',[]):
    matches=[p for p in physical['vias'] if p['net']==v['net'] and math.dist((v['x'],v['y']),(p['x'],p['y']))<1e-7]
    assert len(matches)==1,('Resolve exact old via before relocating it',v,matches)
    physical['vias']=[p for p in physical['vias'] if p is not matches[0]]
patch_shapes=[(w['net'],[w['layer']],LineString(w['points']).buffer(w['width']/2)) for w in patches['wires']]
patch_shapes += [(v['net'],LAYERS,Point(v['x'],v['y']).buffer(.225)) for v in patches['vias']]
# Final placement also moves one small link onto the clear top-side strip.
# Rip only colliding low-current wire segments, never an existing power trunk.
names={e['source_component_id']:e['name'] for e in c if e['type']=='source_component'}
placed={e['pcb_component_id']:names[e['source_component_id']] for e in c if e['type']=='pcb_component'}
for e in c:
    if e['type']=='pcb_smtpad' and placed.get(e['pcb_component_id']) in json.loads(Path('review-placement.json').read_text()):
        p=ports[e['pcb_port_id']]
        patch_shapes.append((m[root(p['source_port_id'])],[e['layer']],geometry(e)))
removed=[];kept=[]
for w in physical['wires']:
    g=LineString(w['points']).buffer(w['width']/2)
    clash=any(n!=w['net'] and w['layer'] in ls and g.distance(h)<.1001 for n,ls,h in patch_shapes)
    clash|=any(v['net']!=w['net'] and g.distance(Point(v['x'],v['y']))<.354 for v in patches['vias'])
    if clash:
        assert w['width']<.299,('Manual signal escape must not cut an existing power trunk',w)
        removed.append(w)
    else:kept.append(w)
physical['wires']=kept+patches['wires']
for v in patches['vias']:
    assert all(math.dist((v['x'],v['y']),(p['x'],p['y']))>=.554 for p in physical['vias']),('Stitch spacing',v)
physical['vias']+=patches['vias']
Path('routing/bulk-top-closure-seed-physical.json').write_text(json.dumps(physical,indent=2)+'\n')
objects=defaultdict(list);retained=[]
for i,w in enumerate(physical['wires']):
    sid=net_sources[w['net']][0]['source_trace_id']
    retained.append(dict(type='pcb_trace',pcb_trace_id=f'closure_retained_wire_{i}',source_trace_id=sid,
        route=[dict(route_type='wire',x=x,y=y,layer=w['layer'],width=w['width']) for x,y in w['points']]))
    objects[w['net']].append(dict(shape=LineString(w['points']).buffer(w['width']/2,quad_segs=16),layers=[w['layer']]))
for i,v in enumerate(physical['vias']):
    sid=net_sources[v['net']][0]['source_trace_id']
    retained.append(dict(type='pcb_trace',pcb_trace_id=f'closure_retained_via_{i}',source_trace_id=sid,
        route=[dict(route_type='via',x=v['x'],y=v['y'],from_layer='top',to_layer='bottom',via_diameter=.45,via_hole_diameter=.3)]))
    objects[v['net']].append(dict(shape=Point(v['x'],v['y']).buffer(.225,quad_segs=16),layers=LAYERS,via=v))
for e in c:
    if e['type'] in ['pcb_smtpad','pcb_plated_hole'] and e.get('pcb_port_id') in ports:
        p=ports[e['pcb_port_id']];n=m[root(p['source_port_id'])]
        objects[n].append(dict(shape=geometry(e,'outer_' if e['type']=='pcb_plated_hole' else ''),
            layers=e.get('layers',[e.get('layer')]),port=p))
jobs=[];islands=[]
for n in sorted(net_sources):
    own=objects[n];tree=STRtree([o['shape'] for o in own]);remaining=set(range(len(own)));components=[]
    while remaining:
        todo=[next(iter(remaining))];seen=set()
        while todo:
            i=todo.pop()
            if i in seen:continue
            seen.add(i);remaining.discard(i);a=own[i]
            for j in tree.query(a['shape'].buffer(1e-7)):
                j=int(j);b=own[j]
                if j not in seen and set(a['layers'])&set(b['layers']) and a['shape'].distance(b['shape'])<1e-7:todo.append(j)
        pp={own[i]['port']['pcb_port_id']:own[i]['port'] for i in seen if 'port' in own[i]}
        if pp:components.append(dict(indices=seen,ports=list(pp.values())))
    components.sort(key=lambda comp:(-len(comp['ports']),-len(comp['indices'])))
    if len(components)<=1:continue
    main=components[0];anchor=main['ports'][0]
    for k,comp in enumerate(components[1:]):
        p=min(comp['ports'],key=lambda p:math.dist((p['x'],p['y']),(anchor['x'],anchor['y'])))
        choices=[s for s in net_sources[n] if p['source_port_id'] in s.get('connected_source_port_ids',[])]
        assert choices,('No source for island terminal',n,p)
        s=choices[0];vv=[own[i]['via'] for i in comp['indices'] if 'via' in own[i]]
        if vv:
            v=min(vv,key=lambda v:math.dist((v['x'],v['y']),(anchor['x'],anchor['y'])))
            start=dict(x=v['x'],y=v['y'],layers=LAYERS,pcb_port_id=f'actual_island_{n}_{k}_via')
        else:start={key:p[key] for key in ['x','y','layers','pcb_port_id']}
        width=max(s.get('min_trace_thickness',.12),{'N148':.3,'N149':.25,'N118':.25,'N125':.25,'N150':.3}.get(n,.15))
        target={**{key:anchor[key] for key in ['x','y','layers','pcb_port_id']},'connected_tree':n!='N148'}
        if n=='N148' and not os.environ.get('FOCUSED_REVIEW'):
            # The authored 0.30 mm fanout connects this through-via to
            # the real supply anchor. Do not land on a thinner old twig.
            assert any('via' in own[i] and math.dist((own[i]['via']['x'],own[i]['via']['y']),(11.32,-13.25))<1e-7 for i in main['indices'])
            target.update(x=11.32,y=-13.25,layers=LAYERS,pcb_port_id='retained_n148_anchor_stitch',actual_pad_port=anchor['pcb_port_id'])
        if n=='N160' and not os.environ.get('FOCUSED_REVIEW'):
            # Route toward the larger existing status-side copper tree.
            # This start is a real stitch in the MCU/debug component.
            vvmain=[own[i]['via'] for i in main['indices'] if 'via' in own[i]]
            v=next(v for v in vvmain if math.dist((v['x'],v['y']),(13.45,-3.2))<1e-7)
            start=dict(x=v['x'],y=v['y'],layers=LAYERS,pcb_port_id='retained_mcu_status_stitch')
            landing=comp['ports'][0]
            target={**{key:landing[key] for key in ['x','y','layers','pcb_port_id']},'connected_tree':True}
        if n=='N162' and not os.environ.get('FOCUSED_REVIEW'):
            # Both endpoint fanouts are explicit, pad-safe conductors.
            # Start and land on their real through-stitches.
            for xy in [(19.75,-18.2),(12.664442,-.536658)]:
                assert any('via' in o and math.dist((o['via']['x'],o['via']['y']),xy)<1e-7 for o in own)
            start=dict(x=19.75,y=-18.2,layers=LAYERS,pcb_port_id='authored_temperature_clock_stitch')
            target=dict(x=12.664442,y=-.536658,layers=LAYERS,pcb_port_id='authored_mcu_clock_stitch',connected_tree=True)
        if n=='N152' and not os.environ.get('FOCUSED_REVIEW'):
            # Start at the existing Ethernet signal stitch rather than
            # searching outward from the small termination-pad pocket.
            old=(-7.196846,-12.067678)
            assert any('via' in own[i] and math.dist((own[i]['via']['x'],own[i]['via']['y']),old)<1e-7 for i in main['indices'])
            start=dict(x=old[0],y=old[1],layers=LAYERS,pcb_port_id='retained_ethernet_termination_stitch')
            target=dict(x=15.9,y=-5.95,layers=LAYERS,pcb_port_id='authored_ethernet_termination_stitch',connected_tree=True)
        jobs.append(dict(source=s['source_trace_id'],start=start,constraints=dict(min_trace_thickness=width),
            target=target))
    islands.append(dict(net=n,components=len(components),ports=[len(comp['ports']) for comp in components],
        details=[dict(ports=[dict(name=names[ next(e['source_component_id'] for e in c if e['type']=='source_port' and e['source_port_id']==p['source_port_id']) ],x=p['x'],y=p['y']) for p in comp['ports']],
          vias=[own[i]['via'] for i in comp['indices'] if 'via' in own[i]]) for comp in components]))
def write(p,v):Path(p).write_text(json.dumps(v,indent=2)+'\n')
clock_priority={'N57':-4,'N209':-3,'N166':-2,'N62':-1,'N162':0,'N161':1,'N157':2}
jobs.sort(key=lambda job:(-job['constraints']['min_trace_thickness'],clock_priority.get(sn(sources[job['source']]),3)))
write('routing/bulk-top-closure-retained.json',retained)
write('routing/bulk-top-closure-plan.json',dict(prefix='bulk-top-closure',placement='routing/review-placement.circuit.json',
    retained='routing/bulk-top-closure-retained.json',extent=[-20.9,-20.9,20.9,20.9],jobs=jobs))
write('reports/bulk-top-physical-islands.json',dict(islands=islands,jobs=jobs,removedForManualEscapes=previous_removed+removed,relocatedVias=patches.get('removeVias',[])))
print(json.dumps(dict(islands=islands,jobs=len(jobs))))