imrishabh18/rp2040-motor-controller

scripts/rebase-top-assembly.py

"""Seed a routing candidate by preserving unchanged copper and remapping real pins.

Moved terminals are explicitly queued for rerouting. This is not a validated plan.
Run native DRC, copper connectivity and manufacturing audits after route repair.
"""
import copy
import json
import math
from pathlib import Path

old = json.loads(Path('dist/baseline/circuit.json').read_text())
new = json.loads(Path('dist/pd-control/unrouted.json').read_text())
plan = json.loads(Path('dist/baseline/route-plan.json').read_text())
srj = json.loads(Path('dist/pd-control/routing-input.json').read_text())

def ports(circuit):
    components = {e['source_component_id']: e['name'] for e in circuit if e['type'] == 'source_component'}
    source = {e['source_port_id']: e for e in circuit if e['type'] == 'source_port'}
    result = {}
    for e in circuit:
        if e['type'] != 'pcb_port' or e.get('source_port_id') not in source:
            continue
        p = source[e['source_port_id']]
        result[e['pcb_port_id']] = (components.get(p['source_component_id'], p['source_component_id']), p.get('pin_number'), p['name'])
    return result

before, after = ports(old), ports(new)
reverse_after = {value: key for key, value in after.items()}
mapping = {key: reverse_after[value] for key, value in before.items() if value in reverse_after}
old_points = {p['pcb_port_id']: p for c in plan['expectedInput']['connections'] for p in c['pointsToConnect'] if p.get('pcb_port_id')}
new_points = {p['pcb_port_id']: p for c in srj['connections'] for p in c['pointsToConnect'] if p.get('pcb_port_id')}
for circuit, points in [(old, old_points), (new, new_points)]:
    for e in circuit:
        if e['type'] == 'pcb_port':
            points.setdefault(e['pcb_port_id'], {'x': e['x'], 'y': e['y'], 'layer': e.get('layers', ['top'])[0]})

def reverse_route(route):
    result = copy.deepcopy(route[::-1])
    # Segment width belongs to its first vertex. Preserve it when reversing.
    for i in range(len(result)-1):
        old_index = len(route)-1-i
        a, b = route[old_index-1], route[old_index]
        if result[i]['route_type'] == 'wire' and a['route_type'] == 'wire' and a.get('layer') == b.get('layer'):
            result[i]['width'] = a['width']
    return result

traces = plan['traces']
for removed in [key for key, value in before.items() if value[0].startswith('TP_')]:
    touching = [t for t in traces if removed in t.get('connectsTo', [])]
    if len(touching) == 1:
        traces.remove(touching[0])
    elif len(touching) == 2:
        a, b = touching
        point = old_points[removed]
        def at_removed(p):
            return math.hypot(p['x']-point['x'], p['y']-point['y']) < 1e-5
        ra = a['route'] if at_removed(a['route'][-1]) else reverse_route(a['route'])
        rb = b['route'] if at_removed(b['route'][0]) else reverse_route(b['route'])
        assert at_removed(ra[-1]) and at_removed(rb[0])
        a['route'] = ra + rb[1:]
        a['connectsTo'] = [p for t in touching for p in t['connectsTo'] if p != removed]
        traces.remove(b)
    else:
        raise ValueError(f'Unexpected test-point topology: {before[removed]} {len(touching)}')

connections = [(c, {p['pcb_port_id'] for p in c['pointsToConnect'] if p.get('pcb_port_id')}) for c in srj['connections']]
parent = {}
def root(value):
    parent.setdefault(value, value)
    if parent[value] != value:
        parent[value] = root(parent[value])
    return parent[value]
for e in new:
    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 value in ids[1:]:
        parent[root(value)] = root(ids[0])
pcb_net = {e['pcb_port_id']: root(e['source_port_id']) for e in new if e['type'] == 'pcb_port' and e.get('source_port_id')}
old_trace_names = {e['source_trace_id']: e.get('name') for e in old if e['type'] == 'source_trace'}
new_trace_ids = {e.get('name'): e['source_trace_id'] for e in new if e['type'] == 'source_trace'}
changed, kept = [], []
for i, t in enumerate(traces):
    old_ids = t.get('connectsTo', [])
    assert all(p in mapping for p in old_ids), (t['pcb_trace_id'], old_ids)
    new_ids = [mapping[p] for p in old_ids]
    candidates = [c for c, ids in connections if new_ids and set(new_ids).issubset(ids)]
    if not new_ids:
        assert t.get('source_net_id'), 'A copper bridge must identify its net'
        connection = {'name':t['source_net_id']}
    elif candidates:
        connection = min(candidates, key=lambda c: len(c['pointsToConnect']))
    else:
        nets = {pcb_net[p] for p in new_ids}
        assert len(nets) == 1, (t['pcb_trace_id'], [after[p] for p in new_ids])
        group = [c for c, ids in connections if any(pcb_net[p] in nets for p in ids)]
        assert group
        connection = {'name': '__'.join(dict.fromkeys(c['name'] for c in group))}
    t['connection_name'] = connection['name']
    old_name = old_trace_names.get(t.get('source_trace_id'))
    source_trace_id = new_trace_ids.get(old_name) if old_name else None
    t.pop('source_trace_id', None)
    t.pop('source_net_id', None)
    if connection.get('source_trace_id'):
        t['source_trace_id'] = connection['source_trace_id']
    elif source_trace_id:
        t['source_trace_id'] = source_trace_id
    elif connection['name'].startswith('source_trace_'):
        t['source_trace_id'] = connection['name'].split('__')[0]
    if connection['name'].startswith('source_net_'):
        t['source_net_id'] = connection['name']
    t['pcb_trace_id'] = f'top_assembly_{i:03}'
    t['connectsTo'] = new_ids
    moved = False
    for old_id, new_id in zip(old_ids, new_ids):
        oldp, newp = old_points[old_id], new_points[new_id]
        for index in [0, len(t['route'])-1]:
            endpoint = t['route'][index]
            if math.hypot(endpoint['x']-oldp['x'], endpoint['y']-oldp['y']) > 1e-5:
                continue
            if math.hypot(newp['x']-oldp['x'], newp['y']-oldp['y']) > 1e-6 or newp.get('layer') != oldp.get('layer'):
                endpoint.update(x=newp['x'], y=newp['y'])
                if newp.get('layer'):
                    endpoint['layer'] = newp['layer']
                moved = True
    # Endpoint tags are regenerated from connectsTo by core.
    for p in t['route']:
        p.pop('start_pcb_port_id', None)
        p.pop('end_pcb_port_id', None)
    if moved:
        changed.append({'id': t['pcb_trace_id'], 'pins': [after[p] for p in new_ids]})
    kept.append(t)

folder = Path('dist/top-assembly')
folder.mkdir(parents=True, exist_ok=True)
(folder/'route-plan.json').write_text(json.dumps({'expectedInput': srj, 'traces': kept}, indent=2))
(folder/'changed-routes.json').write_text(json.dumps(changed, indent=2))
print(json.dumps({'retainedRoutes': len(kept), 'movedTerminalRoutes': changed}, indent=2))