shibosoftwaredev/f1c100s-linux-nema17-stepper-controller

scripts/plan-current-corridor.py

"""Find actual clear layer-transition sites for the manually fanned DAC net.
This is planning evidence only; it never changes runtime copper or DRC rules.
"""
from route_geometry import *
from shapely.ops import unary_union
from shapely import contains_xy,prepare
from scipy.ndimage import label
import numpy as np
from collections import defaultdict,deque

c=load_circuit('routing/review-placement.circuit.json');root=connectivity(c)
m=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'}
physical=json.loads(Path(os.environ.get('CURRENT_CORRIDOR_PHYSICAL','routing/software-current-partial-physical-candidate.json')).read_text())
net=os.environ.get('CURRENT_CORRIDOR_NET','N215')
width=float(os.environ.get('CURRENT_CORRIDOR_WIDTH','.15' if net in ['N161','N162'] else '.12'))
objects=[];pads=[];holes=[]
for e in c:
    if e['type'] in ['pcb_smtpad','pcb_plated_hole']:
        p=ports.get(e.get('pcb_port_id'));n=m.get(root(p['source_port_id'])) if p else None
        item=(n,e.get('layers',[e.get('layer')]),geometry(e,'outer_' if e['type']=='pcb_plated_hole' else ''))
        objects.append(item);pads.append(item)
    elif e['type']=='pcb_hole':holes.append(geometry(e,'hole_'))
for w in physical['wires']:objects.append((w['net'],[w['layer']],LineString(w['points']).buffer(w['width']/2)))
for v in physical['vias']:objects.append((v['net'],LAYERS,Point(v['x'],v['y']).buffer(.225)))
axis=np.round(np.arange(-20.9,20.9001,.05),8);xx,yy=np.meshgrid(axis,axis)
def raster(gs):
    g=unary_union(gs);prepare(g);return contains_xy(g,xx,yy)
vb=[g.buffer(.354) for n,ls,g in objects if n!=net]
vb += [g.buffer(.331) for n,ls,g in pads]
vb += [Point(v['x'],v['y']).buffer(.554) for v in physical['vias']]
vb += [h.buffer(.429) for h in holes]
vb += [Point(x,y).buffer(3.1) for x in [-15.5,15.5] for y in [-15.5,15.5]]
via_ok=~raster(vb)&(abs(xx)<20.72)&(abs(yy)<20.72)
if os.environ.get('CURRENT_CORRIDOR_MAX_VIA_Y'):
    via_ok &= yy <= float(os.environ['CURRENT_CORRIDOR_MAX_VIA_Y'])
components=[]
for layer in LAYERS:
    blockers=[g.buffer(.1003+width/2) for n,ls,g in objects if n!=net and layer in ls]
    # Use the routing utility's stronger via-to-wire exclusion too.
    blockers += [Point(v['x'],v['y']).buffer(.354+width/2) for v in physical['vias'] if v['net']!=net]
    blockers += [h.buffer(.2603) for h in holes]
    if layer in ['top','bottom']:
        radius=(2.8503 if layer=='top' else 2.6003)+.06
        blockers += [Point(x,y).buffer(radius) for x in [-15.5,15.5] for y in [-15.5,15.5]]
    free=~raster(blockers)&(abs(xx)<20.8897)&(abs(yy)<20.8897)
    cc,count=label(free)
    components.append(cc);print(layer,count,flush=True)
def index(xy):return tuple(int(round((a+20.9)/.05)) for a in xy[::-1])
start=tuple(map(float,os.environ.get('CURRENT_CORRIDOR_START','-4.15,-15.1').split(',')))
target=tuple(map(float,os.environ.get('CURRENT_CORRIDOR_TARGET','10.15,-9').split(',')))
starts={(l,int(cc[index(start)])) for l,cc in enumerate(components) if cc[index(start)]}
goals={(l,int(cc[index(target)])) for l,cc in enumerate(components) if cc[index(target)]}
if os.environ.get('CURRENT_CORRIDOR_SCAN'):
    from shapely.strtree import STRtree
    fixed=[g.buffer(.331) for n,ls,g in pads]
    fixed += [Point(v['x'],v['y']).buffer(.554) for v in physical['vias']]
    fixed += [h.buffer(.429) for h in holes]
    fixed += [Point(x,y).buffer(3.1) for x in [-15.5,15.5] for y in [-15.5,15.5]]
    scan_layer=LAYERS.index(os.environ.get('CURRENT_CORRIDOR_SCAN_LAYER','top'))
    allowed=~raster(fixed)&(components[scan_layer]==components[scan_layer][index(start)])
    allowed &= (abs(xx-start[0])<4)&(abs(yy-start[1])<4)
    fw=[w for w in physical['wires'] if w['net']!=net]
    fg=[LineString(w['points']).buffer(w['width']/2) for w in fw];tree=STRtree(fg)
    sites=[]
    for j,i in zip(*np.where(allowed)):
        xy=(float(axis[i]),float(axis[j]));point=Point(xy)
        bad=[fw[int(k)] for k in tree.query(point.buffer(.354)) if fg[int(k)].distance(point)<.354]
        if any(w['width']>=.275 for w in bad):continue
        score=len({w['net'] for w in bad})*100+math.dist(xy,start)
        sites.append(dict(via=xy,score=score,blockingWires=bad))
    sites.sort(key=lambda a:a['score'])
    Path('reports/software-current-transition-rework.json').write_text(json.dumps(sites[:30],indent=2)+'\n')
