Each layout_rules SI requirement (diff/SE Z, skew, max length, spacing, ref plane, vias, layer, return, series R) is one PASS/FAIL/MARGIN finding with FACT and REQUIREMENT. USB/HDMI/PCIe/ETH/LVDS/DDR only; I2C GPIO CC and analog REGN are skipped. No invented 90 Ω.
367 lines
15 KiB
Python
367 lines
15 KiB
Python
"""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
|
||
|
||
import logging
|
||
from datetime import datetime, timezone
|
||
from typing import Any, Iterable, Literal
|
||
|
||
from pydantic import BaseModel
|
||
|
||
from backend.periscopex.models import Finding
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
_STATUS_ORDER = {"ERROR": 0, "WARNING": 1, "INFO": 2}
|
||
_CLASS_ORDER = {"RULE": 0, "RISK": 1, "REVIEW": 2, "INFO": 3}
|
||
|
||
Provenance = Literal["MANDATORY", "RECOMMENDED", "TYPICAL", "EXAMPLE"]
|
||
FindingClass = Literal["RULE", "RISK", "REVIEW", "INFO"]
|
||
EvidenceStatus = Literal["SUFFICIENT", "INSUFFICIENT"]
|
||
|
||
_DEFAULT_ACTION = (
|
||
"Review this finding against the datasheet and the board, then change "
|
||
"the design if it applies."
|
||
)
|
||
_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-I2C-001", "I2C bus requires pull-ups."),
|
||
("PE-RST-001", "Reset pin requires the specified pull."),
|
||
("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, req in (
|
||
("PE-DEC-002", "Decoupling capacitor value vs datasheet (recommended)."),
|
||
("PE-I2C-002", "I2C pull-up value vs datasheet (recommended)."),
|
||
("PE-RST-002", "Reset pull value vs datasheet (recommended)."),
|
||
):
|
||
_add(rid, "RECOMMENDED", "REVIEW", 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", "REVIEW", "Filter topology vs datasheet."),
|
||
("PE-FLT-002", "RECOMMENDED", "REVIEW", "Filter component vs datasheet."),
|
||
("PE-FLT-003", "RECOMMENDED", "REVIEW", "Filter cutoff vs datasheet."),
|
||
("PE-XTAL-001", "MANDATORY", "RULE", "Crystal load capacitance vs Cl."),
|
||
("PE-XTAL-002", "RECOMMENDED", "REVIEW", "Crystal layout / load-cap note."),
|
||
("PE-XTAL-003", "RECOMMENDED", "REVIEW", "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-SI-001", "Differential skew vs length_match mm."),
|
||
):
|
||
_add(rid, "MANDATORY", "RULE", domain="pcb", requirement=req)
|
||
_add("PE-SI-002", "RECOMMENDED", "RISK", domain="pcb",
|
||
requirement="Diff/SE Z0 vs datasheet window (ImpedenceFinder avg/min/max).")
|
||
_add("PE-SI-003", "RECOMMENDED", "RISK", domain="pcb",
|
||
requirement="Max routed length vs datasheet millimetres.")
|
||
_add("PE-SI-004", "RECOMMENDED", "RISK", domain="pcb",
|
||
requirement="Intra-pair spacing vs datasheet millimetres.")
|
||
_add("PE-SI-005", "RECOMMENDED", "RISK", domain="pcb",
|
||
requirement="Reference plane / topology vs datasheet.")
|
||
_add("PE-SI-006", "RECOMMENDED", "RISK", domain="pcb",
|
||
requirement="HS via count vs datasheet min/max.")
|
||
_add("PE-SI-008", "RECOMMENDED", "REVIEW", domain="pcb",
|
||
requirement="Return via next to an impedance-controlled pair.")
|
||
_add("PE-SI-009", "RECOMMENDED", "RISK", domain="pcb",
|
||
requirement="Series resistor on the HS net vs datasheet ohms.")
|
||
_add("PE-SI-010", "TYPICAL", "INFO", domain="pcb",
|
||
requirement="HS bus measured; library has no SI FACT — not 90 Ω folklore.")
|
||
_add("PE-PLC-004", "RECOMMENDED", "REVIEW", domain="pcb",
|
||
requirement="Keepout from layout_rules — not a DRC for pad copper.")
|
||
_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.")
|
||
_add("PE-HIER-001", "TYPICAL", "INFO", domain="pcb",
|
||
requirement="Every component pin should land on a named net.")
|
||
_add("PE-DRT-002", "RECOMMENDED", "RISK", domain="pcb",
|
||
requirement="Vop/Vrated utilization band (not an invented dielectric %).")
|
||
_add("PE-DRT-003", "TYPICAL", "INFO", domain="pcb",
|
||
requirement="Capacitors with Vop and Vrated at or below 80% utilization.")
|
||
_add("PE-TIM-001", "MANDATORY", "RULE", domain="pcb",
|
||
requirement="Reset RC delay vs datasheet t_reset when both numbers exist.")
|
||
_add("PE-TIM-002", "MANDATORY", "RULE", domain="pcb",
|
||
requirement="Strap divider voltage vs Vih/Vil when both numbers exist.")
|
||
_add("PE-PI-001", "RECOMMENDED", "RISK", domain="pcb",
|
||
requirement="IC supply net should have a local decoupling capacitor.")
|
||
_add("PE-PI-002", "RECOMMENDED", "REVIEW", domain="pcb",
|
||
requirement="Local vs bulk on a supply when capacitance values exist.")
|
||
_add("PE-ESD-001", "RECOMMENDED", "REVIEW", domain="pcb",
|
||
requirement="Connector/USB net ESD — REVIEW unless a mandatory FACT exists.")
|
||
_add("PE-RET-001", "TYPICAL", "REVIEW", domain="pcb",
|
||
requirement="High-speed net over a GND pour should have a nearby GND via.")
|
||
|
||
|
||
_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 "").strip() or _DEFAULT_ACTION
|
||
if not (f.recommendation or "").strip():
|
||
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"
|
||
|
||
# ERROR only for a measured MANDATORY RULE. REVIEW/RISK/INFO stay ≤ WARNING.
|
||
if f.status == "ERROR":
|
||
if f.finding_class != "RULE" or prov != "MANDATORY":
|
||
f.status = "WARNING"
|
||
|
||
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
|
||
if not (f.action or "").strip():
|
||
f.action = _DEFAULT_ACTION
|
||
if not (f.recommendation or "").strip():
|
||
f.recommendation = f.action
|
||
return f
|
||
|
||
|
||
def sort_findings(findings: list[Finding]) -> list[Finding]:
|
||
"""ERROR then WARNING then INFO; within that RULE > RISK > REVIEW > INFO."""
|
||
findings.sort(
|
||
key=lambda f: (
|
||
_STATUS_ORDER.get(f.status or "INFO", 9),
|
||
_CLASS_ORDER.get(f.finding_class or "INFO", 9),
|
||
f.designator or "",
|
||
f.rule_id or "",
|
||
f.finding or "",
|
||
)
|
||
)
|
||
return findings
|
||
|
||
|
||
def complete_findings(findings: list[Finding]) -> None:
|
||
for f in findings:
|
||
try:
|
||
complete_finding(f)
|
||
except Exception:
|
||
log.exception(
|
||
"complete_finding failed for %s/%s",
|
||
getattr(f, "designator", "?"),
|
||
getattr(f, "rule_id", None),
|
||
)
|
||
sort_findings(findings)
|
||
|
||
|
||
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
|