diff --git a/periscope/src/backend/periscopex/graph.py b/periscope/src/backend/periscopex/graph.py new file mode 100644 index 0000000..2a2be0c --- /dev/null +++ b/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, + ) diff --git a/tests/test_periscope_graph_rewrite.py b/tests/test_periscope_graph_rewrite.py new file mode 100644 index 0000000..a9dd7f2 --- /dev/null +++ b/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"