if os.environ.get('CURRENT_CORRIDOR_START_LAYERS'):
    starts={node for node in starts if LAYERS[node[0]] in os.environ['CURRENT_CORRIDOR_START_LAYERS'].split(',')}
if os.environ.get('CURRENT_CORRIDOR_GOAL_LAYERS'):
    goals={node for node in goals if LAYERS[node[0]] in os.environ['CURRENT_CORRIDOR_GOAL_LAYERS'].split(',')}
edges=defaultdict(dict)
for v in physical['vias']:
    if v['net']!=net:continue
    j,i=index((v['x'],v['y']))
    nodes=[(l,int(cc[j,i])) for l,cc in enumerate(components) if cc[j,i]]
    for a in nodes:
        for b in nodes:
            if a!=b:edges[a][b]=(0,(v['x'],v['y']))
for j,i in zip(*np.where(via_ok)):
    nodes=[(l,int(cc[j,i])) for l,cc in enumerate(components) if cc[j,i]]
    xy=(float(axis[i]),float(axis[j]))
    score=2*math.dist(xy,start)+math.dist(xy,target)
    for a in nodes:
        for b in nodes:
            if a==b:continue
            old=edges[a].get(b)
            if old is None or score<old[0]:edges[a][b]=(score,xy)
queue=deque(starts);prev={n:None for n in starts};found=None
while queue:
    a=queue.popleft()
    if a in goals:found=a;break
    for b,(_,xy) in sorted(edges[a].items(),key=lambda item:item[1][0]):
        if b not in prev:prev[b]=(a,xy);queue.append(b)
path=[];end=found
while end and prev[end]:
    a,xy=prev[end];path.append(dict(fromLayer=LAYERS[a[0]],toLayer=LAYERS[end[0]],via=xy));end=a
path.reverse()
report=dict(start=start,target=target,startComponents=sorted(starts),goalComponents=sorted(goals),
            reachableComponents=len(prev),found=found is not None,transitions=path)
if os.environ.get('CURRENT_CORRIDOR_BRIDGE'):
    from shapely.strtree import STRtree
    reverse=defaultdict(set)
    for a,neighbors in edges.items():
        for b in neighbors:reverse[b].add(a)
    goal_reached=set(goals);todo=list(goals)
    while todo:
        for a in reverse[todo.pop()]:
            if a not in goal_reached:goal_reached.add(a);todo.append(a)
    fixed=[g.buffer(.331) for n,ls,g in pads]
    fixed += [Point(v['x'],v['y']).buffer(.554) for v in physical['vias']]
    fixed += [h.buffer(.429) for h in holes]
    fixed += [Point(x,y).buffer(3.1) for x in [-15.5,15.5] for y in [-15.5,15.5]]
    allowed=~raster(fixed)&(abs(xx)<20.72)&(abs(yy)<20.72)
    a_mask=np.zeros_like(allowed);b_mask=np.zeros_like(allowed)
    for l,cc in enumerate(components):
        a_mask |= np.isin(cc,[node[1] for node in prev if node[0]==l])
        b_mask |= np.isin(cc,[node[1] for node in goal_reached if node[0]==l])
    fw=[w for w in physical['wires'] if w['net']!=net]
    fg=[LineString(w['points']).buffer(w['width']/2) for w in fw];tree=STRtree(fg)
    sites=[]
    for j,i in zip(*np.where(allowed&a_mask&b_mask)):
        xy=(float(axis[i]),float(axis[j]));point=Point(xy)
        bad=[fw[int(k)] for k in tree.query(point.buffer(.354)) if fg[int(k)].distance(point)<.354]
        if any(w['width']>=.275 for w in bad):continue
        score=len({w['net'] for w in bad})*100+math.dist(xy,start)+math.dist(xy,target)
        sites.append(dict(via=xy,score=score,blockingWires=bad,
            fromLayers=[LAYERS[l] for l,cc in enumerate(components) if (l,int(cc[j,i])) in prev],
            toLayers=[LAYERS[l] for l,cc in enumerate(components) if (l,int(cc[j,i])) in goal_reached]))
    sites.sort(key=lambda a:a['score'])
    Path('reports/software-current-bridge-rework.json').write_text(json.dumps(sites[:30],indent=2)+'\n')
report['referenceCandidates']=[dict(site=s,reachableInputLayers=[LAYERS[l] for l,cc in enumerate(components)
    if (l,int(cc[index(s['inputVia'])])) in prev])
    for s in json.loads(Path('reports/reference-via-first-sites.json').read_text())]
Path('reports/software-current-corridor-'+net+'.json').write_text(json.dumps(report,indent=2)+'\n')
print(json.dumps(report),flush=True)