krishnax12/t113-minimal-linux

scripts/route-remaining.py

"""Route remaining pad groups around the exact saved copper on a 0.1 mm grid.

Saved traces are immutable obstacles to foreign nets. New through vias must
clear copper on every layer and all existing drills. The independent final
copper audit remains the release gate.
"""
import collections
import heapq
import itertools
import json
import math
import sys
from pathlib import Path

import numpy as np
from scipy.ndimage import distance_transform_edt
from shapely import intersects_xy
from shapely.geometry import LineString, Point, Polygon
from shapely.ops import unary_union
from shapely.strtree import STRtree

from routing_geometry import LAYERS, Nets, audit, copper, fill_planes, geom

STEP = .1
CLEARANCE = .151
OUTER_LAYERS = ['top', 'bottom']

class RemainingRouter:
    def __init__(self, circuit):
        self.circuit = circuit
        self.board = next(record for record in circuit if record['type'] == 'pcb_board')
        self.boundary = Polygon([(point['x'], point['y']) for point in self.board['outline']])
        xmin, ymin, xmax, ymax = self.boundary.bounds
        self.x = np.arange(xmin + STEP / 2, xmax, STEP)
        self.y = np.arange(ymin + STEP / 2, ymax, STEP)
        self.xx, self.yy = np.meshgrid(self.x, self.y)
        self.rows, self.cols = self.xx.shape
        self.cells = self.rows * self.cols
        self.serial = 0

    def raster(self, shape):
        return intersects_xy(shape, self.xx, self.yy)

    def refresh(self):
        self.circuit = fill_planes(self.circuit)
        self.items, self.nets = copper(self.circuit)
        for pour in self.nets.of('pcb_copper_pour'):
            shape = pour['brep_shape']
            points = lambda ring: [(point['x'], point['y']) for point in ring['vertices']]
            self.items.append(dict(id=pour['pcb_copper_pour_id'], kind='plane', net=self.nets.net(pour), layer=pour['layer'], geometry=Polygon(points(shape['outer_ring']), [points(ring) for ring in shape.get('inner_rings', [])])))

    def groups(self, net):
        items = [item for item in self.items if item['net'] == net]
        parent = list(range(len(items)))
        def find(index):
            while parent[index] != index:
                parent[index] = parent[parent[index]]
                index = parent[index]
            return index
        def join(pair):
            parent[find(pair[0])] = find(pair[1])
        ids = collections.defaultdict(list)
        ports = collections.defaultdict(list)
        for index, item in enumerate(items):
            ids[item['id']].append(index)
            if item.get('source_port_id'):
                ports[item['source_port_id']].append(index)
        for indices in ids.values():
            for index in indices[1:]:
                join((indices[0], index))
        for connected_ports in self.nets.permanent_internal_connections():
            indices = [index for port in connected_ports for index in ports[port]]
            for index in indices[1:]:
                join((indices[0], index))
        for layer in LAYERS:
            chosen = [index for index, item in enumerate(items) if item['layer'] == layer]
            shapes = [items[index]['geometry'] for index in chosen]
            tree = STRtree(shapes)
            for offset, index in enumerate(chosen):
                for hit in tree.query(shapes[offset].buffer(.0006)):
                    other = chosen[int(hit)]
                    if other > index and items[index]['geometry'].distance(items[other]['geometry']) < .0006:
                        join((index, other))
        roots = {find(index) for index, item in enumerate(items) if item.get('source_port_id')}
        groups = collections.defaultdict(list)
        for index, item in enumerate(items):
            if find(index) in roots:
                groups[find(index)].append(item)
        return sorted(groups.values(), key=len)

    def coordinate(self, state):
        layer, cell = divmod(state, self.cells)
        row, col = divmod(cell, self.cols)
        return layer, row, col

    def point(self, state):
        _, row, col = self.coordinate(state)
        return float(self.x[col]), float(self.y[row])

    def solve(self, net):
        self.refresh()
        groups = self.groups(net)
        if len(groups) < 2:
            return False
        width = .4 if net in {'V5', 'V0V9', 'V3V3', 'GND'} else .15
        source_group, targets = groups[0], [item for group in groups[1:] for item in group]
        foreign = [item for item in self.items if item['net'] != net and item['kind'] != 'plane']
        holes = [geom(record, drill=True) for record in self.nets.of('pcb_hole')]
        obstacles = []
        blocked = []
        for layer in OUTER_LAYERS:
            shapes = [item['geometry'].buffer(CLEARANCE + width / 2) for item in foreign if item['layer'] == layer]
            shapes.extend(hole.buffer(CLEARANCE + width / 2) for hole in holes)
            obstacle = unary_union(shapes)
            obstacles.append(obstacle)
            blocked.append(self.raster(obstacle) | ~self.raster(self.boundary.buffer(-(.19 + width / 2))))
        drilled = self.nets.of('pcb_via') + self.nets.of('pcb_plated_hole') + self.nets.of('pcb_hole')
        via_obstacle = unary_union([item['geometry'].buffer(.3 + CLEARANCE) for item in foreign] + [geom(record, drill=True).buffer(.3 + CLEARANCE if record['type'] == 'pcb_hole' else .301) for record in drilled])
        via_allowed = ~self.raster(via_obstacle) & self.raster(self.boundary.buffer(-.491))
        existing_vias = unary_union([item['geometry'] for item in self.items if item['net'] == net and item['kind'] in {'via', 'pad'} and sum(other['id'] == item['id'] and other['layer'] in OUTER_LAYERS for other in self.items) > 1])
        existing_via_mask = self.raster(existing_vias)
        transfer = via_allowed | existing_via_mask
        starts = []
        goals = []
        surface_goals = []
        inner_targets = unary_union([item['geometry'] for item in targets if item['layer'] in {'inner1', 'inner2'}])
        plane_goals = self.raster(inner_targets) & transfer
        for index, layer in enumerate(OUTER_LAYERS):
            start_shape = unary_union([item['geometry'] for item in source_group if item['layer'] == layer])
            goal_shape = unary_union([item['geometry'] for item in targets if item['layer'] == layer])
            starts.append(self.raster(start_shape.buffer(width / 2 - .001)) & ~blocked[index])
            surface = self.raster(goal_shape.buffer(width / 2 - .001))
            surface_goals.append(surface)
            goals.append((surface | plane_goals) & ~blocked[index])
        goal_mask = np.stack(goals)
        start_mask = np.stack(starts)
        if not start_mask.any() or not goal_mask.any():
            raise RuntimeError(f'{net}: no accessible source/target grid nodes')
        heuristic = distance_transform_edt(~(goals[0] | goals[1])) * STEP
        blocked_edges = set()
        for attempt in range(12):
            queue = []
            distances = {}
            parents = {}
            serial = itertools.count()
            for state in np.flatnonzero(start_mask.ravel()):
                state = int(state)
                _, row, col = self.coordinate(state)
                distances[state] = 0.0
                heapq.heappush(queue, (float(heuristic[row, col]), next(serial), state))
            found = None
            while queue:
                estimate, _, state = heapq.heappop(queue)
                layer, row, col = self.coordinate(state)
                cost = distances[state]
                if estimate > cost + heuristic[row, col] + 1e-9:
                    continue
                if goal_mask[layer, row, col]:
                    found = state
                    break
                for dr, dc in [(-1, 0), (1, 0), (0, -1), (0, 1), (-1, -1), (-1, 1), (1, -1), (1, 1)]:
                    nr, nc = row + dr, col + dc
                    if nr < 0 or nr >= self.rows or nc < 0 or nc >= self.cols or blocked[layer][nr, nc]:
                        continue
                    if dr and dc and (blocked[layer][row, nc] or blocked[layer][nr, col]):
                        continue
                    neighbor = layer * self.cells + nr * self.cols + nc
                    if (state, neighbor) in blocked_edges:
                        continue
                    proposed = cost + STEP * (math.sqrt(2) if dr and dc else 1)
                    if proposed < distances.get(neighbor, math.inf):
                        distances[neighbor] = proposed
                        parents[neighbor] = state
                        heapq.heappush(queue, (proposed + float(heuristic[nr, nc]), next(serial), neighbor))
                if transfer[row, col] and not blocked[1 - layer][row, col]:
                    neighbor = (1 - layer) * self.cells + row * self.cols + col
                    proposed = cost + (0.2 if existing_via_mask[row, col] else 3)
                    if proposed < distances.get(neighbor, math.inf):
                        distances[neighbor] = proposed
                        parents[neighbor] = state
                        heapq.heappush(queue, (proposed + float(heuristic[row, col]), next(serial), neighbor))
            if found is None:
                raise RuntimeError(f'{net}: no path at {width} mm width')
            path = [found]
            while path[-1] in parents:
                path.append(parents[path[-1]])
            path.reverse()
            invalid = []
            for a, b in zip(path, path[1:]):
                layer_a = self.coordinate(a)[0]
                if layer_a == self.coordinate(b)[0] and LineString([self.point(a), self.point(b)]).intersects(obstacles[layer_a]):
                    invalid.append((a, b))
            if invalid:
                blocked_edges.update(invalid)
                blocked_edges.update((b, a) for a, b in invalid)
                continue
            self.active_net = net
            self.active_width = width
            self.active_existing_vias = existing_via_mask
            self.add_path(path)
            layer, row, col = self.coordinate(found)
            if not surface_goals[layer][row, col] and not existing_via_mask[row, col]:
                self.add_via(self.point(found))
            print(json.dumps({'net': net, 'groupsBefore': len(groups), 'pathLengthMm': round(distances[found], 3), 'gridNodes': len(path)}), flush=True)
            return True
        raise RuntimeError(f'{net}: could not clear raster corner conflicts')

    def metadata(self):
        return {'source_net_id': self.nets.ids[self.active_net], 'subcircuit_id': self.board['subcircuit_id']}

    def add_via(self, point):
        self.serial += 1
        self.circuit.append(dict(type='pcb_via', pcb_via_id=f'pcb_via_grid_{self.serial}', x=point[0], y=point[1], outer_diameter=.6, hole_diameter=.3, layers=LAYERS, from_layer='top', to_layer='bottom', **self.metadata()))

    def add_path(self, path):
        runs = [[path[0]]]
        for previous, state in zip(path, path[1:]):
            layer, row, col = self.coordinate(state)
            if layer != self.coordinate(previous)[0]:
                if not self.active_existing_vias[row, col]:
                    self.add_via(self.point(state))
                runs.append([state])
            else:
                runs[-1].append(state)
        for run in runs:
            if len(run) < 2:
                continue
            kept = [run[0]]
            last_direction = None
            for index in range(1, len(run)):
                a, b = self.coordinate(run[index - 1]), self.coordinate(run[index])
                direction = (b[1] - a[1], b[2] - a[2])
                if last_direction is not None and direction != last_direction:
                    kept.append(run[index - 1])
                last_direction = direction
            kept.append(run[-1])
            layer = OUTER_LAYERS[self.coordinate(run[0])[0]]
            route = [dict(route_type='wire', x=self.point(state)[0], y=self.point(state)[1], width=self.active_width, layer=layer) for state in kept]
            self.serial += 1
            self.circuit.append(dict(type='pcb_trace', pcb_trace_id=f'pcb_trace_grid_{self.serial}', route=route, **self.metadata()))

source_path, destination = map(Path, sys.argv[1:3])
router = RemainingRouter(json.loads(source_path.read_text()))
router.refresh()
initial = audit(router.circuit)
assert not initial['violations'], initial['violationCounts']
for entry in sorted(initial['unconnectedNets'], key=lambda entry: (entry['net'] not in {'V0V9', 'V5'}, entry['net'])):
    for connection in range(20):
        if not router.solve(entry['net']):
            break
    else:
        raise RuntimeError('Connection limit reached for ' + entry['net'])
router.refresh()
destination.write_text(json.dumps(router.circuit, separators=(',', ':')) + '\n')
report = audit(router.circuit)
destination.with_suffix('.routing-check.json').write_text(json.dumps(report, indent=2) + '\n')
print(json.dumps({'passed': report['passed'], 'unconnectedNets': len(report['unconnectedNets']), 'violations': report['violationCounts']}), flush=True)
if not report['passed']:
    sys.exit(1)