mohan-bee/compact-pir

validation/check-native-geometry.py

#!/usr/bin/env python3
"""Read-only native routing checks; writes only a JSON report next to its input.

This is not a replacement for native clearance/physical connectivity DRC.
Source connectivity is compared independently of generated IDs. Protected-area
checks use capsule/disc geometry including full copper width, in Circuit JSON
coordinates (X right, Y up). No external Python dependencies are required.
"""
import argparse
from collections import Counter, defaultdict
import hashlib
import json
import math
from pathlib import Path

REGIONS = {
    "rf_bottom_reference": [-13.7, -2.8, -6.1, 3.1],
    "crystal_bottom_reference": [-5.2, 2.8, -0.65, 7.9],
}
LAYERS = {"top", "bottom"}
EPS = 1e-8


class NetResolver:
    def __init__(self, rows):
        self.parents = {}
        self.nets = {r["source_net_id"]: r for r in rows if r["type"] == "source_net"}
        self.traces = {r["source_trace_id"]: r for r in rows if r["type"] == "source_trace"}
        self.ports = {r["source_port_id"]: r for r in rows if r["type"] == "source_port"}
        self.pcb_ports = {r["pcb_port_id"]: r for r in rows if r["type"] == "pcb_port"}
        self.pcb_traces = {r["pcb_trace_id"]: r for r in rows if r["type"] == "pcb_trace"}
        for row in [*self.nets.values(), *self.traces.values(), *self.ports.values()]:
            own = row[row["type"] + "_id"]
            self.find(own)
            key = row.get("subcircuit_connectivity_map_key")
            if key:
                self.join(own, "key:" + key)
            for field in ("connected_source_port_ids", "connected_source_net_ids"):
                for other in row.get(field, []):
                    self.join(own, other)
        for row in rows:
            if row["type"] == "source_component_internal_connection":
                ids = row.get("source_port_ids", [])
                for other in ids[1:]:
                    self.join(ids[0], other)
        self.names = defaultdict(set)
        for nid, row in self.nets.items():
            self.names[self.find(nid)].add(row["name"])

    def find(self, item):
        if item not in self.parents:
            self.parents[item] = item
        while self.parents[item] != item:
            self.parents[item] = self.parents[self.parents[item]]
            item = self.parents[item]
        return item

    def join(self, a, b):
        self.parents[self.find(b)] = self.find(a)

    def source_names(self, item):
        return set(self.names.get(self.find(item), set()))

    def resolve(self, row, include_parent_trace=True):
        """Use explicit source identity/connectivity keys, never parse an ID suffix."""
        names = set()
        for field in ("source_net_id", "source_trace_id", "source_port_id"):
            if row.get(field):
                names.update(self.source_names(row[field]))
        if row.get("subcircuit_connectivity_map_key"):
            names.update(self.source_names("key:" + row["subcircuit_connectivity_map_key"]))
        if row.get("connection_name") in self.nets:
            names.update(self.source_names(row["connection_name"]))
        if include_parent_trace and row.get("pcb_trace_id") in self.pcb_traces:
            parent = self.pcb_traces[row["pcb_trace_id"]]
            if parent is not row:
                names.update(self.resolve(parent, False))
        for port_id in row.get("connectsTo", []):
            port = self.pcb_ports.get(port_id)
            if port:
                names.update(self.source_names(port["source_port_id"]))
        return names


def normalized_pins(rows, resolver):
    components = {r["source_component_id"]: r for r in rows if r["type"] == "source_component"}
    result = {}
    for pid, port in resolver.ports.items():
        component = components.get(port.get("source_component_id"))
        if not component:
            continue  # Native stitching vias have synthetic ports, not chip pins.
        pin = str(port.get("pin_number", port["name"]))
        identity = component["name"] + ".pin" + pin
        value = sorted(resolver.source_names(pid))
        if identity in result and result[identity] != value:
            raise ValueError("Conflicting duplicate source pin: " + identity)
        result[identity] = value
    return result


def distance_point_rect(p, rect):
    return math.hypot(max(rect[0] - p[0], 0, p[0] - rect[2]),
                      max(rect[1] - p[1], 0, p[1] - rect[3]))


