Apply the finding engine to schematic and PCB reviews.
FACT, REQUIREMENT, and INFERENCE are separate fields; datasheet provenance lives in the rule DB so Recommended cannot stay ERROR; LLM output is REVIEW. Designer decisions persist across rescans. Schematic and PCB share one library and the same finding object in the report UI.
This commit is contained in:
@@ -0,0 +1,288 @@
|
||||
"""Shared finding engine for schematic and PCB.
|
||||
|
||||
FACT / REQUIREMENT / INFERENCE stay in separate fields. Datasheet
|
||||
provenance lives in the rule catalog, not in LLM prose. Recommended
|
||||
never becomes ERROR. LLM output is REVIEW, never RULE.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, Iterable, Literal
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from backend.periscopex.models import Finding
|
||||
|
||||
Provenance = Literal["MANDATORY", "RECOMMENDED", "TYPICAL", "EXAMPLE"]
|
||||
FindingClass = Literal["RULE", "RISK", "REVIEW", "INFO"]
|
||||
EvidenceStatus = Literal["SUFFICIENT", "INSUFFICIENT"]
|
||||
|
||||
_LLM_SOURCES = frozenset({None, "review", "pcb_review"})
|
||||
_SOFT_PROVENANCE = frozenset({"RECOMMENDED", "TYPICAL", "EXAMPLE"})
|
||||
|
||||
|
||||
class RuleRecord(BaseModel):
|
||||
"""One row in the rule DB (not free-text from the model)."""
|
||||
rule_id: str
|
||||
provenance: Provenance
|
||||
finding_class: FindingClass
|
||||
domain: Literal["schema", "pcb", "shared"] = "shared"
|
||||
source: str | None = None
|
||||
requirement: str = ""
|
||||
|
||||
|
||||
class DesignerDecision(BaseModel):
|
||||
"""Stored intent so a rescan does not re-nag the same FACT."""
|
||||
decision_id: str
|
||||
fingerprint: str
|
||||
rule_id: str | None = None
|
||||
designator: str = ""
|
||||
net: str | None = None
|
||||
aspect: str | None = None
|
||||
intent: str = ""
|
||||
reason: str = ""
|
||||
user_id: str = ""
|
||||
created_at: str = ""
|
||||
|
||||
|
||||
# rule_id → catalog. Optional override: (rule_id, source).
|
||||
_RULES: dict[str, RuleRecord] = {}
|
||||
_RULES_BY_SOURCE: dict[tuple[str, str], RuleRecord] = {}
|
||||
|
||||
|
||||
def _add(rule_id: str, provenance: Provenance, cls: FindingClass, *,
|
||||
domain: Literal["schema", "pcb", "shared"] = "shared",
|
||||
source: str | None = None, requirement: str = "") -> None:
|
||||
rec = RuleRecord(
|
||||
rule_id=rule_id, provenance=provenance, finding_class=cls,
|
||||
domain=domain, source=source, requirement=requirement,
|
||||
)
|
||||
_RULES[rule_id] = rec
|
||||
if source:
|
||||
_RULES_BY_SOURCE[(rule_id, source)] = rec
|
||||
|
||||
|
||||
def _seed() -> None:
|
||||
if _RULES:
|
||||
return
|
||||
# Schematic
|
||||
for rid, req in (
|
||||
("PE-MUX-001", "Pin function must match the mux / datasheet pin table."),
|
||||
("PE-NC-001", "NC pins must not be connected."),
|
||||
("PE-DEC-001", "Supply pin requires a decoupling capacitor on that net."),
|
||||
("PE-DEC-002", "Decoupling capacitor value vs datasheet."),
|
||||
("PE-I2C-001", "I2C bus requires pull-ups."),
|
||||
("PE-I2C-002", "I2C pull-up value vs datasheet."),
|
||||
("PE-RST-001", "Reset pin requires the specified pull."),
|
||||
("PE-RST-002", "Reset pull value vs datasheet."),
|
||||
("PE-BOM-001", "BOM MPN must match the netlist part."),
|
||||
("PE-BOM-002", "BOM line missing from the netlist."),
|
||||
("PE-DNP-001", "DNP / fitted state vs netlist."),
|
||||
("PE-DRT-001", "Operating voltage must not exceed the capacitor rating."),
|
||||
):
|
||||
_add(rid, "MANDATORY", "RULE", domain="schema", requirement=req)
|
||||
for rid in ("PE-TH-001", "PE-TH-003"):
|
||||
_add(rid, "TYPICAL", "RISK", domain="schema",
|
||||
requirement="Thermal estimate only with I_load and θJA from specs.")
|
||||
_add("PE-PWR-001", "MANDATORY", "RISK", domain="shared",
|
||||
source="power_margin_check",
|
||||
requirement="I_load vs regulator capability (not Iout_max).")
|
||||
_add("PE-PWR-001", "MANDATORY", "RULE", domain="pcb",
|
||||
source="pcb_power_thermal",
|
||||
requirement="Trace width × copper thickness vs I_load (IPC-2221 when ΔT known).")
|
||||
for rid, prov, cls, req in (
|
||||
("PE-SEQ-001", "RECOMMENDED", "RISK", "Power sequencing from datasheet notes."),
|
||||
("PE-FLT-001", "RECOMMENDED", "RISK", "Filter topology vs datasheet."),
|
||||
("PE-FLT-002", "RECOMMENDED", "RISK", "Filter component vs datasheet."),
|
||||
("PE-FLT-003", "RECOMMENDED", "RISK", "Filter cutoff vs datasheet."),
|
||||
("PE-XTAL-001", "MANDATORY", "RULE", "Crystal load capacitance vs Cl."),
|
||||
("PE-XTAL-002", "RECOMMENDED", "RISK", "Crystal layout / load-cap note."),
|
||||
("PE-XTAL-003", "RECOMMENDED", "RISK", "Crystal drive / Cl note."),
|
||||
("PE-LF-001", "TYPICAL", "INFO", "Lifecycle / NRND."),
|
||||
("PE-LF-002", "TYPICAL", "INFO", "Lifecycle / last-time-buy."),
|
||||
("PE-LF-003", "TYPICAL", "INFO", "Lifecycle / obsolete."),
|
||||
("PE-INT-001", "TYPICAL", "REVIEW", "Internal feature (ESD/pull-up) vs net."),
|
||||
("PE-ESR-001", "RECOMMENDED", "RISK", "HF / ESR coverage."),
|
||||
("PE-ERRATA-001", "MANDATORY", "REVIEW", "Published errata."),
|
||||
):
|
||||
_add(rid, prov, cls, domain="schema", requirement=req)
|
||||
# PCB geometry
|
||||
_add("PE-LAY-001", "MANDATORY", "RULE", domain="pcb",
|
||||
requirement="PCB pad net must match the schematic pin net.")
|
||||
_add("PE-LAY-002", "MANDATORY", "RULE", domain="pcb",
|
||||
requirement="Schematic ref must have a footprint (except #PWR).")
|
||||
_add("PE-LAY-003", "TYPICAL", "INFO", domain="pcb",
|
||||
requirement="KiCad /net vs flattened schematic net is the same net.")
|
||||
for rid, req in (
|
||||
("PE-PLC-001", "Decoupling proximity max_distance_mm from layout_rules."),
|
||||
("PE-PLC-002", "Thermal via min_via_count from layout_rules."),
|
||||
("PE-PLC-003", "same_layer decoupling from layout_rules."),
|
||||
("PE-PLC-004", "Keepout from layout_rules."),
|
||||
("PE-SI-001", "Differential skew vs length_match mm."),
|
||||
):
|
||||
_add(rid, "MANDATORY", "RULE", domain="pcb", requirement=req)
|
||||
_add("PE-VIA-001", "TYPICAL", "INFO", domain="pcb",
|
||||
requirement="Via current share from I_load and via count (no invented table).")
|
||||
_add("PE-THM-001", "RECOMMENDED", "RISK", domain="pcb",
|
||||
requirement="Dissipation vs courtyard pour/vias when P is known.")
|
||||
_add("PE-KEL-001", "RECOMMENDED", "RISK", domain="pcb",
|
||||
requirement="Kelvin/sense pin should not share the load net.")
|
||||
_add("PE-STCH-001", "TYPICAL", "INFO", domain="pcb",
|
||||
requirement="GND stitch via in the bbox of a signal over a GND pour.")
|
||||
|
||||
|
||||
_seed()
|
||||
|
||||
|
||||
def lookup_rule(rule_id: str | None, source: str | None = None) -> RuleRecord | None:
|
||||
if not rule_id:
|
||||
return None
|
||||
if source and (rule_id, source) in _RULES_BY_SOURCE:
|
||||
return _RULES_BY_SOURCE[(rule_id, source)]
|
||||
return _RULES.get(rule_id)
|
||||
|
||||
|
||||
def finding_fingerprint(f: Finding) -> str:
|
||||
return "|".join([
|
||||
f.rule_id or f.source or "",
|
||||
f.designator or "",
|
||||
f.net or "",
|
||||
f.aspect or "",
|
||||
])
|
||||
|
||||
|
||||
def complete_finding(f: Finding) -> Finding:
|
||||
"""Fill FACT/REQUIREMENT/INFERENCE, clamp provenance and evidence."""
|
||||
rec = lookup_rule(f.rule_id, f.source)
|
||||
if not (f.facts or "").strip():
|
||||
f.facts = f.finding
|
||||
if not (f.requirement or "").strip():
|
||||
f.requirement = (rec.requirement if rec and rec.requirement else None) or f.why
|
||||
if not (f.action or "").strip():
|
||||
f.action = f.recommendation or ""
|
||||
if not (f.recommendation or "").strip() and f.action:
|
||||
f.recommendation = f.action
|
||||
|
||||
llm = (f.source in {None, "review", "pcb_review"}) and not f.rule_id
|
||||
|
||||
if rec:
|
||||
if not f.provenance:
|
||||
f.provenance = rec.provenance
|
||||
if not f.finding_class:
|
||||
f.finding_class = rec.finding_class
|
||||
elif llm or f.source in {"review", "pcb_review"}:
|
||||
f.finding_class = f.finding_class or "REVIEW"
|
||||
elif f.source:
|
||||
f.finding_class = f.finding_class or "RULE"
|
||||
else:
|
||||
f.finding_class = f.finding_class or "REVIEW"
|
||||
|
||||
if f.source in {"review", "pcb_review"}:
|
||||
f.finding_class = "REVIEW"
|
||||
|
||||
quote = (f.source_quote or "").strip()
|
||||
if f.evidence_status == "INSUFFICIENT":
|
||||
pass
|
||||
elif llm and not quote:
|
||||
f.evidence_status = "INSUFFICIENT"
|
||||
elif not (f.facts or "").strip() or not (f.requirement or "").strip():
|
||||
f.evidence_status = "INSUFFICIENT"
|
||||
else:
|
||||
f.evidence_status = f.evidence_status or "SUFFICIENT"
|
||||
|
||||
if f.evidence_status == "INSUFFICIENT":
|
||||
if f.status == "ERROR":
|
||||
f.status = "WARNING" if quote or rec else "INFO"
|
||||
if f.finding_class == "RULE":
|
||||
f.finding_class = "INFO"
|
||||
if not (f.inference or "").strip():
|
||||
f.inference = (
|
||||
"Insufficient evidence — not inventing a violation or default "
|
||||
"(no 1 oz / 10 °C / 50 Ω / IEC)."
|
||||
)
|
||||
|
||||
prov = f.provenance or (rec.provenance if rec else None)
|
||||
if prov in _SOFT_PROVENANCE and f.status == "ERROR":
|
||||
f.status = "WARNING"
|
||||
if f.finding_class == "RULE":
|
||||
f.finding_class = "RISK"
|
||||
|
||||
if f.finding_class == "RULE" and f.confidence is None:
|
||||
f.confidence = 0.9 if f.evidence_status == "SUFFICIENT" else 0.3
|
||||
elif f.finding_class == "REVIEW" and f.confidence is None:
|
||||
f.confidence = 0.55 if quote else 0.35
|
||||
elif f.confidence is None:
|
||||
f.confidence = 0.5 if f.evidence_status == "SUFFICIENT" else 0.25
|
||||
|
||||
if not (f.why or "").strip():
|
||||
f.why = f.requirement
|
||||
if not (f.finding or "").strip():
|
||||
f.finding = f.facts
|
||||
return f
|
||||
|
||||
|
||||
def complete_findings(findings: list[Finding]) -> None:
|
||||
for f in findings:
|
||||
complete_finding(f)
|
||||
|
||||
|
||||
def apply_decisions(
|
||||
findings: list[Finding],
|
||||
decisions: Iterable[DesignerDecision] | Iterable[dict[str, Any]],
|
||||
) -> None:
|
||||
"""Rescan must not re-nag a stored designer decision as ERROR."""
|
||||
by_fp: dict[str, DesignerDecision] = {}
|
||||
for raw in decisions:
|
||||
d = raw if isinstance(raw, DesignerDecision) else DesignerDecision.model_validate(raw)
|
||||
by_fp[d.fingerprint] = d
|
||||
if not by_fp:
|
||||
return
|
||||
for f in findings:
|
||||
fp = finding_fingerprint(f)
|
||||
d = by_fp.get(fp)
|
||||
if not d:
|
||||
continue
|
||||
f.suppressed = True
|
||||
f.decision_id = d.decision_id
|
||||
f.status = "INFO"
|
||||
f.finding_class = "INFO"
|
||||
extra = f"Designer decision {d.decision_id}: {d.intent}"
|
||||
if d.reason:
|
||||
extra = f"{extra} — {d.reason}"
|
||||
if extra not in (f.inference or ""):
|
||||
f.inference = ((f.inference or "") + " " + extra).strip()
|
||||
if not (f.action or "").strip():
|
||||
f.action = "Keep the stored intent; no layout/schematic change required."
|
||||
f.recommendation = f.action
|
||||
|
||||
|
||||
def decision_from_review(
|
||||
finding: Finding,
|
||||
*,
|
||||
state: str,
|
||||
reason: str,
|
||||
user_id: str,
|
||||
) -> DesignerDecision | None:
|
||||
if state not in {"wontfix", "false_positive"}:
|
||||
return None
|
||||
ts = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")
|
||||
fp = finding_fingerprint(finding)
|
||||
return DesignerDecision(
|
||||
decision_id=f"DEC-{finding.designator}-{ts}",
|
||||
fingerprint=fp,
|
||||
rule_id=finding.rule_id,
|
||||
designator=finding.designator,
|
||||
net=finding.net,
|
||||
aspect=finding.aspect,
|
||||
intent=state,
|
||||
reason=reason,
|
||||
user_id=user_id,
|
||||
created_at=datetime.now(timezone.utc).isoformat(),
|
||||
)
|
||||
|
||||
|
||||
def upsert_decision(store: list[dict[str, Any]], decision: DesignerDecision) -> list[dict[str, Any]]:
|
||||
out = [row for row in store if row.get("fingerprint") != decision.fingerprint]
|
||||
out.append(decision.model_dump())
|
||||
return out
|
||||
@@ -356,6 +356,19 @@ class Finding(BaseModel):
|
||||
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
|
||||
# Finding engine (docs/motore-finding.md) — optional for legacy JSON.
|
||||
facts: str = ""
|
||||
requirement: str = ""
|
||||
inference: str = ""
|
||||
provenance: Literal["MANDATORY", "RECOMMENDED", "TYPICAL", "EXAMPLE"] | None = None
|
||||
finding_class: Literal["RULE", "RISK", "REVIEW", "INFO"] | None = None
|
||||
confidence: float | None = None
|
||||
evidence_status: Literal["SUFFICIENT", "INSUFFICIENT"] | None = None
|
||||
calculation: str = ""
|
||||
assumptions: list[str] = []
|
||||
action: str = ""
|
||||
decision_id: str | None = None
|
||||
suppressed: bool = False
|
||||
|
||||
|
||||
class ValidationReport(BaseModel):
|
||||
|
||||
@@ -5,7 +5,7 @@ from __future__ import annotations
|
||||
import logging
|
||||
from collections import Counter
|
||||
|
||||
from backend.periscopex.derating import build_derating_table
|
||||
from backend.periscopex.finding_engine import complete_findings
|
||||
from backend.periscopex.models import DesignGraph, Finding, LayoutGraph
|
||||
from backend.periscopex.pcb_net_match import check_pcb_net_match
|
||||
from backend.periscopex.pcb_power_thermal import (
|
||||
@@ -31,6 +31,7 @@ def assign_pcb_finding_ids(findings: list[Finding]) -> None:
|
||||
f.finding_id = f"PCB-{f.designator}-{counter[f.designator]:03d}"
|
||||
if not (f.recommendation or "").strip():
|
||||
f.recommendation = _FALLBACK_FIX
|
||||
complete_findings(findings)
|
||||
|
||||
|
||||
def check_pcb_derating(graph: DesignGraph) -> list[Finding]:
|
||||
|
||||
@@ -141,6 +141,9 @@ def check_pcb_power_traces(
|
||||
),
|
||||
source="pcb_power_thermal",
|
||||
rule_id="PE-PWR-001",
|
||||
evidence_status="INSUFFICIENT",
|
||||
calculation="",
|
||||
assumptions=["IPC-2221 not applied without copper thickness."],
|
||||
net=net,
|
||||
pins=[],
|
||||
))
|
||||
@@ -162,6 +165,8 @@ def check_pcb_power_traces(
|
||||
recommendation="Add Tjmax to the shared library extraction, then re-run PCB review.",
|
||||
source="pcb_power_thermal",
|
||||
rule_id="PE-PWR-001",
|
||||
evidence_status="INSUFFICIENT",
|
||||
assumptions=["ΔT is Tjmax − 25 °C from the datasheet, never a default 10 °C."],
|
||||
net=net,
|
||||
pins=[],
|
||||
))
|
||||
@@ -189,6 +194,10 @@ def check_pcb_power_traces(
|
||||
),
|
||||
source="pcb_power_thermal",
|
||||
rule_id="PE-PWR-001",
|
||||
calculation=(
|
||||
f"I = k·ΔT^0.44·A^0.725 → {i_allow:.3g} A; I_load={i_load:.3g} A"
|
||||
),
|
||||
evidence_status="SUFFICIENT",
|
||||
net=net,
|
||||
pins=[],
|
||||
))
|
||||
|
||||
@@ -30,6 +30,11 @@ are in the context. Never assume 1 oz or 50 Ω.
|
||||
- Creepage/clearance only if the extraction or IEC number is in context.
|
||||
|
||||
### Findings
|
||||
This is a **design review** of geometry vs the library JSON, not a second \
|
||||
datasheet exam. Put board measurements in `finding` (FACT), library text in \
|
||||
`why` (REQUIREMENT). Recommended layout notes are never ERROR. Missing \
|
||||
width/thickness/I_load → say insufficient evidence, do not assume 1 oz.
|
||||
|
||||
Every finding MUST have status ERROR, WARNING, or INFO and a non-empty \
|
||||
recommendation. Call submit_review. Empty findings with checked_areas is \
|
||||
valid when the layout matches the extraction.
|
||||
|
||||
@@ -20,6 +20,7 @@ from dotenv import load_dotenv
|
||||
|
||||
load_dotenv()
|
||||
|
||||
from backend.periscopex.finding_engine import complete_findings
|
||||
from backend.periscopex.models import (
|
||||
ComponentConstraints,
|
||||
ComponentType,
|
||||
@@ -149,6 +150,12 @@ neighbor's datasheet that you fetched via `get_datasheet_excerpt` — this \
|
||||
links the page number to the right datasheet.
|
||||
- **recommendation**: What to change (for ERROR/WARNING only).
|
||||
|
||||
This is a **design review**, not a design rule. Put observations in \
|
||||
`finding` (FACT), datasheet text in `why` (REQUIREMENT), and judgment \
|
||||
only there — do not invent millimetres, IEC numbers, or typical values. \
|
||||
Recommended datasheet notes are never ERROR. If evidence is missing, say \
|
||||
so (Unverified) instead of guessing.
|
||||
|
||||
### Calibration
|
||||
ERROR only for clear violations: required pin floating, voltage exceeding \
|
||||
absolute max, required external component completely missing, wrong \
|
||||
@@ -922,12 +929,19 @@ def _parse_review(
|
||||
mpn=mpn,
|
||||
source_designator=src_designator,
|
||||
finding=item["finding"],
|
||||
facts=str(item.get("finding") or ""),
|
||||
requirement=why,
|
||||
inference=str(item.get("inference") or ""),
|
||||
why=why,
|
||||
status=status,
|
||||
source_page=page,
|
||||
source_quote=item.get("source_quote", ""),
|
||||
recommendation=item.get("recommendation", ""),
|
||||
action=str(item.get("recommendation") or ""),
|
||||
reference=f"{src_mpn} datasheet p.{page if page is not None else '?'}",
|
||||
source="review",
|
||||
finding_class="REVIEW",
|
||||
evidence_status="SUFFICIENT" if quote else "INSUFFICIENT",
|
||||
))
|
||||
except (KeyError, TypeError, ValueError) as exc:
|
||||
print(f"Skipping malformed finding for {ic_ref}: {exc}", file=sys.stderr)
|
||||
@@ -937,11 +951,12 @@ def _parse_review(
|
||||
|
||||
|
||||
def assign_finding_ids(findings: list[Finding]) -> None:
|
||||
"""Assign finding_id: {designator}-{001}, {002}, ..."""
|
||||
"""Assign finding_id: {designator}-{001}, {002}, ... then run the finding engine."""
|
||||
counter: Counter[str] = Counter()
|
||||
for f in findings:
|
||||
counter[f.designator] += 1
|
||||
f.finding_id = f"{f.designator}-{counter[f.designator]:03d}"
|
||||
complete_findings(findings)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user