Ship Sprint 0 CAD foundations so findings, KiCad hierarchy, BOM match, eval, and optional PCB ingest have a stable contract.
Extend Finding with rule_id/net/pins, flatten multi-sheet .kicad_sch, flag MPN mismatches, score simple_project, and parse .kicad_pcb without running layout DRC. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -0,0 +1,73 @@
|
||||
"""BOM vs schematic property matching.
|
||||
|
||||
Compares per-reference MPN/value from the schematic property table against
|
||||
the uploaded BOM. Silent when the schematic map is empty (PADS/EDIF) so
|
||||
we never invent orphans from a format that has no schematic properties.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from backend.pinscopex.models import Finding
|
||||
|
||||
|
||||
def _norm_mpn(value: object) -> str:
|
||||
return " ".join(str(value or "").split()).upper()
|
||||
|
||||
|
||||
def check_bom_schematic_match(
|
||||
schematic: dict[str, dict],
|
||||
bom: dict[str, dict],
|
||||
) -> list[Finding]:
|
||||
if not schematic:
|
||||
return []
|
||||
|
||||
findings: list[Finding] = []
|
||||
refs = sorted(set(schematic) | set(bom))
|
||||
for ref in refs:
|
||||
if ref.startswith("#"):
|
||||
continue
|
||||
sch = schematic.get(ref) or {}
|
||||
bom_row = bom.get(ref) or {}
|
||||
if ref not in schematic:
|
||||
findings.append(Finding(
|
||||
designator=ref,
|
||||
mpn=str(bom_row.get("mpn") or ""),
|
||||
aspect="bom_match",
|
||||
source="bom_match",
|
||||
status="WARNING",
|
||||
finding=(
|
||||
f"BOM lists {ref} but the schematic has no such reference."
|
||||
),
|
||||
why=(
|
||||
"An extra BOM line that is not in the netlist will not be "
|
||||
"validated against a datasheet and may indicate a stale BOM."
|
||||
),
|
||||
recommendation=f"Remove {ref} from the BOM or add it to the schematic.",
|
||||
rule_id="PS-BOM-002",
|
||||
pins=[],
|
||||
))
|
||||
continue
|
||||
sch_mpn = _norm_mpn(sch.get("mpn"))
|
||||
bom_mpn = _norm_mpn(bom_row.get("mpn"))
|
||||
if sch_mpn and bom_mpn and sch_mpn != bom_mpn:
|
||||
findings.append(Finding(
|
||||
designator=ref,
|
||||
mpn=str(sch.get("mpn") or ""),
|
||||
aspect="bom_match",
|
||||
source="bom_match",
|
||||
status="ERROR",
|
||||
finding=(
|
||||
f"{ref} schematic MPN '{sch.get('mpn')}' does not match "
|
||||
f"BOM MPN '{bom_row.get('mpn')}'."
|
||||
),
|
||||
why=(
|
||||
"Datasheet review and library lookup follow one MPN. "
|
||||
"A mismatch means the wrong die or a stale BOM row."
|
||||
),
|
||||
recommendation=(
|
||||
f"Make {ref}'s BOM and schematic MPN identical, then re-run."
|
||||
),
|
||||
rule_id="PS-BOM-001",
|
||||
pins=[ref],
|
||||
))
|
||||
return findings
|
||||
@@ -0,0 +1,144 @@
|
||||
"""Score a validation report against a golden key set.
|
||||
|
||||
Used by the simple_project eval harness: finding counts, % Unverified,
|
||||
citation hit-rate among LLM quotes, precision/recall vs golden keys.
|
||||
Deterministic checks without a quote are excluded from the citation
|
||||
denominator so pin-mux/BOM noise cannot inflate the rate.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from backend.pinscopex.models import DesignGraph, Finding, ValidationReport
|
||||
from backend.pinscopex.pin_mux_check import check_pin_mux_feasibility
|
||||
from backend.pinscopex.led_current_check import check_led_current
|
||||
from backend.pinscopex.passive_rail_check import (
|
||||
check_i2c_pullups,
|
||||
check_reset_pullups,
|
||||
check_supply_decoupling,
|
||||
)
|
||||
from backend.pinscopex.bom_match_check import check_bom_schematic_match
|
||||
|
||||
|
||||
class EvalScores(BaseModel):
|
||||
finding_count: int
|
||||
by_status: dict[str, int]
|
||||
unverified_pct: float
|
||||
citation_hit_rate: float | None
|
||||
precision: float
|
||||
recall: float
|
||||
extra_keys: list[str]
|
||||
missing_keys: list[str]
|
||||
graph_ok: bool = True
|
||||
graph_errors: list[str] = []
|
||||
|
||||
|
||||
def finding_key(f: Finding) -> str:
|
||||
if f.rule_id:
|
||||
return f"{f.rule_id}|{f.designator}|{f.net or ''}"
|
||||
return f"{f.source or 'review'}|{f.designator}|{f.net or f.finding}"
|
||||
|
||||
|
||||
def _is_review(f: Finding) -> bool:
|
||||
return not f.source or f.source == "review"
|
||||
|
||||
|
||||
def citation_hit_rate(findings: list[Finding]) -> float | None:
|
||||
quoted = [
|
||||
f for f in findings
|
||||
if _is_review(f) and (f.source_quote or "").strip()
|
||||
]
|
||||
if not quoted:
|
||||
return None
|
||||
hits = sum(1 for f in quoted if not (f.why or "").startswith("Unverified:"))
|
||||
return hits / len(quoted)
|
||||
|
||||
|
||||
def unverified_pct(findings: list[Finding]) -> float:
|
||||
if not findings:
|
||||
return 0.0
|
||||
n = sum(1 for f in findings if (f.why or "").startswith("Unverified:"))
|
||||
return 100.0 * n / len(findings)
|
||||
|
||||
|
||||
def score_keys(produced: set[str], golden: set[str]) -> tuple[float, float, list[str], list[str]]:
|
||||
extra = sorted(produced - golden)
|
||||
missing = sorted(golden - produced)
|
||||
precision = 1.0 if not produced else len(produced & golden) / len(produced)
|
||||
recall = 1.0 if not golden else len(produced & golden) / len(golden)
|
||||
return precision, recall, extra, missing
|
||||
|
||||
|
||||
def run_deterministic_on_graph(graph: DesignGraph) -> list[Finding]:
|
||||
cmap: dict = {}
|
||||
out: list[Finding] = []
|
||||
out.extend(check_pin_mux_feasibility(graph, cmap))
|
||||
out.extend(check_led_current(graph))
|
||||
out.extend(check_supply_decoupling(graph, cmap))
|
||||
out.extend(check_i2c_pullups(graph, cmap))
|
||||
out.extend(check_reset_pullups(graph, cmap))
|
||||
out.extend(check_bom_schematic_match(graph.schematic_fields, graph.bom_fields))
|
||||
return out
|
||||
|
||||
|
||||
def score_report(
|
||||
findings: list[Finding],
|
||||
golden_keys: set[str],
|
||||
*,
|
||||
graph: DesignGraph | None = None,
|
||||
golden_meta: dict | None = None,
|
||||
) -> EvalScores:
|
||||
keys = {finding_key(f) for f in findings}
|
||||
precision, recall, extra, missing = score_keys(keys, golden_keys)
|
||||
by_status: dict[str, int] = {"ERROR": 0, "WARNING": 0, "INFO": 0}
|
||||
for f in findings:
|
||||
by_status[f.status] = by_status.get(f.status, 0) + 1
|
||||
graph_errors: list[str] = []
|
||||
if graph is not None and golden_meta:
|
||||
for ref in golden_meta.get("required_refs") or []:
|
||||
if ref not in graph.components:
|
||||
graph_errors.append(f"missing ref {ref}")
|
||||
min_c = golden_meta.get("min_components")
|
||||
if min_c and len(graph.components) < int(min_c):
|
||||
graph_errors.append(
|
||||
f"components {len(graph.components)} < {min_c}"
|
||||
)
|
||||
min_n = golden_meta.get("min_nets")
|
||||
if min_n and len(graph.nets) < int(min_n):
|
||||
graph_errors.append(f"nets {len(graph.nets)} < {min_n}")
|
||||
return EvalScores(
|
||||
finding_count=len(findings),
|
||||
by_status=by_status,
|
||||
unverified_pct=unverified_pct(findings),
|
||||
citation_hit_rate=citation_hit_rate(findings),
|
||||
precision=precision,
|
||||
recall=recall,
|
||||
extra_keys=extra,
|
||||
missing_keys=missing,
|
||||
graph_ok=not graph_errors,
|
||||
graph_errors=graph_errors,
|
||||
)
|
||||
|
||||
|
||||
def eval_simple_project(
|
||||
root: str | Path,
|
||||
report: ValidationReport | None = None,
|
||||
) -> EvalScores:
|
||||
root = Path(root)
|
||||
graph = DesignGraph.model_validate_json(
|
||||
(root / "design_graph.json").read_text()
|
||||
)
|
||||
golden = {}
|
||||
gpath = root / "eval_golden.json"
|
||||
if gpath.is_file():
|
||||
import json
|
||||
golden = json.loads(gpath.read_text())
|
||||
if report is not None:
|
||||
findings = list(report.findings)
|
||||
else:
|
||||
findings = run_deterministic_on_graph(graph)
|
||||
keys = set(golden.get("deterministic_keys") or [])
|
||||
return score_report(findings, keys, graph=graph, golden_meta=golden)
|
||||
@@ -269,6 +269,11 @@ def build_graph(
|
||||
# only tokenise correctly with the BOM's ref list as a lookup. EDIF
|
||||
# netlists ignore known_refs (designators are unambiguous tokens).
|
||||
bom = parse_bom(bom_path, reference_col=reference_col, mpn_col=mpn_col)
|
||||
bom_fields = {
|
||||
ref: {"mpn": entry.get("mpn"), "value": entry.get("value", "")}
|
||||
for ref, entry in bom.items()
|
||||
}
|
||||
schematic_fields: dict[str, dict] = {}
|
||||
parts, raw_nets, fmt = parse_netlist_any(
|
||||
netlist_path,
|
||||
known_refs=set(bom.keys()),
|
||||
@@ -277,6 +282,10 @@ def build_graph(
|
||||
if fmt.startswith("kicad"):
|
||||
from backend.pinscopex.parsers_kicad import kicad_part_fields
|
||||
for ref, extra in kicad_part_fields(netlist_path).items():
|
||||
schematic_fields[ref] = {
|
||||
"mpn": extra.get("mpn"),
|
||||
"value": extra.get("value", ""),
|
||||
}
|
||||
entry = bom.setdefault(
|
||||
ref,
|
||||
{"value": "", "footprint": "", "mpn": None, "lcsc": None, "datasheet_url": None},
|
||||
@@ -397,4 +406,9 @@ def build_graph(
|
||||
pins=pin_connections,
|
||||
)
|
||||
|
||||
return DesignGraph(components=components, nets=nets)
|
||||
return DesignGraph(
|
||||
components=components,
|
||||
nets=nets,
|
||||
bom_fields=bom_fields,
|
||||
schematic_fields=schematic_fields,
|
||||
)
|
||||
|
||||
@@ -244,6 +244,12 @@ class DesignGraph(BaseModel):
|
||||
"""
|
||||
components: dict[str, Component] = {}
|
||||
nets: dict[str, Net] = {}
|
||||
# KiCad property table vs uploaded BOM (empty on PADS/EDIF).
|
||||
bom_fields: dict[str, dict] = {}
|
||||
schematic_fields: dict[str, dict] = {}
|
||||
# KiCad property table vs uploaded BOM (empty on PADS/EDIF).
|
||||
bom_fields: dict[str, dict] = {}
|
||||
schematic_fields: dict[str, dict] = {}
|
||||
|
||||
# -- Traversal helpers --------------------------------------------------
|
||||
|
||||
@@ -330,6 +336,12 @@ class Finding(BaseModel):
|
||||
recommendation: str = ""
|
||||
reference: str = ""
|
||||
source: str | None = None # None/"review" = LLM; "pin_mux_check"/"led_current_check"/"supply_decoupling_check"/… = deterministic
|
||||
net: str | None = None # net name for CAD telemetry / SI filters
|
||||
pins: list[str] = [] # e.g. ["U3.54"] for pan-and-zoom
|
||||
rule_id: str | None = None # deterministic id, e.g. PS-MUX-001
|
||||
cad_sheet: str | None = None # schematic sheet filename for plugin sync
|
||||
cad_uuid: str | None = None # KiCad symbol/pin uuid
|
||||
variant: str | None = None # DNP / ECO / assembly variant
|
||||
|
||||
|
||||
class ValidationReport(BaseModel):
|
||||
@@ -420,3 +432,42 @@ class ResolvedPassive(BaseModel):
|
||||
power_rating: str | None = None
|
||||
dielectric: str | None = None
|
||||
raw_fields: dict[str, str] = {}
|
||||
|
||||
|
||||
class LayoutPad(BaseModel):
|
||||
number: str
|
||||
x: float
|
||||
y: float
|
||||
net: str = ""
|
||||
|
||||
|
||||
class LayoutFootprint(BaseModel):
|
||||
reference: str
|
||||
footprint: str = ""
|
||||
x: float
|
||||
y: float
|
||||
layer: str = ""
|
||||
pads: list[LayoutPad] = []
|
||||
|
||||
|
||||
class LayoutSegment(BaseModel):
|
||||
start: tuple[float, float]
|
||||
end: tuple[float, float]
|
||||
width: float = 0.0
|
||||
layer: str = ""
|
||||
net: str = ""
|
||||
|
||||
|
||||
class LayoutVia(BaseModel):
|
||||
x: float
|
||||
y: float
|
||||
net: str = ""
|
||||
|
||||
|
||||
class LayoutGraph(BaseModel):
|
||||
"""Parsed `.kicad_pcb` geometry. Optional; schema validation does not require it."""
|
||||
nets: dict[str, int] = {}
|
||||
footprints: dict[str, LayoutFootprint] = {}
|
||||
segments: list[LayoutSegment] = []
|
||||
vias: list[LayoutVia] = []
|
||||
|
||||
|
||||
@@ -1,8 +1,9 @@
|
||||
"""KiCad netlist (XML / s-expression) and single-sheet ``.kicad_sch`` parser.
|
||||
"""KiCad netlist (XML / s-expression) and ``.kicad_sch`` parser.
|
||||
|
||||
Yields the same ``(parts, nets)`` shape as PADS/EDIF so graph build is format-agnostic.
|
||||
``.kicad_sch`` uses embedded ``lib_symbols`` plus wires/labels; hierarchical
|
||||
sheets in other files are not followed (export a netlist for those).
|
||||
``.kicad_sch`` uses embedded ``lib_symbols`` plus wires/labels. Hierarchical
|
||||
``(sheet …)`` entries are followed from the root file (path-jailed under the
|
||||
project directory).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -10,6 +11,7 @@ from __future__ import annotations
|
||||
import math
|
||||
import re
|
||||
import xml.etree.ElementTree as ET
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterator
|
||||
|
||||
@@ -307,7 +309,30 @@ class _DSU:
|
||||
self.p[rb] = ra
|
||||
|
||||
|
||||
def parse_kicad_sch(tree: Any) -> tuple[dict[str, str], dict[str, list[tuple[str, str]]], dict[str, dict]]:
|
||||
_KIND_RANK = {"unnamed": 0, "local": 1, "hier": 2, "global": 3}
|
||||
|
||||
|
||||
@dataclass
|
||||
class _SchSheet:
|
||||
parts: dict[str, str]
|
||||
nets: dict[str, list[tuple[str, str]]]
|
||||
fields: dict[str, dict]
|
||||
net_scope: dict[str, str]
|
||||
sheetfiles: list[str] = field(default_factory=list)
|
||||
|
||||
|
||||
def _sheetfiles(tree: Any) -> list[str]:
|
||||
out: list[str] = []
|
||||
for sheet in _kids(tree, "sheet"):
|
||||
for p in _kids(sheet, "property"):
|
||||
if len(p) >= 3 and str(p[1]) == "Sheetfile":
|
||||
rel = str(p[2]).strip()
|
||||
if rel:
|
||||
out.append(rel)
|
||||
return out
|
||||
|
||||
|
||||
def _parse_kicad_sch_sheet(tree: Any) -> _SchSheet:
|
||||
lib_pins: dict[str, dict[tuple[int, str], tuple[float, float]]] = {}
|
||||
for sym in _kids(_kid(tree, "lib_symbols") or [], "symbol"):
|
||||
lid = str(sym[1]) if len(sym) > 1 else ""
|
||||
@@ -318,7 +343,7 @@ def parse_kicad_sch(tree: Any) -> tuple[dict[str, str], dict[str, list[tuple[str
|
||||
fields: dict[str, dict] = {}
|
||||
pin_at: dict[tuple[str, str], tuple[int, int]] = {}
|
||||
dsu = _DSU()
|
||||
labels: dict[tuple[int, int], str] = {}
|
||||
labels: dict[tuple[int, int], tuple[str, str]] = {}
|
||||
power_pts: list[tuple[tuple[int, int], str]] = []
|
||||
|
||||
def prop(sym: Any, key: str) -> str:
|
||||
@@ -327,6 +352,13 @@ def parse_kicad_sch(tree: Any) -> tuple[dict[str, str], dict[str, list[tuple[str
|
||||
return str(p[2])
|
||||
return ""
|
||||
|
||||
def set_label(pt: tuple[int, int], name: str, kind: str) -> None:
|
||||
if not name:
|
||||
return
|
||||
prev = labels.get(pt)
|
||||
if prev is None or _KIND_RANK[kind] >= _KIND_RANK[prev[1]]:
|
||||
labels[pt] = (name, kind)
|
||||
|
||||
for sym in _kids(tree, "symbol"):
|
||||
lib_id = _val(sym, "lib_id")
|
||||
ix, iy, rot = _at(sym)
|
||||
@@ -392,13 +424,34 @@ def parse_kicad_sch(tree: Any) -> tuple[dict[str, str], dict[str, list[tuple[str
|
||||
for a, b in zip(coords, coords[1:]):
|
||||
dsu.union(a, b)
|
||||
return
|
||||
if tag in {"label", "global_label", "hierarchical_label"}:
|
||||
if tag == "label":
|
||||
name = str(node[1]) if len(node) > 1 else ""
|
||||
x, y, _ = _at(node)
|
||||
pt = _snap(x, y)
|
||||
dsu.add(pt)
|
||||
if name:
|
||||
labels[pt] = name
|
||||
set_label(pt, name, "local")
|
||||
return
|
||||
if tag == "global_label":
|
||||
name = str(node[1]) if len(node) > 1 else ""
|
||||
x, y, _ = _at(node)
|
||||
pt = _snap(x, y)
|
||||
dsu.add(pt)
|
||||
set_label(pt, name, "global")
|
||||
return
|
||||
if tag == "hierarchical_label":
|
||||
name = str(node[1]) if len(node) > 1 else ""
|
||||
x, y, _ = _at(node)
|
||||
pt = _snap(x, y)
|
||||
dsu.add(pt)
|
||||
set_label(pt, name, "hier")
|
||||
return
|
||||
if tag == "sheet":
|
||||
for pin in _kids(node, "pin"):
|
||||
name = str(pin[1]) if len(pin) > 1 else ""
|
||||
x, y, _ = _at(pin)
|
||||
pt = _snap(x, y)
|
||||
dsu.add(pt)
|
||||
set_label(pt, name, "hier")
|
||||
return
|
||||
if tag == "junction":
|
||||
x, y, _ = _at(node)
|
||||
@@ -410,33 +463,143 @@ def parse_kicad_sch(tree: Any) -> tuple[dict[str, str], dict[str, list[tuple[str
|
||||
|
||||
collect_pts(tree)
|
||||
|
||||
for pt, name in labels.items():
|
||||
for pt in labels:
|
||||
dsu.add(pt)
|
||||
for pt, _name in power_pts:
|
||||
dsu.add(pt)
|
||||
|
||||
# Merge labels/power onto coinciding pin/wire points (already same snap keys).
|
||||
nets: dict[str, list[tuple[str, str]]] = {}
|
||||
root_name: dict[tuple[int, int], str] = {}
|
||||
for pt, name in labels.items():
|
||||
root_name[dsu.find(pt)] = name
|
||||
root_kind: dict[tuple[int, int], str] = {}
|
||||
for pt, (name, kind) in labels.items():
|
||||
r = dsu.find(pt)
|
||||
prev = root_kind.get(r, "unnamed")
|
||||
if _KIND_RANK[kind] >= _KIND_RANK[prev]:
|
||||
root_name[r] = name
|
||||
root_kind[r] = kind
|
||||
for pt, name in power_pts:
|
||||
root_name.setdefault(dsu.find(pt), name)
|
||||
r = dsu.find(pt)
|
||||
prev = root_kind.get(r, "unnamed")
|
||||
if _KIND_RANK["global"] >= _KIND_RANK[prev]:
|
||||
root_name[r] = name
|
||||
root_kind[r] = "global"
|
||||
|
||||
grouped: dict[tuple[int, int], list[tuple[str, str]]] = {}
|
||||
for (ref, pin), pt in pin_at.items():
|
||||
grouped.setdefault(dsu.find(pt), []).append((ref, pin))
|
||||
|
||||
nets: dict[str, list[tuple[str, str]]] = {}
|
||||
net_scope: dict[str, str] = {}
|
||||
used_names: set[str] = set()
|
||||
for root, pins in grouped.items():
|
||||
name = root_name.get(root)
|
||||
kind = root_kind.get(root, "unnamed")
|
||||
if not name:
|
||||
ref0, pin0 = pins[0]
|
||||
name = f"Net-({ref0}-Pad{pin0})"
|
||||
kind = "unnamed"
|
||||
while name in used_names:
|
||||
name = name + "_"
|
||||
used_names.add(name)
|
||||
nets[name] = pins
|
||||
net_scope[name] = kind
|
||||
|
||||
return _SchSheet(
|
||||
parts=parts,
|
||||
nets=nets,
|
||||
fields=fields,
|
||||
net_scope=net_scope,
|
||||
sheetfiles=_sheetfiles(tree),
|
||||
)
|
||||
|
||||
|
||||
def parse_kicad_sch(tree: Any) -> tuple[dict[str, str], dict[str, list[tuple[str, str]]], dict[str, dict]]:
|
||||
sheet = _parse_kicad_sch_sheet(tree)
|
||||
return sheet.parts, sheet.nets, sheet.fields
|
||||
|
||||
|
||||
def _uniq_pins(pins: list[tuple[str, str]]) -> list[tuple[str, str]]:
|
||||
seen: set[tuple[str, str]] = set()
|
||||
out: list[tuple[str, str]] = []
|
||||
for p in pins:
|
||||
if p not in seen:
|
||||
seen.add(p)
|
||||
out.append(p)
|
||||
return out
|
||||
|
||||
|
||||
def _safe_sheetfile(parent: Path, rel: str, project_root: Path) -> Path:
|
||||
rel_norm = rel.replace("\\", "/").strip()
|
||||
if not rel_norm or rel_norm.startswith("/") or ".." in Path(rel_norm).parts:
|
||||
raise ValueError(f"Sheetfile path rejected: {rel}")
|
||||
child = (parent.parent / rel_norm).resolve()
|
||||
root = project_root.resolve()
|
||||
try:
|
||||
child.relative_to(root)
|
||||
except ValueError:
|
||||
raise ValueError(f"Sheetfile path rejected: {rel}") from None
|
||||
return child
|
||||
|
||||
|
||||
def parse_kicad_sch_project(
|
||||
root_path: str | Path,
|
||||
) -> tuple[dict[str, str], dict[str, list[tuple[str, str]]], dict[str, dict]]:
|
||||
root = Path(root_path).resolve()
|
||||
project_root = root.parent
|
||||
seen: set[Path] = set()
|
||||
loaded: list[tuple[Path, _SchSheet]] = []
|
||||
|
||||
def visit(path: Path) -> None:
|
||||
path = path.resolve()
|
||||
if path in seen:
|
||||
raise ValueError(f"Cyclic sheet include: {path.name}")
|
||||
if not path.is_file():
|
||||
raise ValueError(f"Missing sheet file: {path.name}")
|
||||
seen.add(path)
|
||||
text = path.read_text(encoding="utf-8", errors="replace")
|
||||
tree = _parse_sexp(text)
|
||||
if _tag(tree) != "kicad_sch":
|
||||
raise ValueError(f"Expected kicad_sch in {path.name}, got {_tag(tree)!r}")
|
||||
sheet = _parse_kicad_sch_sheet(tree)
|
||||
loaded.append((path, sheet))
|
||||
for rel in sheet.sheetfiles:
|
||||
child = _safe_sheetfile(path, rel, project_root)
|
||||
visit(child)
|
||||
|
||||
visit(root)
|
||||
|
||||
parts: dict[str, str] = {}
|
||||
fields: dict[str, dict] = {}
|
||||
global_nets: dict[str, list[tuple[str, str]]] = {}
|
||||
hier_nets: dict[str, list[tuple[str, str]]] = {}
|
||||
local_nets: dict[str, list[tuple[str, str]]] = {}
|
||||
multi = len(loaded) > 1
|
||||
|
||||
for path, sheet in loaded:
|
||||
for ref, fp in sheet.parts.items():
|
||||
if ref in parts:
|
||||
raise ValueError(f"Duplicate reference {ref} in {path.name}")
|
||||
parts[ref] = fp
|
||||
fields[ref] = sheet.fields.get(ref, {})
|
||||
for name, pins in sheet.nets.items():
|
||||
scope = sheet.net_scope.get(name, "unnamed")
|
||||
if scope == "global":
|
||||
global_nets[name] = _uniq_pins(global_nets.get(name, []) + pins)
|
||||
elif scope == "hier":
|
||||
hier_nets[name] = _uniq_pins(hier_nets.get(name, []) + pins)
|
||||
else:
|
||||
out_name = f"{path.stem}/{name}" if multi else name
|
||||
local_nets[out_name] = _uniq_pins(local_nets.get(out_name, []) + pins)
|
||||
|
||||
nets: dict[str, list[tuple[str, str]]] = {}
|
||||
for name, pins in global_nets.items():
|
||||
nets[name] = pins
|
||||
for name, pins in hier_nets.items():
|
||||
nets[name] = _uniq_pins(nets.get(name, []) + pins)
|
||||
for name, pins in local_nets.items():
|
||||
out = name
|
||||
while out in nets:
|
||||
out = out + "_"
|
||||
nets[out] = pins
|
||||
|
||||
return parts, nets, fields
|
||||
|
||||
@@ -461,7 +624,7 @@ def parse_kicad(
|
||||
tree = _parse_sexp(text)
|
||||
tag = _tag(tree)
|
||||
if tag == "kicad_sch":
|
||||
parts, nets, fields = parse_kicad_sch(tree)
|
||||
parts, nets, fields = parse_kicad_sch_project(p)
|
||||
elif tag == "export":
|
||||
parts, nets, fields = parse_kicad_sexp_netlist(tree)
|
||||
else:
|
||||
|
||||
@@ -0,0 +1,131 @@
|
||||
"""KiCad `.kicad_pcb` ingest — footprints, pads, nets, segments, vias.
|
||||
|
||||
No SI/DRC. Schema validation stays complete without this file.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from backend.pinscopex.models import (
|
||||
LayoutFootprint,
|
||||
LayoutGraph,
|
||||
LayoutPad,
|
||||
LayoutSegment,
|
||||
LayoutVia,
|
||||
)
|
||||
from backend.pinscopex.parsers_kicad import (
|
||||
_at,
|
||||
_fnum,
|
||||
_kid,
|
||||
_kids,
|
||||
_parse_sexp,
|
||||
_rotate,
|
||||
_tag,
|
||||
_val,
|
||||
)
|
||||
|
||||
|
||||
def _xy(node: object, name: str) -> tuple[float, float]:
|
||||
k = _kid(node, name)
|
||||
if not k or len(k) < 3:
|
||||
return 0.0, 0.0
|
||||
return _fnum(k[1]), _fnum(k[2])
|
||||
|
||||
|
||||
def _prop(node: object, key: str) -> str:
|
||||
for p in _kids(node, "property"):
|
||||
if len(p) >= 3 and str(p[1]) == key:
|
||||
return str(p[2])
|
||||
return ""
|
||||
|
||||
|
||||
def _pad_net(pad: object) -> str:
|
||||
n = _kid(pad, "net")
|
||||
if n and len(n) >= 3:
|
||||
return str(n[2])
|
||||
return ""
|
||||
|
||||
|
||||
def parse_kicad_pcb(path: str | Path) -> LayoutGraph:
|
||||
p = Path(path)
|
||||
tree = _parse_sexp(p.read_text(encoding="utf-8", errors="replace"))
|
||||
if _tag(tree) != "kicad_pcb":
|
||||
raise ValueError(f"Expected kicad_pcb, got {_tag(tree)!r}")
|
||||
|
||||
nets: dict[str, int] = {}
|
||||
footprints: dict[str, LayoutFootprint] = {}
|
||||
segments: list[LayoutSegment] = []
|
||||
vias: list[LayoutVia] = []
|
||||
|
||||
for node in tree[1:]:
|
||||
if not isinstance(node, list) or not node:
|
||||
continue
|
||||
tag = _tag(node)
|
||||
if tag == "net" and len(node) >= 3 and not any(isinstance(x, list) and x and x[0] == "node" for x in node[1:]):
|
||||
try:
|
||||
code = int(_fnum(node[1]))
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
name = str(node[2])
|
||||
if name:
|
||||
nets[name] = code
|
||||
continue
|
||||
if tag in {"footprint", "module"}:
|
||||
fp_name = str(node[1]) if len(node) > 1 and not isinstance(node[1], list) else ""
|
||||
fx, fy, frot = _at(node)
|
||||
layer = _val(node, "layer")
|
||||
ref = _prop(node, "Reference")
|
||||
if not ref or ref.startswith("#"):
|
||||
continue
|
||||
pads: list[LayoutPad] = []
|
||||
for pad in _kids(node, "pad"):
|
||||
num = str(pad[1]) if len(pad) > 1 else ""
|
||||
if not num:
|
||||
continue
|
||||
px, py, _ = _at(pad)
|
||||
rx, ry = _rotate(px, py, frot)
|
||||
pads.append(LayoutPad(
|
||||
number=num,
|
||||
x=fx + rx,
|
||||
y=fy + ry,
|
||||
net=_pad_net(pad),
|
||||
))
|
||||
footprints[ref] = LayoutFootprint(
|
||||
reference=ref,
|
||||
footprint=fp_name,
|
||||
x=fx,
|
||||
y=fy,
|
||||
layer=layer,
|
||||
pads=pads,
|
||||
)
|
||||
continue
|
||||
if tag == "segment":
|
||||
net_el = _kid(node, "net")
|
||||
net_name = ""
|
||||
if net_el and len(net_el) >= 2:
|
||||
code = int(_fnum(net_el[1]))
|
||||
net_name = next((n for n, c in nets.items() if c == code), str(code))
|
||||
segments.append(LayoutSegment(
|
||||
start=_xy(node, "start"),
|
||||
end=_xy(node, "end"),
|
||||
width=_fnum(_val(node, "width") or 0),
|
||||
layer=_val(node, "layer"),
|
||||
net=net_name,
|
||||
))
|
||||
continue
|
||||
if tag == "via":
|
||||
net_el = _kid(node, "net")
|
||||
net_name = ""
|
||||
if net_el and len(net_el) >= 2:
|
||||
code = int(_fnum(net_el[1]))
|
||||
net_name = next((n for n, c in nets.items() if c == code), str(code))
|
||||
vx, vy, _ = _at(node)
|
||||
vias.append(LayoutVia(x=vx, y=vy, net=net_name))
|
||||
|
||||
return LayoutGraph(
|
||||
nets=nets,
|
||||
footprints=footprints,
|
||||
segments=segments,
|
||||
vias=vias,
|
||||
)
|
||||
@@ -84,6 +84,9 @@ def check_supply_decoupling(
|
||||
f"near {ref}."
|
||||
),
|
||||
reference="netlist topology",
|
||||
net=net_name,
|
||||
pins=[f"{ref}.{pin_num}"],
|
||||
rule_id="PS-DEC-001",
|
||||
))
|
||||
return findings
|
||||
|
||||
@@ -133,6 +136,9 @@ def check_i2c_pullups(
|
||||
f"to the I2C I/O rail."
|
||||
),
|
||||
reference="netlist topology",
|
||||
net=net_name,
|
||||
pins=[f"{ref}.{pin_num}"],
|
||||
rule_id="PS-I2C-001",
|
||||
))
|
||||
return findings
|
||||
|
||||
@@ -184,6 +190,9 @@ def check_reset_pullups(
|
||||
f"from a reset supervisor / GPIO."
|
||||
),
|
||||
reference="netlist topology",
|
||||
net=net_name,
|
||||
pins=[f"{ref}.{pin_num}"],
|
||||
rule_id="PS-RST-001",
|
||||
))
|
||||
return findings
|
||||
|
||||
|
||||
@@ -169,4 +169,7 @@ def _feasibility_finding(
|
||||
),
|
||||
recommendation=rec,
|
||||
reference=f"{mpn or ref} alternate-function table",
|
||||
net=net_name,
|
||||
pins=[f"{ref}.{pin_num}"],
|
||||
rule_id="PS-MUX-001",
|
||||
)
|
||||
|
||||
@@ -520,6 +520,37 @@ async def upload_netlist(project_id: str, file: UploadFile, request: Request):
|
||||
}
|
||||
|
||||
|
||||
@router.post("/projects/{project_id}/upload/pcb")
|
||||
async def upload_pcb(project_id: str, file: UploadFile, request: Request):
|
||||
storage = get_storage(request)
|
||||
result = proj_svc.resolve_project_access(storage, get_user_id(request), project_id)
|
||||
if not result:
|
||||
raise HTTPException(404, "Project not found")
|
||||
user_id = result[0]
|
||||
data = await file.read()
|
||||
if len(data) > MAX_UPLOAD_BYTES:
|
||||
raise HTTPException(413, f"File too large (max {MAX_UPLOAD_BYTES // 1024 // 1024} MB)")
|
||||
import tempfile, os
|
||||
from backend.pinscopex.parsers_kicad_pcb import parse_kicad_pcb
|
||||
|
||||
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".kicad_pcb")
|
||||
try:
|
||||
tmp.write(data)
|
||||
tmp.close()
|
||||
layout = parse_kicad_pcb(tmp.name)
|
||||
except Exception as e:
|
||||
raise HTTPException(400, f"Invalid KiCad PCB: {e}")
|
||||
finally:
|
||||
os.unlink(tmp.name)
|
||||
key = proj_svc.save_pcb(storage, user_id, project_id, data)
|
||||
return {
|
||||
"path": key,
|
||||
"footprints": len(layout.footprints),
|
||||
"nets": len(layout.nets),
|
||||
"segments": len(layout.segments),
|
||||
}
|
||||
|
||||
|
||||
def _build_designator_pins(
|
||||
parts: dict[str, str],
|
||||
nets: dict[str, list[tuple[str, str]]],
|
||||
|
||||
@@ -247,23 +247,16 @@ def _build_deduped(
|
||||
new_why = "Unverified: " + new_why
|
||||
|
||||
try:
|
||||
result.append(Finding(
|
||||
finding_id=canon.finding_id,
|
||||
designator=canon.designator,
|
||||
mpn=canon.mpn,
|
||||
aspect=canon.aspect,
|
||||
finding=str(group.get("finding") or canon.finding),
|
||||
why=new_why,
|
||||
source_page=group.get("source_page", canon.source_page),
|
||||
source_quote=canon.source_quote,
|
||||
source_designator=canon.source_designator,
|
||||
status=final_status,
|
||||
recommendation=str(
|
||||
result.append(canon.model_copy(update={
|
||||
"finding": str(group.get("finding") or canon.finding),
|
||||
"why": new_why,
|
||||
"source_page": group.get("source_page", canon.source_page),
|
||||
"status": final_status,
|
||||
"recommendation": str(
|
||||
group.get("recommendation") or canon.recommendation
|
||||
),
|
||||
reference=str(group.get("reference") or canon.reference),
|
||||
source=canon.source,
|
||||
))
|
||||
"reference": str(group.get("reference") or canon.reference),
|
||||
}))
|
||||
except Exception:
|
||||
log.exception("dedupe: failed to build merged Finding")
|
||||
return None
|
||||
|
||||
@@ -394,20 +394,15 @@ def _build_normalized(
|
||||
new_why = "Unverified: " + new_why
|
||||
|
||||
try:
|
||||
result.append(Finding(
|
||||
finding_id=canon.finding_id,
|
||||
designator=canon.designator,
|
||||
mpn=canon.mpn,
|
||||
aspect=canon.aspect,
|
||||
finding=str(entry.get("finding") or canon.finding),
|
||||
why=new_why,
|
||||
source_page=entry.get("source_page", canon.source_page),
|
||||
source_quote=str(entry.get("source_quote") or canon.source_quote),
|
||||
source_designator=canon.source_designator,
|
||||
status=final_status,
|
||||
recommendation=str(entry.get("recommendation") or canon.recommendation),
|
||||
reference=str(entry.get("reference") or canon.reference),
|
||||
))
|
||||
result.append(canon.model_copy(update={
|
||||
"finding": str(entry.get("finding") or canon.finding),
|
||||
"why": new_why,
|
||||
"source_page": entry.get("source_page", canon.source_page),
|
||||
"source_quote": str(entry.get("source_quote") or canon.source_quote),
|
||||
"status": final_status,
|
||||
"recommendation": str(entry.get("recommendation") or canon.recommendation),
|
||||
"reference": str(entry.get("reference") or canon.reference),
|
||||
}))
|
||||
except Exception:
|
||||
log.exception("normalize: failed to build merged Finding")
|
||||
return None
|
||||
|
||||
@@ -1496,6 +1496,25 @@ async def _stage_passive_extraction(ctx: PipelineContext) -> None:
|
||||
{"stage": "passive_extraction", "status": "complete"})
|
||||
|
||||
|
||||
def _write_layout_graph(ws: PipelineWorkspace, project_id: str) -> None:
|
||||
"""Parse optional `.kicad_pcb`. Fail-soft — schema validation must still run."""
|
||||
pcb = ws.local_path("uploads/pcb.kicad_pcb")
|
||||
if not pcb.is_file():
|
||||
return
|
||||
try:
|
||||
from backend.pinscopex.parsers_kicad_pcb import parse_kicad_pcb
|
||||
|
||||
layout = parse_kicad_pcb(pcb)
|
||||
out = ws.local_path("layout_graph.json")
|
||||
out.write_text(layout.model_dump_json(indent=2) + "\n")
|
||||
logger.info(
|
||||
"layout_graph: %s footprints, %s nets for %s",
|
||||
len(layout.footprints), len(layout.nets), project_id,
|
||||
)
|
||||
except Exception:
|
||||
logger.exception("kicad_pcb parse failed — continuing without layout")
|
||||
|
||||
|
||||
async def _stage_graph_build(ctx: PipelineContext) -> None:
|
||||
"""Stage 4 — Build the design graph from netlist, BOM, and extracted data."""
|
||||
broker.publish(ctx.project_id, "step_update",
|
||||
@@ -1525,6 +1544,7 @@ async def _stage_graph_build(ctx: PipelineContext) -> None:
|
||||
|
||||
graph_path = ctx.ws.local_path("design_graph.json")
|
||||
graph_path.write_text(ctx.graph.model_dump_json(indent=2) + "\n")
|
||||
_write_layout_graph(ctx.ws, ctx.project_id)
|
||||
|
||||
broker.publish(ctx.project_id, "step_update",
|
||||
{"stage": "graph_build", "status": "complete",
|
||||
@@ -2088,6 +2108,7 @@ async def run_regen_pipeline(
|
||||
|
||||
graph_path = ws.local_path("design_graph.json")
|
||||
graph_path.write_text(graph.model_dump_json(indent=2) + "\n")
|
||||
_write_layout_graph(ws, project_id)
|
||||
|
||||
broker.publish(project_id, "step_update",
|
||||
{"stage": "graph_build", "status": "complete",
|
||||
|
||||
@@ -5,6 +5,7 @@ Each project lives at users/{user_id}/projects/{id}/ with:
|
||||
uploads/bom.csv — uploaded BOM
|
||||
uploads/netlist.asc — uploaded netlist
|
||||
uploads/datasheets/*.pdf — uploaded datasheets
|
||||
uploads/pcb.kicad_pcb — optional KiCad board (layout checks)
|
||||
extracted/ — IC extraction output
|
||||
patterns/ — passive patterns
|
||||
models/ — cached component specs
|
||||
@@ -82,6 +83,7 @@ class ProjectMeta(BaseModel):
|
||||
updated: str = ""
|
||||
has_bom: bool = False
|
||||
has_netlist: bool = False
|
||||
has_pcb: bool = False
|
||||
# "pads" | "edif" | None — None for legacy projects (pre-EDIF-support).
|
||||
# Legacy reads fall back to looking for netlist.asc on disk.
|
||||
netlist_format: str | None = None
|
||||
@@ -658,6 +660,15 @@ def save_netlist(
|
||||
return key
|
||||
|
||||
|
||||
def save_pcb(
|
||||
storage: StorageBackend, user_id: str, project_id: str, data: bytes
|
||||
) -> str:
|
||||
key = f"{_project_prefix(user_id, project_id)}/uploads/pcb.kicad_pcb"
|
||||
storage.write_bytes(key, data)
|
||||
update_project(storage, user_id, project_id, has_pcb=True)
|
||||
return key
|
||||
|
||||
|
||||
def save_datasheet(
|
||||
storage: StorageBackend, user_id: str, project_id: str, mpn: str, data: bytes
|
||||
) -> str:
|
||||
|
||||
@@ -49,6 +49,7 @@ from backend.pinscopex.passive_rail_check import (
|
||||
check_reset_pullups,
|
||||
check_supply_decoupling,
|
||||
)
|
||||
from backend.pinscopex.bom_match_check import check_bom_schematic_match
|
||||
|
||||
TRACE_VERSION = 1
|
||||
|
||||
@@ -70,6 +71,9 @@ def _run_deterministic_checks(
|
||||
("supply_decoupling_check", lambda: check_supply_decoupling(graph, constraints_map)),
|
||||
("i2c_pullup_check", lambda: check_i2c_pullups(graph, constraints_map)),
|
||||
("reset_pullup_check", lambda: check_reset_pullups(graph, constraints_map)),
|
||||
("bom_match_check", lambda: check_bom_schematic_match(
|
||||
graph.schematic_fields, graph.bom_fields,
|
||||
)),
|
||||
):
|
||||
try:
|
||||
out.extend(fn())
|
||||
|
||||
Reference in New Issue
Block a user