imrishabh18/nema-23-stepper-controller

scripts/audit-all-net-copper.py

"""Drill-aware physical connectivity audit of every PCB copper net on all layers.

Logical source connectivity is compared with an independent geometry graph.
Every copper polygon participates, including unconnected component pads.
Plated barrels join layers; drills remove copper before intersections are tested.
"""
import json, collections, hashlib, sys
from pathlib import Path
from shapely import make_valid
from shapely.geometry import Point, Polygon, LineString, box
from shapely.affinity import rotate, translate
from shapely.strtree import STRtree
from copper_geometry import trace_segments, subtract_drills
input_path = Path(sys.argv[1] if len(sys.argv)>1 else 'dist/index/circuit.json')
output_path = Path(next((a.split('=',1)[1] for a in sys.argv if a.startswith('--output=')), 'checks/all-net-copper-connectivity.json'))
j = json.loads(input_path.read_text())
layers = ['top', 'inner1', 'inner2', 'bottom']
parent = {}
def root(x):
 parent.setdefault(x,x)
 if parent[x] != x: parent[x] = root(parent[x])
 return parent[x]
def join(a,b): parent[root(a)] = root(b)
for e in j:
 ids = e.get('connected_source_port_ids',[])+e.get('connected_source_net_ids',[]) if e['type']=='source_trace' else e.get('source_port_ids',[]) if e['type']=='source_component_internal_connection' else []
 for x in ids[1:]: join(ids[0],x)
ports = {e['pcb_port_id']: root(e['source_port_id']) for e in j if e['type']=='pcb_port' and e.get('source_port_id')}
traces = {e['source_trace_id']: root((e.get('connected_source_port_ids',[])+e.get('connected_source_net_ids',[]))[0]) for e in j if e['type']=='source_trace' and e.get('connected_source_port_ids',[])+e.get('connected_source_net_ids',[])}
names = collections.defaultdict(set)
for e in j:
 if e['type']=='source_net': names[root(e['source_net_id'])].add(e['name'])
 elif e['type']=='source_trace' and e.get('name'): names[traces[e['source_trace_id']]].add(e['name'])
components = {e['source_component_id']:e['name'] for e in j if e['type']=='source_component'}
sourceports = {e['source_port_id']:e for e in j if e['type']=='source_port'}
port_names = {}
for e in j:
 if e['type']=='pcb_port' and e.get('source_port_id') in sourceports:
  p=sourceports[e['source_port_id']]
  port_names[e['pcb_port_id']] = components.get(p.get('source_component_id'), '?')+'.'+str(p.get('name',p.get('pin_number',p['source_port_id'])))
via_nets = collections.defaultdict(set)
for e in j:
 if e['type']=='pcb_trace' and e.get('source_trace_id') in traces:
  for r in e['route']:
   if r['route_type']=='via': via_nets[(round(r['x'],6),round(r['y'],6))].add(traces[e['source_trace_id']])
shapes=[]; meta=[]; barrels=[]; unresolved=[]
def add(g,l,id,net,port=None):
 if l not in layers or g.is_empty:return
 if g.geom_type in ('MultiPolygon','GeometryCollection'):
  for piece in g.geoms:add(piece,l,id,net,port)
 elif g.area>1e-12:
  shapes.append(g);meta.append({'layer':l,'id':id,'net':net,'port':port})
def net_for(e):
 candidates=set()
 if e.get('source_net_id'):candidates.add(root(e['source_net_id']))
 if e.get('source_trace_id') in traces:candidates.add(traces[e['source_trace_id']])
 if e.get('pcb_port_id') in ports:candidates.add(ports[e['pcb_port_id']])
 if e['type']=='pcb_via':candidates.update(via_nets[(round(e['x'],6),round(e['y'],6))])
 if len(candidates)!=1:
  unresolved.append({'id':e.get(e['type']+'_id'),'candidateNets':sorted(candidates)})
  return None
 return next(iter(candidates))
