Share library/extracted for schematic and PCB review.

PCB AI skips a second PDF exam (pintable cache, pdf_path=None) and the report
gains a PCB exam section plus power-trace, thermal, via, Kelvin, and GND-stitch
checks grounded in board geometry and extracted specs.
This commit is contained in:
2026-09-20 08:32:23 +02:00
parent f3f96ea4d1
commit 8f4ebc645c
15 changed files with 1014 additions and 107 deletions
+16
View File
@@ -8,6 +8,13 @@ from collections import Counter
from backend.periscopex.derating import build_derating_table
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 (
check_pcb_gnd_stitch,
check_pcb_kelvin,
check_pcb_power_traces,
check_pcb_thermal_copper,
check_pcb_via_current,
)
from backend.periscopex.placement_check import check_placement
from backend.periscopex.si_check import check_si
@@ -86,6 +93,10 @@ def merge_schema_pcb_reports(
if st in summary:
summary[st] = summary.get(st, 0) + 1
out["summary"] = summary
schema_nr = list((schema or {}).get("not_reviewed") or [])
pcb_nr = list((pcb or {}).get("not_reviewed") or [])
if schema_nr or pcb_nr:
out["not_reviewed"] = schema_nr + pcb_nr
return out
@@ -101,6 +112,11 @@ def run_pcb_checks(
("placement_check", lambda: check_placement(graph, constraints_map, layout)),
("si_check", lambda: check_si(graph, constraints_map, layout)),
("pcb_derating", lambda: check_pcb_derating(graph)),
("pcb_power_traces", lambda: check_pcb_power_traces(graph, constraints_map, layout)),
("pcb_via_current", lambda: check_pcb_via_current(graph, constraints_map, layout)),
("pcb_thermal_copper", lambda: check_pcb_thermal_copper(graph, constraints_map, layout)),
("pcb_kelvin", lambda: check_pcb_kelvin(graph, constraints_map)),
("pcb_gnd_stitch", lambda: check_pcb_gnd_stitch(graph, layout)),
):
try:
out.extend(fn())
+455
View File
@@ -0,0 +1,455 @@
"""Power-trace, via, thermal-copper, and Kelvin checks on a LayoutGraph.
Skip when current, width, copper thickness, or datasheet numbers are missing.
IPC-2221 is cited only when ΔT can be taken from datasheet Tjmax 25 °C.
"""
from __future__ import annotations
import re
from backend.periscopex.models import (
ComponentType,
DesignGraph,
Finding,
LayoutGraph,
NetType,
)
from backend.periscopex.pcb_net_match import kicad_nets_match, normalize_kicad_hierarchy_net
from backend.periscopex.placement_check import _in_poly
from backend.periscopex.thermal_check import (
_LOAD_KEYS,
_TA_C,
_THETA_KEYS,
_VIN_PIN,
_VOUT_PIN,
_first,
_net_voltage,
_pin_net_by_role,
_specs_values,
)
from backend.periscopex.validate import _match_constraints
# IPC-2221 §6.2 (empirical): I = k · ΔT^0.44 · A^0.725, A in mil², I in A.
_IPC_B = 0.44
_IPC_C = 0.725
_IPC_K_EXT = 0.048
_IPC_K_INT = 0.024
_TJMAX_KEYS = ("tj_max", "tj_max_c", "tjmax", "max_junction_temp_c", "t_jmax")
_SENSE_RE = re.compile(r"(kelvin|sense|isns|i_sns|cs\+|cs-|iout_sns)", re.I)
def _mm_to_mil(mm: float) -> float:
return mm / 0.0254
def _ipc2221_amps(area_mil2: float, dt_c: float, *, external: bool) -> float:
k = _IPC_K_EXT if external else _IPC_K_INT
return k * (dt_c ** _IPC_B) * (area_mil2 ** _IPC_C)
def _min_width_layers(layout: LayoutGraph, sch_net: str) -> tuple[float | None, list[str]]:
widths: list[float] = []
layers: set[str] = set()
for s in layout.segments:
if not s.net or s.width <= 0:
continue
if kicad_nets_match(s.net, sch_net):
widths.append(s.width)
if s.layer:
layers.add(s.layer)
if not widths:
return None, sorted(layers)
return min(widths), sorted(layers)
def _via_stats(layout: LayoutGraph, sch_net: str) -> tuple[int, float | None]:
n = 0
drills: list[float] = []
for v in layout.vias:
if not v.net or not kicad_nets_match(v.net, sch_net):
continue
n += 1
if v.drill and v.drill > 0:
drills.append(v.drill)
return n, (min(drills) if drills else None)
def _external_layers(layers: list[str]) -> bool:
if not layers:
return True
return all(
ly.upper().startswith("F.") or ly.upper().startswith("B.")
for ly in layers
)
def _load_on_output(
graph: DesignGraph, cmap: dict, ref: str,
) -> tuple[float, str, str] | None:
"""I_load (not Iout_max) on an IC output net, plus that net name."""
comp = graph.components.get(ref)
if not comp or comp.component_type != ComponentType.IC:
return None
cons = _match_constraints(comp.mpn or comp.value, cmap)
values = _specs_values(comp)
i_load = _first(values, _LOAD_KEYS)
if i_load is None:
return None
vout = _pin_net_by_role(graph, comp, cons, _VOUT_PIN)
if not vout:
return None
return i_load, vout, ref
def check_pcb_power_traces(
graph: DesignGraph,
constraints_map: dict,
layout: LayoutGraph | None,
) -> list[Finding]:
"""Trace width × copper thickness vs datasheet I_load (IPC-2221 when ΔT known)."""
if layout is None:
return []
t_mm = layout.stackup.copper_thickness_mm if layout.stackup else None
out: list[Finding] = []
seen_nets: set[str] = set()
for ref, comp in sorted(graph.components.items()):
got = _load_on_output(graph, constraints_map, ref)
if not got:
continue
i_load, net, _ = got
key = normalize_kicad_hierarchy_net(net)
if key in seen_nets:
continue
seen_nets.add(key)
w_mm, layers = _min_width_layers(layout, net)
if w_mm is None:
continue
if t_mm is None or t_mm <= 0:
out.append(Finding(
designator=ref,
mpn=comp.mpn or "",
aspect="layout_power",
finding=(
f"{net} carries I_load={i_load:.3g} A with min trace "
f"{w_mm:g} mm; copper thickness is not in the PCB stackup."
),
why="IPC-2221 ampacity needs both width and copper thickness from the board.",
status="INFO",
recommendation=(
"Fill the KiCad stackup copper thickness (or export it) and re-run PCB review."
),
source="pcb_power_thermal",
rule_id="PE-PWR-001",
net=net,
pins=[],
))
continue
area_mil2 = _mm_to_mil(w_mm) * _mm_to_mil(t_mm)
values = _specs_values(comp)
tjmax = _first(values, _TJMAX_KEYS)
if tjmax is None or tjmax <= _TA_C:
out.append(Finding(
designator=ref,
mpn=comp.mpn or "",
aspect="layout_power",
finding=(
f"{net} I_load={i_load:.3g} A on {w_mm:g} mm × {t_mm * 1000:g} µm copper "
f"(A≈{area_mil2:.3g} mil²). Tjmax missing — IPC-2221 not applied."
),
why="ΔT for IPC-2221 is Tjmax 25 °C from the datasheet, never a default 10 °C.",
status="INFO",
recommendation="Add Tjmax to the shared library extraction, then re-run PCB review.",
source="pcb_power_thermal",
rule_id="PE-PWR-001",
net=net,
pins=[],
))
continue
dt = tjmax - _TA_C
ext = _external_layers(layers)
i_allow = _ipc2221_amps(area_mil2, dt, external=ext)
if i_load <= i_allow:
continue
out.append(Finding(
designator=ref,
mpn=comp.mpn or "",
aspect="layout_power",
finding=(
f"{net} I_load={i_load:.3g} A exceeds IPC-2221 capacity "
f"{i_allow:.3g} A ({'external' if ext else 'internal'} k="
f"{_IPC_K_EXT if ext else _IPC_K_INT}, ΔT={dt:.0f} °C, "
f"w={w_mm:g} mm, t={t_mm * 1000:g} µm)."
),
why="IPC-2221 §6.2 I = k·ΔT^0.44·A^0.725. I_load is the datasheet/typical load, not Iout_max.",
status="ERROR",
recommendation=(
f"Widen {net} (and/or add copper/vias/planes) so ampacity ≥ {i_load:.3g} A, "
"or reduce load current."
),
source="pcb_power_thermal",
rule_id="PE-PWR-001",
net=net,
pins=[],
))
return out
def check_pcb_via_current(
graph: DesignGraph,
constraints_map: dict,
layout: LayoutGraph | None,
) -> list[Finding]:
"""Share I_load across vias; no invented via-ampacity table."""
if layout is None:
return []
out: list[Finding] = []
seen: set[str] = set()
for ref, comp in sorted(graph.components.items()):
got = _load_on_output(graph, constraints_map, ref)
if not got:
continue
i_load, net, _ = got
key = normalize_kicad_hierarchy_net(net)
if key in seen:
continue
seen.add(key)
n, drill = _via_stats(layout, net)
if n <= 0:
continue
per = i_load / n
out.append(Finding(
designator=ref,
mpn=comp.mpn or "",
aspect="layout_power",
finding=(
f"{net} shares I_load={i_load:.3g} A across {n} via(s) "
f"(≈{per:.3g} A each"
+ (f", min drill {drill:g} mm" if drill else "")
+ "). No via current rating in the library extraction."
),
why="Via current is reported from board geometry + datasheet I_load only.",
status="INFO",
recommendation=(
"If the datasheet or stackup vendor quotes via current, add more/larger vias "
f"so each via stays under that rating at {i_load:.3g} A."
),
source="pcb_power_thermal",
rule_id="PE-VIA-001",
net=net,
pins=[],
))
return out
def check_pcb_thermal_copper(
graph: DesignGraph,
constraints_map: dict,
layout: LayoutGraph | None,
) -> list[Finding]:
"""Dissipating IC with P from I_load×drop and no pour/vias under courtyard."""
if layout is None:
return []
out: list[Finding] = []
for ref, comp in sorted(graph.components.items()):
if comp.component_type != ComponentType.IC:
continue
cons = _match_constraints(comp.mpn or comp.value, constraints_map)
values = _specs_values(comp)
i_load = _first(values, _LOAD_KEYS)
if i_load is None:
continue
vin_n = _pin_net_by_role(graph, comp, cons, _VIN_PIN)
vout_n = _pin_net_by_role(graph, comp, cons, _VOUT_PIN)
vin = _net_voltage(graph, vin_n) if vin_n else None
vout = _net_voltage(graph, vout_n) if vout_n else None
if vin is None or vout is None or vin <= vout:
continue
p = i_load * (vin - vout)
fp = layout.footprints.get(ref)
if not fp or len(fp.courtyard) < 3:
continue
theta = _first(values, _THETA_KEYS)
via_n = 0
for v in layout.vias:
if _in_poly(v.x, v.y, fp.courtyard):
via_n += 1
pour = False
for z in layout.zones:
if not z.net:
continue
if vout_n and kicad_nets_match(z.net, vout_n):
pass
elif vin_n and kicad_nets_match(z.net, vin_n):
pass
elif graph.nets.get(normalize_kicad_hierarchy_net(z.net)) and (
graph.nets.get(z.net) and graph.nets[z.net].net_type == NetType.GROUND
):
pass
else:
# GND by name
leaf = normalize_kicad_hierarchy_net(z.net).upper()
if not (leaf == "GND" or leaf.endswith("/GND") or "GND" in leaf):
continue
for outline in z.outlines:
if len(outline) >= 3 and _in_poly(fp.x, fp.y, outline):
pour = True
break
if via_n or pour:
continue
out.append(Finding(
designator=ref,
mpn=comp.mpn or "",
aspect="layout_thermal",
finding=(
f"{ref} dissipates ≈{p:.3g} W (I_load={i_load:.3g} A) but the courtyard "
"has no thermal vias and no overlapping copper pour."
),
why=(
"P = I_load×(VinVout) from the shared extraction/specs. "
+ (f"θJA={theta:.3g} °C/W. " if theta else "θJA missing. ")
+ "No millimetres invented."
),
status="WARNING",
recommendation=(
"Add a copper pour and/or thermal vias under the package as the datasheet "
"layout page specifies, then re-run PCB review."
),
source="pcb_power_thermal",
rule_id="PE-THM-001",
net=vout_n,
pins=[],
))
return out
def _is_gnd_name(name: str) -> bool:
leaf = normalize_kicad_hierarchy_net(name).upper()
return leaf in {"GND", "AGND", "DGND", "PGND", "VSS"} or leaf.endswith("/GND")
def check_pcb_gnd_stitch(
graph: DesignGraph,
layout: LayoutGraph | None,
) -> list[Finding]:
"""INFO when a signal span has a GND pour but no GND via in its bbox."""
if layout is None:
return []
gnd_zones = [
z for z in layout.zones
if z.net and _is_gnd_name(z.net) and z.outlines
]
if not gnd_zones:
return []
gnd_vias = [
v for v in layout.vias
if v.net and _is_gnd_name(v.net)
]
out: list[Finding] = []
seen: set[str] = set()
for net_name, net in sorted(graph.nets.items()):
if net.net_type != NetType.SIGNAL:
continue
key = normalize_kicad_hierarchy_net(net_name)
if key in seen:
continue
segs = [
s for s in layout.segments
if s.net and kicad_nets_match(s.net, net_name)
]
if len(segs) < 1:
continue
xs: list[float] = []
ys: list[float] = []
for s in segs:
xs.extend([s.start[0], s.end[0]])
ys.extend([s.start[1], s.end[1]])
xmin, xmax = min(xs), max(xs)
ymin, ymax = min(ys), max(ys)
if xmax - xmin < 2.0 and ymax - ymin < 2.0:
continue
if any(xmin <= v.x <= xmax and ymin <= v.y <= ymax for v in gnd_vias):
continue
seen.add(key)
refs = [p.component_ref for p in net.pins[:1]]
ref = refs[0] if refs else net_name
out.append(Finding(
designator=ref,
mpn=(graph.components[ref].mpn if ref in graph.components else "") or "",
aspect="layout_return",
finding=(
f"Signal '{net_name}' spans {xmax - xmin:.1f}×{ymax - ymin:.1f} mm "
"over a GND pour with no GND via in that bounding box."
),
why=(
"Return path / stitch vias are inferred from board zones and vias only "
"(no IEC clearance invented)."
),
status="INFO",
recommendation=(
f"Add GND stitch vias along '{net_name}' so the return current "
"has a nearby via to the ground plane."
),
source="pcb_power_thermal",
rule_id="PE-STCH-001",
net=net_name,
pins=[],
))
return out
def check_pcb_kelvin(
graph: DesignGraph,
constraints_map: dict,
) -> list[Finding]:
"""Sense/Kelvin pin on a net that also carries the load (not a 4-wire tap)."""
out: list[Finding] = []
for ref, comp in sorted(graph.components.items()):
if comp.component_type != ComponentType.IC:
continue
cons = _match_constraints(comp.mpn or comp.value, constraints_map)
if not cons:
continue
for pin in cons.pintable:
blob = " ".join(
x for x in (str(pin.number), pin.name, pin.description or "")
if x
)
if not _SENSE_RE.search(blob):
continue
net = graph.pin_net(ref, str(pin.number))
if not net:
continue
others = [
r for r in graph.components_on_net(net)
if r != ref
]
ics = [
r for r in others
if graph.components.get(r)
and graph.components[r].component_type == ComponentType.IC
]
if len(others) <= 1 and not ics:
continue
if len(others) < 2:
continue
out.append(Finding(
designator=ref,
mpn=comp.mpn or "",
aspect="layout_kelvin",
finding=(
f"{ref} pin {pin.number} ({pin.name}) looks like a current-sense/Kelvin "
f"pin on '{net}' with {len(others)} other parts on the same net."
),
why="A Kelvin/sense pin should tap the shunt, not share the high-current path.",
status="WARNING",
recommendation=(
f"Route {ref}.{pin.number} as a dedicated Kelvin pair to the sense resistor; "
"do not share that net with the load current."
),
source="pcb_power_thermal",
rule_id="PE-KEL-001",
net=net,
pins=[str(pin.number)],
))
return out
+49 -23
View File
@@ -5,41 +5,67 @@ from __future__ import annotations
import math
from backend.periscopex.functional_groups import FunctionalGroupsReport
from backend.periscopex.models import DesignGraph, LayoutGraph
from backend.periscopex.models import ComponentConstraints, DesignGraph, LayoutGraph
from backend.periscopex.pcb_inventory import PcbInventoryReport
from backend.periscopex.si_check import net_length_mm
PCB_SYSTEM_PROMPT = """\
You are an electrical engineer reviewing a PCB layout against the IC \
datasheet. This is an EXAM of the existing board — do not propose a new \
floorplan, do not invent millimetres, and do not write Gerbers.
datasheet knowledge already stored in the shared library extraction \
(library/extracted). Do NOT re-read or request the PDF — schematic review \
already did the deep datasheet exam once.
### Coverage checklist (layout)
- Domains (power-rail islands) and functional groups (IC + satellites: \
decoupling, bulk, filter, crystal, pullup).
- Placement: decoupling and crystal load caps vs datasheet layout notes \
and any numeric layout_rules (max_distance_mm, same_layer, thermal vias).
- Routing: net lengths, differential-pair skew, obvious stubs; length \
match only when the datasheet gives millimetres.
- Impedance: comment on Z0 only when stackup + width numbers are in the \
context. Never assume 50 Ω.
- Derating: flag a capacitor only when both operating and rated voltages \
are present and Vop exceeds Vrated.
- Filters: topology already on the schematic — check whether filter parts \
sit with the IC they serve when coordinates exist.
- Datasheet layout pages (typical application / PCB layout) vs this \
neighborhood.
This is an EXAM of the existing board — do not propose a new floorplan, \
do not invent millimetres, and do not write Gerbers.
### Coverage checklist (layout-only)
- Use pintable, layout_rules, and abs-max from the library JSON in context.
- Domains and functional groups (IC + satellites).
- Placement vs numeric layout_rules (max_distance_mm, same_layer, thermal vias).
- Routing: lengths, skew, stubs — length match only with datasheet millimetres.
- Power copper: comment on width/thickness vs I_load only when those numbers \
are in the context. Never assume 1 oz or 50 Ω.
- Thermal: copper pour / vias under the package vs θJA / layout notes in the extraction.
- Kelvin/sense pins, crystal keepout — only if named in the pintable/layout_rules.
- Creepage/clearance only if the extraction or IEC number is in context.
### Findings
Every finding MUST have status ERROR, WARNING, or INFO and a non-empty \
recommendation (what to change on the board). INFO still needs a next \
step (e.g. "Measure Z0 after stackup is filled in").
Call submit_review. Empty findings with checked_areas is valid when the \
layout matches the datasheet.
recommendation. Call submit_review. Empty findings with checked_areas is \
valid when the layout matches the extraction.
"""
def format_library_extraction(cons: ComponentConstraints) -> str:
"""Compact shared-library dump so PCB review does not re-ingest the PDF."""
lines = [
"### Shared library extraction (library/extracted — do not re-read PDF)",
f"MPN: {cons.mpn}",
f"subtype: {cons.component_subtype or ''}",
f"model_version: {cons.model_version}",
]
if cons.package_info:
lines.append(
f"package: {cons.package_info.package} "
f"({cons.package_info.pin_count} pins)"
)
if cons.layout_rules:
lines.append("layout_rules:")
for rule in cons.layout_rules[:40]:
lines.append(f" {rule}")
if cons.absolute_maximum_ratings:
lines.append("absolute_maximum_ratings:")
for r in cons.absolute_maximum_ratings[:30]:
lines.append(
f" {r.parameter}: min={r.min} max={r.max} {r.unit} (p.{r.source_page})"
)
if cons.pintable:
lines.append("pintable (number name):")
for p in cons.pintable[:80]:
lines.append(f" {p.number} {p.name}")
return "\n".join(lines)
def build_pcb_layout_context(
ic_ref: str,
graph: DesignGraph,
+32 -12
View File
@@ -36,19 +36,39 @@ _ANALYSIS_BUSY = frozenset({
_PLACEMENT_ACTIVE = frozenset({"queued", "running"})
def _load_constraints_map(extracted_dir: Path) -> dict[str, ComponentConstraints]:
def _load_constraints_map(
extracted_dir: Path,
storage: StorageBackend | None = None,
) -> dict[str, ComponentConstraints]:
"""Project extracted/ plus shared ``library/extracted`` (one store for schema+PCB)."""
result: dict[str, ComponentConstraints] = {}
if not extracted_dir.is_dir():
if extracted_dir.is_dir():
for f in extracted_dir.glob("*.json"):
try:
c = ComponentConstraints.model_validate_json(
f.read_text(encoding="utf-8"),
)
except Exception:
logger.exception("skipping bad extraction %s", f)
continue
result[c.mpn] = c
if storage is None:
return result
for f in extracted_dir.glob("*.json"):
try:
c = ComponentConstraints.model_validate_json(
f.read_text(encoding="utf-8"),
)
except Exception:
logger.exception("skipping bad extraction %s", f)
try:
keys = storage.list_recursive("library/extracted/")
except Exception:
logger.exception("listing library/extracted failed")
return result
for key in keys:
if not key.endswith(".json"):
continue
result[c.mpn] = c
try:
data = storage.read_json(key)
c = ComponentConstraints.model_validate(data)
except Exception:
logger.exception("skipping bad library extraction %s", key)
continue
result.setdefault(c.mpn, c)
return result
@@ -110,7 +130,7 @@ async def run_pcb_pipeline(
return
_step(project_id, "classify", "running", "domains and groups")
cmap = _load_constraints_map(ws.local_path("extracted"))
cmap = _load_constraints_map(ws.local_path("extracted"), storage)
plan = build_placement_plan(graph, cmap)
fg_path = ws.local_path("functional_groups.json")
fg_path.write_text(plan.model_dump_json(indent=2) + "\n")
@@ -168,7 +188,7 @@ async def run_pcb_pipeline(
_finish_cancelled(storage, user_id, project_id)
return
_step(project_id, "ai_review", "running", "per-IC datasheet vs layout")
_step(project_id, "ai_review", "running", "layout vs shared library extraction")
coverage: dict[str, list[str]] = {}
skipped: list[dict] = []
try:
+29 -11
View File
@@ -1,4 +1,4 @@
"""Per-IC PCB datasheet review — same agentic loop as schematic, layout context."""
"""Per-IC PCB layout review — consumes shared library extraction, no second PDF exam."""
from __future__ import annotations
@@ -7,12 +7,16 @@ 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.models import ComponentType, DesignGraph, Finding, LayoutGraph
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.periscopex.pcb_review import (
PCB_SYSTEM_PROMPT,
build_pcb_layout_context,
format_library_extraction,
)
from backend.periscopex.validate import ReviewResult, _match_constraints
from backend.services.api_logs import ApiLogger
from backend.services.validation import _find_pdf, review_ic_async
from backend.services.validation import review_ic_async
log = logging.getLogger(__name__)
@@ -27,6 +31,10 @@ def _ensure_recs(findings: list[Finding]) -> None:
f.recommendation = _FIX
def _has_library_extraction(cons) -> bool:
return cons is not None and bool(cons.pintable)
async def review_pcb_ics(
graph: DesignGraph,
constraints_map: dict,
@@ -38,7 +46,11 @@ async def review_pcb_ics(
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."""
"""Layout-only AI exam using ``library/extracted`` (or project extracted/).
Does not attach a datasheet PDF. ICs without a library pintable are skipped
with a reason to run schematic review first.
"""
findings: list[Finding] = []
coverage: dict[str, list[str]] = {}
skipped: list[dict] = []
@@ -50,14 +62,20 @@ async def review_pcb_ics(
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"})
cons = _match_constraints(mpn, constraints_map)
if not _has_library_extraction(cons):
skipped.append({
"designator": ref,
"reason": "no library extraction — run schematic review first",
})
continue
extra = build_pcb_layout_context(ref, graph, layout, plan, inventory)
extra = "\n\n".join([
format_library_extraction(cons),
build_pcb_layout_context(ref, graph, layout, plan, inventory),
])
try:
result, _trace = await review_ic_async(
graph, constraints_map, ref, str(pdf),
graph, constraints_map, ref, None,
on_progress=on_progress,
api_logger=api_logger,
pdf_dir=pdf_dir,
+26 -27
View File
@@ -288,7 +288,7 @@ async def review_ic_async(
graph: DesignGraph,
constraints_map: ConstraintsMap,
ic_ref: str,
pdf_path: str,
pdf_path: str | None,
on_progress: ProgressCallback | None = None,
api_logger: ApiLogger | None = None,
trace_git_commit: str = "unknown",
@@ -301,21 +301,21 @@ async def review_ic_async(
) -> tuple[ReviewResult, dict]:
"""Review one IC against its datasheet. Async, multi-turn.
Returns ``(ReviewResult, trace)`` — ``trace`` is a transcript dict of the
full agentic loop (turns, tool calls + outputs, final submission) for
offline inspection. Trace assembly is best-effort and never affects the
review result.
``pdf_path`` may be ``None`` when the caller already has library
extraction (PCB layout-only exam) — no PDF is attached and citations
are not re-verified against a PDF.
"""
comp = graph.components[ic_ref]
mpn = comp.mpn or comp.value
# Datasheet identity for the trace — hash the original PDF, not the
# trimmed copy, so the reference is stable across trim-heuristic changes.
try:
ds_md5 = hashlib.md5(Path(pdf_path).read_bytes()).hexdigest()
except Exception:
log.exception("trace: datasheet md5 failed for %s", ic_ref)
ds_md5 = None
ds_md5 = None
if pdf_path:
try:
ds_md5 = hashlib.md5(Path(pdf_path).read_bytes()).hexdigest()
except Exception:
log.exception("trace: datasheet md5 failed for %s", ic_ref)
# Pre-compute which designators the excerpt tool will accept for this
# review (neighbors via signal nets only — power/GND fan-out filtered).
@@ -338,7 +338,7 @@ async def review_ic_async(
current_ic=ic_ref,
connected_designators=connected_designators,
graph=graph,
pdf_dir=pdf_dir or Path(pdf_path).parent,
pdf_dir=pdf_dir or (Path(pdf_path).parent if pdf_path else Path(".")),
storage=storage,
cache=excerpt_cache if excerpt_cache is not None else {},
fetch_budget=_PER_REVIEW_FETCH_BUDGET,
@@ -347,7 +347,7 @@ async def review_ic_async(
)
# Trim PDF up-front — both primary and fallback attempts share it.
trimmed_pdf = _select_review_pages(pdf_path)
trimmed_pdf = _select_review_pages(pdf_path) if pdf_path else None
try:
async def _run(provider, model) -> tuple[ReviewResult, dict]:
t0 = time.monotonic()
@@ -377,15 +377,13 @@ async def review_ic_async(
if extra_context.strip():
user_text += "\n\n" + extra_context.strip()
user_blocks: list = []
if trimmed_pdf:
user_blocks.append(PdfBlock(path=Path(trimmed_pdf), cacheable=True))
user_blocks.append(TextBlock(text=user_text, cacheable=True))
initial_msg = Message(
role="user",
content=[
PdfBlock(path=Path(trimmed_pdf), cacheable=True),
TextBlock(
text=user_text,
cacheable=True,
),
],
content=user_blocks,
)
messages: list[Message] = [initial_msg]
@@ -465,13 +463,14 @@ async def review_ic_async(
mpn_by_designator=mpn_by_designator,
connected=connected_designators,
)
verify_finding_citations(
result.findings,
default_pdf=Path(pdf_path),
default_mpn=mpn,
pdf_dir=excerpt_state.pdf_dir,
mpn_by_designator=mpn_by_designator,
)
if pdf_path:
verify_finding_citations(
result.findings,
default_pdf=Path(pdf_path),
default_mpn=mpn,
pdf_dir=excerpt_state.pdf_dir,
mpn_by_designator=mpn_by_designator,
)
turn_record["tool_calls"].append({
"name": "submit_review",
"input": tc.input,
@@ -610,7 +609,7 @@ async def review_ic_async(
return await call_with_fallback("validation", _run)
finally:
if trimmed_pdf != pdf_path:
if trimmed_pdf and pdf_path and trimmed_pdf != pdf_path:
Path(trimmed_pdf).unlink(missing_ok=True)