michele
16c606ae3d
Parallel pcb_status pipeline: parse board, classify domains/groups, inventory traces, PE-LAY/PLC/SI/DRT checks, per-IC datasheet review. Findings merge into the report UI. No auto-place or pcbnew write-back.
82 lines
3.2 KiB
Python
82 lines
3.2 KiB
Python
"""Per-IC PCB datasheet review — same agentic loop as schematic, layout context."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
from backend.periscopex.cad_bridge import annotate_findings_cad, cad_index_from_graph
|
|
from backend.periscopex.functional_groups import FunctionalGroupsReport
|
|
from backend.periscopex.models import ComponentType, DesignGraph, Finding, LayoutGraph, ValidationReport
|
|
from backend.periscopex.pcb_inventory import PcbInventoryReport
|
|
from backend.periscopex.pcb_review import PCB_SYSTEM_PROMPT, build_pcb_layout_context
|
|
from backend.periscopex.validate import ReviewResult
|
|
from backend.services.api_logs import ApiLogger
|
|
from backend.services.validation import _find_pdf, review_ic_async
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
_FIX = "Adjust the layout to match the datasheet recommendation, then re-run PCB review."
|
|
|
|
|
|
def _ensure_recs(findings: list[Finding]) -> None:
|
|
for f in findings:
|
|
if f.source in (None, "review"):
|
|
f.source = "pcb_review"
|
|
if not (f.recommendation or "").strip():
|
|
f.recommendation = _FIX
|
|
|
|
|
|
async def review_pcb_ics(
|
|
graph: DesignGraph,
|
|
constraints_map: dict,
|
|
layout: LayoutGraph | None,
|
|
plan: FunctionalGroupsReport | None,
|
|
inventory: PcbInventoryReport | None,
|
|
pdf_dir: Path,
|
|
storage=None,
|
|
api_logger: ApiLogger | None = None,
|
|
on_progress=None,
|
|
) -> tuple[list[Finding], dict[str, list[str]], list[dict]]:
|
|
"""Review each IC with a datasheet. Fail-soft per IC. No auto-place."""
|
|
findings: list[Finding] = []
|
|
coverage: dict[str, list[str]] = {}
|
|
skipped: list[dict] = []
|
|
cache: dict = {}
|
|
for ref, comp in sorted(graph.components.items()):
|
|
if comp.component_type != ComponentType.IC:
|
|
continue
|
|
mpn = (comp.mpn or "").strip() or (comp.value or "").strip()
|
|
if not mpn:
|
|
skipped.append({"designator": ref, "reason": "no MPN in BOM"})
|
|
continue
|
|
pdf = _find_pdf(mpn, pdf_dir, storage=storage)
|
|
if pdf is None:
|
|
skipped.append({"designator": ref, "reason": "no datasheet PDF"})
|
|
continue
|
|
extra = build_pcb_layout_context(ref, graph, layout, plan, inventory)
|
|
try:
|
|
result, _trace = await review_ic_async(
|
|
graph, constraints_map, ref, str(pdf),
|
|
on_progress=on_progress,
|
|
api_logger=api_logger,
|
|
pdf_dir=pdf_dir,
|
|
storage=storage,
|
|
excerpt_cache=cache,
|
|
extra_context=extra,
|
|
system_prompt=PCB_SYSTEM_PROMPT,
|
|
log_stage="pcb_review",
|
|
)
|
|
except Exception:
|
|
log.exception("PCB AI review failed for %s — skipping", ref)
|
|
skipped.append({"designator": ref, "reason": "pcb_review error"})
|
|
continue
|
|
if not isinstance(result, ReviewResult):
|
|
continue
|
|
_ensure_recs(result.findings)
|
|
annotate_findings_cad(result.findings, cad_index_from_graph(graph))
|
|
findings.extend(result.findings)
|
|
if result.checked_areas:
|
|
coverage[ref] = list(result.checked_areas)
|
|
return findings, coverage, skipped
|