def pad(e,outer=False,hole=False):
 if hole and 'hole_diameter' in e:return Point(e['x'],e['y']).buffer(e['hole_diameter']/2,quad_segs=64)
 if outer and e.get('shape')=='circular_hole_with_rect_pad':
  g=box(-e['rect_pad_width']/2,-e['rect_pad_height']/2,e['rect_pad_width']/2,e['rect_pad_height']/2)
  return translate(rotate(g,e.get('rect_ccw_rotation',0),origin=(0,0)),e['x'],e['y'])
 if e.get("shape")=="polygon":return make_valid(Polygon([(p["x"],p["y"]) for p in e["points"]]))
 prefix='hole_' if hole else 'outer_' if outer else ''
 if e['shape']=='circle':g=Point(0,0).buffer(e.get(prefix+'diameter',2*e.get('radius',0))/2,quad_segs=64)
 else:
  w,h=e[prefix+'width'],e[prefix+'height']; r=min(w,h)/2 if 'pill' in e['shape'] else e.get('corner_radius',0)
  if r:
   if abs(w-2*r)<1e-8 or abs(h-2*r)<1e-8:g=LineString([(-w/2+r,-h/2+r),(w/2-r,h/2-r)]).buffer(r,quad_segs=64)
   else:g=box(-w/2+r,-h/2+r,w/2-r,h/2-r).buffer(r,quad_segs=64)
  else:g=box(-w/2,-h/2,w/2,h/2)
 return translate(rotate(g,e.get('ccw_rotation',0),origin=(0,0)),e['x'],e['y'])

for e in j:
 t=e['type'];id=e.get(t+'_id')
 if t=='pcb_copper_pour':
  b=e['brep_shape']; xy=lambda r:[(v['x'],v['y']) for v in r['vertices']]
  add(make_valid(Polygon(xy(b['outer_ring']),[xy(r) for r in b['inner_rings']])),e['layer'],id,net_for(e))
 elif t=='pcb_smtpad':add(pad(e),e['layer'],id,net_for(e),e.get('pcb_port_id'))
 elif t in ('pcb_plated_hole','pcb_via'):
  net=net_for(e);ids=[]
  if t=='pcb_via':g=Point(e['x'],e['y']).buffer(e['outer_diameter']/2,quad_segs=64)
  else:g=pad(e,outer=True)
  for l in e['layers']:
   start=len(shapes);add(g,l,id,net,e.get('pcb_port_id'));ids.extend(range(start,len(shapes)))
  barrels.append(ids)
 elif t=='pcb_trace':
  net=net_for(e)
  for a,b,layer,width in trace_segments(e['route']):add(LineString([a,b]).buffer(width/2,quad_segs=64),layer,id,net)
drills=[]
for e in j:
 if e['type']=='pcb_via':drills.append((Point(e['x'],e['y']).buffer(e['hole_diameter']/2,quad_segs=64),e['layers']))
 elif e['type']=='pcb_plated_hole':drills.append((pad(e,hole=True),e['layers']))
 elif e['type']=='pcb_hole':
  if e['hole_shape']!='circle':raise ValueError('Unsupported NPTH shape')
  drills.append((Point(e['x'],e['y']).buffer(e['hole_diameter']/2,quad_segs=64),layers))
shapes,meta,barrels=subtract_drills(shapes,meta,barrels,drills,lambda m:m['layer'])
expected_pad_ports={e['pcb_port_id'] for e in j if e['type'] in ('pcb_smtpad','pcb_plated_hole') and e.get('pcb_port_id')}
actual_pad_ports={m['port'] for m in meta if m['port']}
erased_pads=sorted(port_names.get(p,p) for p in expected_pad_ports-actual_pad_ports)
parent={}
for i in range(len(shapes)):root(i)
for ids in barrels:
 for i in ids[1:]:join(ids[0],i)
contacts=[]
for l in layers:
 indices=[i for i,m in enumerate(meta) if m['layer']==l];tree=STRtree([shapes[i] for i in indices])
 for i in indices:
  for k in tree.query(shapes[i].buffer(1e-7)):
   other=indices[k]
   if other<=i:continue
   if shapes[i].distance(shapes[other])<=1e-7:
    join(i,other)
    if meta[i]['net']!=meta[other]['net']:contacts.append({'layer':l,'elements':[meta[i]['id'],meta[other]['id']],'nets':[meta[i]['net'],meta[other]['net']]})
