Rewrite DesignGraph builder as original Periscope graph.py.

Src module classifies refs, infers net type/voltage, and prefers board nets
when a .kicad_pcb is present. PinScope graph.py remains in dependency/.

periscope/src/backend/periscopex/graph.py

@@ -0,0 +1,367 @@
+"""Build a Periscope DesignGraph from netlist, BOM, datasheets, and optional PCB.
+
+This is original Periscope orchestration. Inherited PinScope ``graph.py`` stays
+in ``periscope/dependency/`` until this module is the live winner.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from pathlib import Path
+
+from backend.periscopex.models import (
+    CadIndexEntry,
+    Component,
+    ComponentConstraints,
+    ComponentModel,
+    ComponentSpecs,
+    ComponentType,
+    DesignGraph,
+    Net,
+    NetType,
+    PinConnection,
+    SimpleComponentSpecs,
+)
+from backend.periscopex.parsers import parse_bom, parse_netlist_any
+from backend.periscopex.resolve_passives import SkippedItem, resolve_bom, resolved_to_specs
+from backend.periscopex.utils import safe_mpn
+
+# Ref-des prefix → type. Longer prefixes must be checked first (LED before L).
+_REF_PREFIXES: tuple[tuple[str, ComponentType], ...] = (
+    ("LED", ComponentType.DISCRETE),
+    ("RN", ComponentType.RESISTOR),
+    ("FB", ComponentType.INDUCTOR),
+    ("SW", ComponentType.SWITCH),
+    ("TP", ComponentType.TEST_POINT),
+    ("IC", ComponentType.IC),
+    ("MH", ComponentType.MECHANICAL),
+    ("FM", ComponentType.FIDUCIAL),
+    ("R", ComponentType.RESISTOR),
+    ("C", ComponentType.CAPACITOR),
+    ("L", ComponentType.INDUCTOR),
+    ("U", ComponentType.IC),
+    ("J", ComponentType.CONNECTOR),
+    ("X", ComponentType.CRYSTAL),
+    ("Y", ComponentType.CRYSTAL),
+    ("D", ComponentType.DISCRETE),
+    ("Q", ComponentType.DISCRETE),
+    ("T", ComponentType.TRANSFORMER),
+    ("F", ComponentType.FUSE),
+)
+
+_FOOTPRINT_HINTS: tuple[tuple[re.Pattern[str], ComponentType], ...] = (
+    (re.compile(
+        r"(?i)(?:^|[\s_])("
+        r"CONN(?:_|\b)|TERM(?:\b|_BLK)|HEADER|SOCKET|JACK|RECEPTACLE|PLUG|"
+        r"SCREW\s*TERM|PINHEADER|BARREL|BANANA|XT30|XT60|XT90|USB|"
+        r"WURTH\s*746\d|TE\s*282834|TE\s*2828\d|MOLEX|JST"
+        r")"
+    ), ComponentType.CONNECTOR),
+    (re.compile(r"(?i)TestPoint|TEST[_\s]POINT|\bTP_"), ComponentType.TEST_POINT),
+    (re.compile(r"(?i)^LED[\s_]|\bLED\s+\d{3,4}"), ComponentType.DISCRETE),
+    (re.compile(r"(?i)^CAP[\s_]|\bCAP_|CAPACITOR"), ComponentType.CAPACITOR),
+    (re.compile(r"(?i)^RES[\s_]|\bRES_|RESISTOR"), ComponentType.RESISTOR),
+    (re.compile(r"(?i)^IND[\s_]|\bIND_|INDUCTOR"), ComponentType.INDUCTOR),
+    (re.compile(r"(?i)DO214|DO220|SOD\d|SMD?J5|SMB_|SOT-?23"), ComponentType.DISCRETE),
+)
+
+_POWER_PREFIXES = (
+    "VCC", "VDD", "VBUS", "VBAT", "VSYS", "VSUP", "VPWR",
+    "AVDD", "DVDD", "AVCC", "DVCC", "PVDD", "PVCC",
+    "V_",
+)
+_GROUND_NAMES = {"GND", "AGND", "DGND", "PGND", "VSS", "AVSS", "DVSS", "PVSS"}
+_GROUND_SUFFIXES = ("_GND", "GND")
+
+
+def _classify_component(ref: str, footprint: str) -> ComponentType:
+    letters = re.match(r"^[A-Za-z]+", ref)
+    if letters:
+        token = letters.group().upper()
+        for prefix, ctype in _REF_PREFIXES:
+            if token == prefix or token.startswith(prefix):
+                # Require exact prefix match on the letter run (RN not R).
+                if token == prefix:
+                    return ctype
+        for prefix, ctype in _REF_PREFIXES:
+            if token == prefix:
+                return ctype
+    fp = footprint or ""
+    for pattern, ctype in _FOOTPRINT_HINTS:
+        if pattern.search(fp):
+            return ctype
+    return ComponentType.UNKNOWN
+
+
+def _parse_rail_voltage(name: str) -> float | None:
+    m = re.match(r"^\+(\d+)V(\d+)$", name)
+    if m:
+        return float(f"{m.group(1)}.{m.group(2)}")
+    m = re.match(r"^\+(\d+(?:\.\d+)?)V$", name)
+    if m:
+        return float(m.group(1))
+    m = re.search(r"(\d+)V(\d+)", name)
+    if m:
+        return float(f"{m.group(1)}.{m.group(2)}")
+    m = re.search(r"(\d+(?:\.\d+)?)V(?:\d|$|_)", name)
+    if m:
+        return float(m.group(1))
+    return None
+
+
+def _infer_net_properties(name: str) -> tuple[NetType, float | None]:
+    upper = name.upper()
+    if upper in _GROUND_NAMES or any(upper.endswith(s) for s in _GROUND_SUFFIXES):
+        return NetType.GROUND, 0.0
+    if name.startswith("+"):
+        return NetType.POWER, _parse_rail_voltage(name)
+    if any(upper.startswith(p) for p in _POWER_PREFIXES):
+        return NetType.POWER, _parse_rail_voltage(name)
+    if re.match(r"^\d+V\d*", upper):
+        return NetType.POWER, _parse_rail_voltage(name)
+    return NetType.SIGNAL, None
+
+
+def _norm_mpn(s: str) -> str:
+    return re.sub(r"[/_\-\s]", "", s).upper()
+
+
+def _load_datasheets(directory: str | Path) -> dict[str, tuple[Path, ComponentConstraints]]:
+    result: dict[str, tuple[Path, ComponentConstraints]] = {}
+    dirpath = Path(directory)
+    if not dirpath.is_dir():
+        return result
+    for json_file in dirpath.glob("*.json"):
+        raw = json.loads(json_file.read_text())
+        constraints = ComponentConstraints.model_validate(raw)
+        result[constraints.mpn] = (json_file, constraints)
+    return result
+
+
+def _match_datasheet(
+    mpn: str | None,
+    datasheets: dict[str, tuple[Path, ComponentConstraints]],
+) -> tuple[Path | None, ComponentConstraints | None]:
+    if not mpn:
+        return None, None
+    if mpn in datasheets:
+        return datasheets[mpn]
+    want = _norm_mpn(mpn)
+    for ds_mpn, hit in datasheets.items():
+        if _norm_mpn(ds_mpn) == want:
+            return hit
+    return None, None
+
+
+def _load_component_models(directory: str | Path) -> dict[str, ComponentSpecs]:
+    result: dict[str, ComponentSpecs] = {}
+    dirpath = Path(directory)
+    if not dirpath.is_dir():
+        return result
+    for json_file in dirpath.glob("*.json"):
+        raw = json.loads(json_file.read_text())
+        model = ComponentModel.model_validate(raw)
+        result[model.mpn] = model.specs
+    return result
+
+
+def _save_component_model(mpn: str, specs: ComponentSpecs, directory: Path) -> None:
+    directory.mkdir(parents=True, exist_ok=True)
+    model = ComponentModel(mpn=mpn, specs=specs)
+    (directory / f"{safe_mpn(mpn)}.json").write_text(
+        model.model_dump_json(indent=2) + "\n"
+    )
+
+
+def _bom_rows(bom: dict) -> dict[str, dict]:
+    out: dict[str, dict] = {}
+    for ref, entry in bom.items():
+        row = {"mpn": entry.get("mpn"), "value": entry.get("value", "")}
+        if "dnp" in entry:
+            row["dnp"] = entry.get("dnp")
+        if entry.get("variant") is not None:
+            row["variant"] = entry.get("variant")
+        out[ref] = row
+    return out
+
+
+def _apply_pcb_nets(pcb_path: Path, parts: dict, raw_nets: dict) -> tuple[dict, dict]:
+    from backend.periscopex.parsers_kicad_pcb import nets_from_pcb, parse_kicad_pcb
+
+    layout = parse_kicad_pcb(pcb_path)
+    pcb_nets = nets_from_pcb(layout)
+    if not pcb_nets:
+        return parts, raw_nets
+    for ref, fp in layout.footprints.items():
+        parts.setdefault(ref, fp.footprint or "")
+    return parts, pcb_nets
+
+
+def _merge_kicad_fields(netlist_path: Path, bom: dict) -> dict[str, dict]:
+    from backend.periscopex.parsers_kicad import kicad_part_fields
+
+    schematic_fields: dict[str, dict] = {}
+    for ref, extra in kicad_part_fields(netlist_path).items():
+        schematic_fields[ref] = {
+            "mpn": extra.get("mpn"),
+            "value": extra.get("value", ""),
+            "cad_uuid": extra.get("cad_uuid") or "",
+            "cad_sheet": extra.get("cad_sheet") or "",
+        }
+        entry = bom.setdefault(
+            ref,
+            {"value": "", "footprint": "", "mpn": None, "lcsc": None, "datasheet_url": None},
+        )
+        if extra.get("mpn") and (not entry.get("mpn") or entry.get("mpn") == entry.get("value")):
+            entry["mpn"] = extra["mpn"]
+        if extra.get("lcsc") and not entry.get("lcsc"):
+            entry["lcsc"] = extra["lcsc"]
+        if extra.get("value") and not entry.get("value"):
+            entry["value"] = extra["value"]
+        if extra.get("footprint") and not entry.get("footprint"):
+            entry["footprint"] = extra["footprint"]
+    return schematic_fields
+
+
+def _pin_name_for(
+    ref: str,
+    pin_num: str,
+    constraints_by_ref: dict[str, ComponentConstraints],
+    components: dict[str, Component],
+    mpn_specs: dict[str, ComponentSpecs],
+) -> str | None:
+    constraints = constraints_by_ref.get(ref)
+    if constraints:
+        pin_obj = constraints.pin_by_number(pin_num)
+        return pin_obj.name if pin_obj else None
+    comp = components.get(ref)
+    if not comp or not comp.mpn:
+        return None
+    specs = mpn_specs.get(comp.mpn)
+    if isinstance(specs, SimpleComponentSpecs) and specs.pintable:
+        pin_obj = specs.pin_by_number(pin_num)
+        return pin_obj.name if pin_obj else None
+    return None
+
+
+def build_graph(
+    netlist_path: str | Path,
+    bom_path: str | Path,
+    datasheets_dir: str | Path = "datasheets/extracted",
+    patterns_dir: str | Path = "component-patterns",
+    component_models_dir: str | Path = "component-models",
+    *,
+    reference_col: str = "Reference",
+    mpn_col: str = "Manufacturer Part Number",
+    skipped: list[SkippedItem] | None = None,
+    include_subdesigns: set[str] | None = None,
+    pcb_path: str | Path | None = None,
+) -> DesignGraph:
+    """Deterministic graph: BOM → netlist → optional board nets → datasheets → DesignGraph."""
+    bom = parse_bom(bom_path, reference_col=reference_col, mpn_col=mpn_col)
+    bom_fields = _bom_rows(bom)
+    parts, raw_nets, fmt = parse_netlist_any(
+        netlist_path,
+        known_refs=set(bom.keys()),
+        include_subdesigns=include_subdesigns,
+    )
+    if pcb_path is not None:
+        pcb = Path(pcb_path)
+        if pcb.is_file():
+            parts, raw_nets = _apply_pcb_nets(pcb, parts, raw_nets)
+
+    schematic_fields: dict[str, dict] = {}
+    if fmt.startswith("kicad"):
+        schematic_fields = _merge_kicad_fields(Path(netlist_path), bom)
+
+    datasheets = _load_datasheets(datasheets_dir)
+    models_dir = Path(component_models_dir)
+    mpn_specs: dict[str, ComponentSpecs] = _load_component_models(models_dir)
+    mpn_subtype: dict[str, str] = {}
+
+    for rp in resolve_bom(
+        bom_path, patterns_dir, reference_col=reference_col, mpn_col=mpn_col, skipped=skipped
+    ):
+        if rp.component_subtype:
+            mpn_subtype[rp.mpn] = rp.component_subtype
+        if rp.mpn in mpn_specs:
+            continue
+        try:
+            specs = resolved_to_specs(rp)
+            mpn_specs[rp.mpn] = specs
+            _save_component_model(rp.mpn, specs, models_dir)
+        except Exception as e:
+            if skipped is not None:
+                skipped.append(SkippedItem(rp.mpn, "passive_specs", str(e)))
+
+    if not parts:
+        net_refs = {ref for pins in raw_nets.values() for ref, _ in pins}
+        parts = {ref: bom.get(ref, {}).get("footprint", "") for ref in (set(bom) | net_refs)}
+
+    components: dict[str, Component] = {}
+    for ref, footprint in parts.items():
+        bom_entry = bom.get(ref, {})
+        value = bom_entry.get("value", "")
+        mpn = bom_entry.get("mpn") or None
+        ctype = _classify_component(ref, footprint)
+        if not mpn and ctype == ComponentType.IC:
+            mpn = (value or "").strip() or None
+        components[ref] = Component(
+            reference=ref,
+            value=value,
+            footprint=footprint,
+            component_type=ctype,
+            mpn=mpn,
+            pins={},
+        )
+
+    constraints_by_ref: dict[str, ComponentConstraints] = {}
+    for ref, comp in components.items():
+        if not comp.mpn:
+            continue
+        _, constraints = _match_datasheet(comp.mpn, datasheets)
+        if constraints:
+            constraints_by_ref[ref] = constraints
+            if constraints.component_subtype:
+                comp.component_subtype = constraints.component_subtype
+        if comp.mpn in mpn_specs:
+            comp.specs = mpn_specs[comp.mpn]
+            if not comp.component_subtype:
+                specs = mpn_specs[comp.mpn]
+                subtype = getattr(specs, "component_subtype", None)
+                if subtype:
+                    comp.component_subtype = subtype
+        if not comp.component_subtype and comp.mpn in mpn_subtype:
+            comp.component_subtype = mpn_subtype[comp.mpn]
+
+    nets: dict[str, Net] = {}
+    for net_name, pin_list in raw_nets.items():
+        net_type, voltage = _infer_net_properties(net_name)
+        pin_connections: list[PinConnection] = []
+        for ref, pin_num in pin_list:
+            if ref in components:
+                components[ref].pins[pin_num] = net_name
+            pin_connections.append(PinConnection(
+                component_ref=ref,
+                pin_number=pin_num,
+                pin_name=_pin_name_for(ref, pin_num, constraints_by_ref, components, mpn_specs),
+            ))
+        nets[net_name] = Net(
+            name=net_name, net_type=net_type, voltage=voltage, pins=pin_connections
+        )
+
+    cad_index: dict[str, CadIndexEntry] = {}
+    for ref, extra in schematic_fields.items():
+        uuid = extra.get("cad_uuid") or ""
+        sheet = extra.get("cad_sheet") or ""
+        if uuid or sheet:
+            cad_index[ref] = CadIndexEntry(uuid=uuid, sheet=sheet)
+
+    return DesignGraph(
+        components=components,
+        nets=nets,
+        bom_fields=bom_fields,
+        schematic_fields=schematic_fields,
+        cad_index=cad_index,
+    )

tests/test_periscope_graph_rewrite.py

@@ -0,0 +1,50 @@
+"""Native graph builder lives in periscope/src; via≠pad stays on the KiCad PCB parser."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import backend.periscopex.graph as graph
+from backend.periscopex.graph import _classify_component, _infer_net_properties
+from backend.periscopex.models import ComponentType, NetType
+from backend.periscopex.parsers_kicad_pcb import LayoutPad, LayoutVia
+
+
+def test_graph_module_is_src_rewrite():
+    path = Path(graph.__file__).resolve()
+    assert path.parts[-5:] == ("periscope", "src", "backend", "periscopex", "graph.py")
+    text = path.read_text(encoding="utf-8")
+    assert "Native Periscope overlay" not in text
+    assert "original Periscope orchestration" in text
+
+
+def test_classify_prefixes_and_unknown():
+    assert _classify_component("FB1", "") == ComponentType.INDUCTOR
+    assert _classify_component("RN4", "") == ComponentType.RESISTOR
+    assert _classify_component("F1", "") == ComponentType.FUSE
+    assert _classify_component("R10", "") == ComponentType.RESISTOR
+    assert _classify_component("9", "") == ComponentType.UNKNOWN
+    assert _classify_component("9", "CONN_HEADER") == ComponentType.CONNECTOR
+
+
+def test_infer_ground_power_signal():
+    assert _infer_net_properties("GND") == (NetType.GROUND, 0.0)
+    assert _infer_net_properties("AGND") == (NetType.GROUND, 0.0)
+    ntype, volts = _infer_net_properties("+3V3")
+    assert ntype == NetType.POWER
+    assert volts == 3.3
+    ntype, volts = _infer_net_properties("3V3_DIGITAL")
+    assert ntype == NetType.POWER
+    assert volts == 3.3
+    ntype, volts = _infer_net_properties("I2C1_SCL")
+    assert ntype == NetType.SIGNAL
+    assert volts is None
+
+
+def test_via_is_not_pad():
+    via = LayoutVia(x=0.0, y=0.0, net="GND", drill=0.3)
+    pad = LayoutPad(number="1", x=0.0, y=0.0, net="GND")
+    assert type(via) is not type(pad)
+    assert not hasattr(via, "number")
+    assert pad.number == "1"
+    assert via.net == "GND"