ERROR stays only for RULE+MANDATORY. Hierarchy walks component→pin→net→block. Derating is PASS/MARGIN/RISK from Vop/Vrated. Timing and PI skip without numbers. ESD and HS return path are REVIEW, not RULE ERROR.
323 lines
13 KiB
Python
323 lines
13 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
|
||
|
||
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"]
|
||
|
||
_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-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 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
|