groups=collections.defaultdict(list)
for i in range(len(shapes)):groups[root(i)].append(i)
net_groups=collections.defaultdict(set)
shorts=[];floating=[];group_details=[]
for key,indices in groups.items():
 nets={meta[i]['net'] for i in indices if meta[i]['net'] is not None}
 for n in nets:net_groups[n].add(key)
 pads={meta[i]['port'] for i in indices if meta[i]['port']}
 detail={'component':key,'nets':sorted(nets),'netNames':sorted({name for n in nets for name in names[n]}),'ports':sorted(port_names.get(p,p) for p in pads),'elements':sorted({meta[i]['id'] for i in indices}),'layers':sorted({meta[i]['layer'] for i in indices})}
 group_details.append(detail)
 if len(nets)>1:shorts.append(detail)
 if not pads:floating.append(detail)
# Preserve bare-copper discontinuities in the report, then verify continuity
# through permanently linked duplicate switch terminals. Never join the two
# switched sides or infer connectivity through resistors/capacitors/ICs.
raw_net_groups={n:set(keys) for n,keys in net_groups.items()}
physical_components=len(groups)
port_components=collections.defaultdict(set)
source_to_pcb=collections.defaultdict(list)
for e in j:
 if e['type']=='pcb_port' and e.get('source_port_id'):source_to_pcb[e['source_port_id']].append(e['pcb_port_id'])
for key,indices in groups.items():
 for i in indices:
  if meta[i]['port']:port_components[meta[i]['port']].add(key)
internal_links=[]
for e in j:
 if e['type']!='source_component' or e.get('manufacturer_part_number')!='SKRPACE010':continue
 for pair in e.get('internally_connected_source_port_ids',[]):
  pin_numbers=[str(sourceports[p].get('pin_number')) for p in pair]
  if sorted(pin_numbers) not in [['1','2'],['3','4']]:raise ValueError('Unexpected SKRPACE010 internal pin pair')
  keys={key for p in pair for pcb in source_to_pcb[p] for key in port_components[pcb]}
  if len(keys)>1:
   ordered=sorted(keys)
   for key in ordered[1:]:join(ordered[0],key)
   internal_links.append({'component':e['name'],'pins':pin_numbers,'bareCopperComponents':ordered,'evidence':'https://tech.alpsalpine.com/e/products/detail/SKRPACE010/','reason':'Permanent internal terminal pair; switch-actuated contacts remain separate'})
net_groups={n:{root(key) for key in keys} for n,keys in raw_net_groups.items()}
opens=[]
for net,keys in net_groups.items():
 if len(keys)>1:opens.append({'net':net,'names':sorted(names[net]),'components':[g for g in group_details if root(g['component']) in keys]})
result={'status':'passed' if not (shorts or opens or floating or unresolved or erased_pads) else 'failed', 'circuitJsonSha256':hashlib.sha256(input_path.read_bytes()).hexdigest(),'auditedLayers':layers,'method':'Vector copper intersection after removal of all via, plated and non-plated drills; plated barrels join layers; compared against source netlist','toleranceMm':1e-7,'copperNetCount':len(net_groups),'physicalCopperComponentCount':physical_components,'assembledElectricalComponentCount':len({root(k) for k in groups}),'verifiedInternalConnections':internal_links,'bareCopperDiscontinuities':[{'net':n,'names':sorted(names[n]),'physicalComponentCount':len(keys)} for n,keys in raw_net_groups.items() if len(keys)>1],'copperPadPortCount':len({m['port'] for m in meta if m['port']}),'copperPourCount':sum(e['type']=='pcb_copper_pour' for e in j),'geometryPieceCount':len(shapes),'shorts':shorts,'opens':opens,'floatingCopper':floating,'unresolvedCopper':unresolved,'padsErasedByDrilling':erased_pads,'crossNetContacts':contacts,'netComponents':group_details}
output_path.parent.mkdir(parents=True,exist_ok=True)
output_path.write_text(json.dumps(result,indent=2)+'\n')
print(json.dumps({k:v for k,v in result.items() if k!='netComponents'},indent=2))
if result['status']!='passed':sys.exit('All-net copper audit failed')