krishnax12/t113-minimal-linux

scripts/reuse-reference-routing.py

"""Reuse reference copper only where it reaches retained pads without new DRC conflicts.

Prune disconnected copper and unterminated branches. The result still requires
fresh routing and a full independent connectivity/clearance audit.
"""
import json, sys, collections
from pathlib import Path
from shapely.strtree import STRtree
from shapely.geometry import Point, Polygon
from shapely.ops import unary_union
from routing_geometry import Nets, copper, audit, fill_planes

reference_path, source_path, destination = map(Path,sys.argv[1:4])
reference = json.loads(reference_path.read_text())
source = json.loads(source_path.read_text())
old_nets, new_nets = Nets(reference), Nets(source)
root_subcircuit = new_nets.of('pcb_board')[0]['subcircuit_id']
trace_ids = collections.defaultdict(list)
for trace in new_nets.of('source_trace'):
    terminals=trace.get('connected_source_port_ids',[])+trace.get('connected_source_net_ids',[])
    if terminals:trace_ids[new_nets.source(terminals[0])].append(trace['source_trace_id'])
regions=json.loads(Path('routing/plane-regions.json').read_text())
board=Polygon([(p['x'],p['y']) for p in new_nets.of('pcb_board')[0]['outline']])
plane_regions=[(r['net'],r['layer'],Polygon([(p['x'],p['y']) for p in r['outline']]) if 'outline' in r else board) for r in regions]
candidates=[]
for record in reference:
    if record['type'] not in ('pcb_trace','pcb_via'):continue
    net=old_nets.net(record)
    if net not in new_nets.ids:continue
    cached=json.loads(json.dumps(record))
    cached[cached['type']+'_id']='reference_'+cached[cached['type']+'_id']
    cached['source_net_id']=new_nets.ids[net]
    cached['subcircuit_id']=root_subcircuit
    cached.pop('source_trace_id',None)
    cached.pop('pcb_port_ids',None)
    cached.pop('pcb_group_id',None)
    cached.pop('subcircuit_connectivity_map_key',None)
    if trace_ids[net]:cached['source_trace_id']=trace_ids[net][0]
    if cached['type']=='pcb_trace':
        for point in cached['route']:
            point.pop('start_pcb_port_id',None)
            point.pop('end_pcb_port_id',None)
    candidates.append(cached)

fixed=[r for r in source if r['type'] not in ('pcb_trace','pcb_via','pcb_copper_pour')]
# Preserve newly reviewed fixed prewires; they are checked before this step.
protected=[r for r in source if r['type'] in ('pcb_trace','pcb_via')]
protected_ids={r[r['type']+'_id'] for r in protected}
# The reference and rebuilt prewires can differ only by coordinate rounding.
# Keep one copy so the router does not normalize overlapping parallel tracks.
protected_geometry,_=copper(fixed+protected)
protected_tracks=collections.defaultdict(list)
for item in protected_geometry:
    if item['kind']=='trace':protected_tracks[(item['net'],item['layer'])].append(item['geometry'])
protected_coverage={key:unary_union(shapes).buffer(.001) for key,shapes in protected_tracks.items()}
reference_geometry,_=copper(fixed+candidates)
covered_segments=collections.defaultdict(list)
for item in reference_geometry:
    if item['kind']!='trace':continue
    coverage=protected_coverage.get((item['net'],item['layer']))
    covered_segments[item['id'].rsplit('_',1)[0]].append(coverage is not None and coverage.covers(item['geometry']))
candidates=[record for record in candidates if not (record['type']=='pcb_trace' and all(covered_segments[record['pcb_trace_id']]))]
def record_id(identifier):
    return identifier.rsplit('_', 1)[0] if identifier.startswith('reference_pcb_trace_') else identifier

removed=[]
for cycle in range(10):
    combined=fixed+protected+candidates
    report=audit(combined)
    cached_ids={r[r['type']+'_id'] for r in candidates}
    bad=set()
    for violation in report['violations']:
        if violation['type']=='copper_clearance':
            for field in ['a','b']:
                if record_id(violation[field]) in cached_ids:bad.add(record_id(violation[field]))
        elif violation['type'] in ('nonplated_hole_clearance','board_edge_clearance') and violation.get('id') in cached_ids:bad.add(violation['id'])
        elif violation['type']=='drill_clearance':
            for field in ['a','b']:
                if record_id(violation[field]) in cached_ids:bad.add(record_id(violation[field]))
    if not bad:break
    removed.extend(sorted(bad))
    candidates=[r for r in candidates if r[r['type']+'_id'] not in bad]
else:raise RuntimeError('Reference clearance pruning did not converge')

items,nets=copper(fixed+protected+candidates)
for item in items:
    if item['kind']=='trace':item['id']=item['id'].rsplit('_',1)[0]
adjacency=collections.defaultdict(set)
roles={}
for item in items:
    roles[item['id']]=item['kind']
    adjacency[item['id']]
for layer in ['top','inner1','inner2','bottom']:
    layer_items=[item for item in items if item['layer']==layer]
    geometries=[item['geometry'] for item in layer_items]
    tree=STRtree(geometries)
    for i,a in enumerate(layer_items):
        for index in tree.query(geometries[i].buffer(.0006)):
            b=layer_items[int(index)]
            if a['id']==b['id'] or a['net']!=b['net']:continue
            if a['geometry'].distance(b['geometry'])<.0006:
                adjacency[a['id']].add(b['id']);adjacency[b['id']].add(a['id'])
pads={key for key,role in roles.items() if role=='pad'}
reachable=set(pads)
queue=list(pads)
while queue:
    key=queue.pop()
    for neighbor in adjacency[key]:
        if neighbor not in reachable:reachable.add(neighbor);queue.append(neighbor)
candidates=[r for r in candidates if r[r['type']+'_id'] in reachable]
keep_ids={r[r['type']+'_id'] for r in candidates}|protected_ids|pads
terminals=pads|protected_ids
for via in candidates:
    if via['type']=='pcb_via' and any(nets.net(via)==net and layer in via['layers'] and region.covers(Point(via['x'],via['y'])) for net,layer,region in plane_regions):terminals.add(via['pcb_via_id'])
while True:
    leaves={key for key in keep_ids-terminals if len(adjacency[key]&keep_ids)<=1}
    if not leaves:break
    keep_ids-=leaves
    removed.extend(sorted(leaves))
candidates=[r for r in candidates if r[r['type']+'_id'] in keep_ids]
# Identical barrels must not produce duplicate drills when fixed copper overlaps.
seen={(round(r['x'],6),round(r['y'],6),nets.net(r)) for r in protected if r['type']=='pcb_via'}
unique=[]
for record in candidates:
    if record['type']=='pcb_via':
        key=(round(record['x'],6),round(record['y'],6),nets.net(record))
        if key in seen:continue
        seen.add(key)
    unique.append(record)
result=fixed+protected+unique
destination.write_text(json.dumps(result,indent=2)+'\n')
report=audit(fill_planes(result))
destination.with_suffix('.reuse-check.json').write_text(json.dumps(report,indent=2)+'\n')
print(json.dumps({'retainedReferenceRecords':len(unique),'discardedRecords':len(set(removed)),'violations':report['violationCounts'],'remainingUnconnectedNets':[entry['net'] for entry in report['unconnectedNets']]},indent=2))
if report['violations']:sys.exit(1)