def distance_point_segment(p, a, b):
    dx, dy = b[0] - a[0], b[1] - a[1]
    den = dx * dx + dy * dy
    t = 0 if den == 0 else max(0, min(1, ((p[0] - a[0]) * dx + (p[1] - a[1]) * dy) / den))
    return math.hypot(p[0] - a[0] - t * dx, p[1] - a[1] - t * dy)


def cross(a, b, c):
    return (b[0] - a[0]) * (c[1] - a[1]) - (b[1] - a[1]) * (c[0] - a[0])


def distance_segments(a, b, c, d):
    if cross(a, b, c) * cross(a, b, d) < 0 and cross(c, d, a) * cross(c, d, b) < 0:
        return 0.0
    return min(distance_point_segment(a, c, d), distance_point_segment(b, c, d),
               distance_point_segment(c, a, b), distance_point_segment(d, a, b))


def distance_segment_rect(a, b, rect):
    if distance_point_rect(a, rect) == 0 or distance_point_rect(b, rect) == 0:
        return 0.0
    corners = [(rect[0], rect[1]), (rect[2], rect[1]), (rect[2], rect[3]), (rect[0], rect[3])]
    return min(distance_segments(a, b, c, d) for c, d in zip(corners, corners[1:] + corners[:1]))


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("circuit", type=Path)
    parser.add_argument("--reference", type=Path,
                        default=Path(__file__).resolve().parents[2] / "outputs/circuit.json")
    args = parser.parse_args()
    raw = args.circuit.read_bytes()
    rows = json.loads(raw)
    baseline = json.loads(args.reference.read_text())
    resolver, old_resolver = NetResolver(rows), NetResolver(baseline)
    pins, old_pins = normalized_pins(rows, resolver), normalized_pins(baseline, old_resolver)
    pin_differences = [{"pin": key, "reference_nets": old_pins.get(key), "current_nets": pins.get(key)}
                       for key in sorted(pins.keys() | old_pins.keys()) if pins.get(key) != old_pins.get(key)]
    components = sorted(r["name"] for r in rows if r["type"] == "source_component")
    old_components = sorted(r["name"] for r in baseline if r["type"] == "source_component")
    boards = [r for r in rows if r["type"] == "pcb_board"]
    dimensions_ok = len(boards) == 1 and all(abs(boards[0].get(k, 0) - v) < EPS
                                           for k, v in (("width", 29), ("height", 25), ("num_layers", 2)))
    layer_errors, unresolved, protected, invalid_geometry = [], [], [], []
    bottom_segments = 0
    non_ground_vias = 0
    route_via_checks = 0

    def check_layers(row, declared, context):
        for layer in declared:
            if layer not in LAYERS:
                layer_errors.append({"item": row.get(row["type"] + "_id"), "context": context, "layer": layer})

    def check_disc(row, x, y, diameter, names, context):
        if names == {"GND"}:
            return
        if not math.isfinite(diameter) or diameter <= 0:
            invalid_geometry.append({"item": context, "reason": "Missing/invalid via copper diameter"})
            return
        for name, rect in REGIONS.items():
            penetration = diameter / 2 - distance_point_rect((x, y), rect)
            if penetration > EPS:
                protected.append({"item": context, "kind": "via", "region": name, "nets": sorted(names),
                                  "center_mm": [x, y], "copper_diameter_mm": diameter,
                                  "copper_intrusion_mm": penetration})

    for row in rows:
        typ = row["type"]
        if typ in ("pcb_smtpad", "pcb_plated_hole", "pcb_via", "pcb_port", "pcb_copper_pour"):
            declared = list(row.get("layers", []))
            for key in ("layer", "from_layer", "to_layer"):
                if row.get(key):
                    declared.append(row[key])
            check_layers(row, declared, "copper item")
        if typ not in ("pcb_trace", "pcb_via"):
            continue
        names = resolver.resolve(row)
        identity = row[typ + "_id"]
        if not names:
            unresolved.append({"item": identity, "kind": typ})
        if typ == "pcb_via":
            non_ground_vias += names != {"GND"}
            check_disc(row, row["x"], row["y"], row.get("outer_diameter", 0), names, identity)
            continue
        route = row.get("route", [])
        for i, point in enumerate(route):
            declared = [point[k] for k in ("layer", "from_layer", "to_layer") if point.get(k)]
            check_layers(row, declared, "route point " + str(i))
            if point.get("route_type") == "via":
                route_via_checks += 1
                check_disc(row, point["x"], point["y"],
                           point.get("via_diameter", point.get("outer_diameter", 0)), names,
                           identity + ":route_via" + str(i))
        for i, (a, b) in enumerate(zip(route, route[1:])):
            at, bt = a.get("route_type"), b.get("route_type")
            if at == bt == "wire":
                if a.get("layer") != b.get("layer"):
                    invalid_geometry.append({"item": identity, "segment": i, "reason": "Wire layer transition lacks explicit via point"})
                    continue
                layer, width = a.get("layer"), a.get("width", b.get("width", 0))
            elif at == "wire" and bt == "via":
                layer, width = a.get("layer"), a.get("width", 0)
            elif at == "via" and bt == "wire":
                layer, width = b.get("layer"), b.get("width", 0)
            else:
                continue
            if not math.isfinite(width) or width <= 0:
                invalid_geometry.append({"item": identity, "segment": i, "reason": "Missing/invalid conductor width"})
                continue
            if layer != "bottom" or names == {"GND"}:
                continue
            bottom_segments += 1
            start, end = (a["x"], a["y"]), (b["x"], b["y"])
            for name, rect in REGIONS.items():
                penetration = width / 2 - distance_segment_rect(start, end, rect)
                if penetration > EPS:
                    protected.append({"item": identity, "segment": i, "kind": "bottom trace", "region": name,
                                      "nets": sorted(names), "start_mm": start, "end_mm": end,
                                      "width_mm": width, "copper_intrusion_mm": penetration})
    errors = [r for r in rows if r["type"].endswith("_error")]
    success = (dimensions_ok and components == old_components and not pin_differences and not layer_errors
               and not unresolved and not protected and not invalid_geometry and not errors)
    report = {
        "input": str(args.circuit.resolve()), "input_sha256": hashlib.sha256(raw).hexdigest(),
        "reference": str(args.reference.resolve()),
        "scope": "Dimensions, allowed copper layers, source component/pin-net equivalence, full-width protected RF/crystal trace/via geometry, emitted build errors. Requires separate physical clearance/connectivity DRC.",
        "result": "PASS" if success else "FAIL",
        "board": {k: boards[0].get(k) for k in ("width", "height", "num_layers", "thickness")} if boards else None,
        "dimensions_29x25mm_two_layers": dimensions_ok,
        "counts": dict(Counter(r["type"] for r in rows)),
        "component_references_equal": components == old_components,
        "normalized_pin_count": len(pins), "reference_pin_count": len(old_pins),
        "normalized_component_pin_net_differences": pin_differences,
        "invalid_copper_layers": layer_errors, "unresolved_trace_or_via_nets": unresolved,
        "protected_rectangles_circuit_json_mm": REGIONS,
        "non_ground_bottom_trace_segments_checked": bottom_segments,
        "non_ground_via_records_checked": non_ground_vias,
        "inline_route_via_points_checked": route_via_checks,
        "protected_region_violations": protected,
        "invalid_route_geometry": invalid_geometry,
        "emitted_build_errors": errors,
    }
    output = args.circuit.with_name(args.circuit.stem + ".geometry-check.json")
    output.write_text(json.dumps(report, indent=2) + "\n")
    print(json.dumps({"report": str(output), "result": report["result"],
                      "dimensions_match": dimensions_ok, "pin_net_differences": len(pin_differences),
                      "invalid_layers": len(layer_errors), "unresolved_nets": len(unresolved),
                      "protected_region_violations": len(protected), "invalid_route_geometry": len(invalid_geometry),
                      "emitted_build_errors": len(errors)}, indent=2))
    raise SystemExit(0 if success else 1)


if __name__ == "__main__":
    main()