Fill ferrite bead impedance from the MPN datasheet PDF.

graph_build searches project uploads and library blob refs, persists recovered Z, and still loads when the PDF has no impedance. Ohms are not written to henries.
This commit is contained in:
2026-09-21 00:29:12 +02:00
parent a2d5900b64
commit 72bcab50a1
8 changed files with 210 additions and 15 deletions
@@ -317,7 +317,7 @@ def simple_to_typed_passive_specs(simple: SimpleComponentSpecs) -> ComponentSpec
formatted = value_formatted
if not formatted and impedance_ohm is not None:
formatted = _format_value(impedance_ohm, "ohm")
return InductorSpecs(
spec = InductorSpecs(
component_subtype=subtype_for_specs,
value_henries=None,
value_formatted=formatted or "FB",
@@ -327,6 +327,9 @@ def simple_to_typed_passive_specs(simple: SimpleComponentSpecs) -> ComponentSpec
dcr_ohms=dcr_ohms,
impedance_ohm=impedance_ohm,
)
from backend.periscopex.ferrite_z import recover_bead_specs
extra = " ".join(str(v) for v in vals.values() if v is not None)
return recover_bead_specs(spec, extra_text=extra)
if subtype.startswith("passive.inductor"):
raw = vals.get("value_henries")
@@ -6,6 +6,7 @@ Never invent Z from MPN codes. Prefer datasheet text, then a quoted
from __future__ import annotations
import json
import re
from pathlib import Path
from typing import TYPE_CHECKING
@@ -94,22 +95,86 @@ def z_from_pdf(path: Path | str, *, max_chars: int = 80_000) -> tuple[float, str
def find_mpn_pdf(mpn: str, *dirs: Path | str) -> Path | None:
"""Locate a manufacturer PDF for *mpn* under project uploads or library blobs."""
safe = safe_mpn(mpn)
names = (f"{safe}.pdf", f"{mpn}.pdf")
for d in dirs:
root = Path(d)
if root.is_file() and root.suffix.lower() == ".pdf":
stem = root.stem.lower()
if safe.lower() in stem or mpn.lower() in stem:
return root
continue
if not root.is_dir():
continue
for name in names:
p = root / name
if p.is_file():
return p
for sub in (
root / "uploads" / "datasheets",
root / "datasheets",
root / "library" / "datasheets",
):
for name in names:
p = sub / name
if p.is_file():
return p
blob = _pdf_from_library_ref(root, safe)
if blob is not None:
return blob
if root.name.lower() in {"datasheets", "blobs"} or "datasheet" in root.name.lower():
for p in root.glob("*.pdf"):
if safe.lower() in p.stem.lower() or mpn.lower() in p.stem.lower():
return p
return None
def _pdf_from_library_ref(root: Path, safe: str) -> Path | None:
refs = (
root / "library" / "datasheets" / "refs" / f"{safe}.json",
root / "datasheets" / "refs" / f"{safe}.json",
root / "refs" / f"{safe}.json",
)
for ref in refs:
if not ref.is_file():
continue
try:
data = json.loads(ref.read_text())
except (OSError, json.JSONDecodeError):
continue
rels: list[str] = []
key = str(data.get("blob_key") or "").strip()
if key:
rels.append(key)
digest = str(data.get("hash") or "").strip()
if digest:
rels.extend(
(
f"library/datasheets/blobs/{digest}.pdf",
f"datasheets/blobs/{digest}.pdf",
f"blobs/{digest}.pdf",
)
)
bases = (root, root.parent, ref.parents[2] if len(ref.parents) >= 3 else root)
for base in bases:
for rel in rels:
p = Path(rel) if Path(rel).is_absolute() else base / rel
if p.is_file():
return p
return None
def default_pdf_roots(workspace_dir: Path | str, data_dir: Path | str | None = None) -> list[Path]:
"""Project datasheet folder plus the shared library root used by storage."""
ws = Path(workspace_dir)
roots = [ws / "uploads" / "datasheets", ws / "datasheets"]
if data_dir:
d = Path(data_dir)
roots.extend((d, d / "library"))
return roots
def recover_bead_specs(
specs: InductorSpecs,
*,
+17 -11
View File
@@ -157,11 +157,19 @@ def _match_datasheet(
return None, None
def _load_component_models(directory: str | Path) -> dict[str, ComponentSpecs]:
def _load_component_models(
directory: str | Path,
extra_pdf_dirs: tuple[Path | str, ...] | list[Path | str] = (),
) -> dict[str, ComponentSpecs]:
result: dict[str, ComponentSpecs] = {}
dirpath = Path(directory)
if not dirpath.is_dir():
return result
pdf_dirs: list[Path | str] = [
dirpath.parent / "uploads" / "datasheets",
dirpath.parent / "datasheets",
*extra_pdf_dirs,
]
for json_file in dirpath.glob("*.json"):
try:
raw = json.loads(json_file.read_text())
@@ -173,16 +181,11 @@ def _load_component_models(directory: str | Path) -> dict[str, ComponentSpecs]:
if getattr(specs, "component_subtype", None) == "passive.ferrite_bead":
from backend.periscopex.ferrite_z import find_mpn_pdf, recover_bead_specs
pdf = find_mpn_pdf(
model.mpn,
dirpath.parent / "uploads" / "datasheets",
dirpath.parent / "datasheets",
)
pdf = find_mpn_pdf(model.mpn, *pdf_dirs)
before_z = getattr(specs, "impedance_ohm", None)
before_h = getattr(specs, "value_henries", None)
filled = recover_bead_specs(specs, extra_text=json.dumps(raw), pdf_path=pdf)
if (
filled.impedance_ohm != getattr(specs, "impedance_ohm", None)
or filled.value_henries != getattr(specs, "value_henries", None)
):
if filled.impedance_ohm != before_z or filled.value_henries != before_h:
try:
_save_component_model(model.mpn, filled, dirpath)
except Exception:
@@ -283,6 +286,7 @@ def build_graph(
skipped: list[SkippedItem] | None = None,
include_subdesigns: set[str] | None = None,
pcb_path: str | Path | None = None,
extra_pdf_dirs: tuple[Path | str, ...] | list[Path | str] | None = None,
) -> DesignGraph:
"""Deterministic graph: BOM → netlist → optional board nets → datasheets → DesignGraph."""
bom = parse_bom(bom_path, reference_col=reference_col, mpn_col=mpn_col)
@@ -303,7 +307,9 @@ def build_graph(
datasheets = _load_datasheets(datasheets_dir)
models_dir = Path(component_models_dir)
mpn_specs: dict[str, ComponentSpecs] = _load_component_models(models_dir)
mpn_specs: dict[str, ComponentSpecs] = _load_component_models(
models_dir, extra_pdf_dirs=extra_pdf_dirs or (),
)
mpn_subtype: dict[str, str] = {}
for rp in resolve_bom(
@@ -22,6 +22,7 @@ from backend.periscopex.cad_bridge import (
from backend.periscopex.finding_engine import apply_decisions
from backend.periscopex.functional_groups import build_placement_plan
from backend.periscopex.graph import build_graph
from backend.periscopex.ferrite_z import default_pdf_roots
from backend.periscopex.layout_rules import needs_layout_rules_refresh
from backend.periscopex.models import (
ComponentConstraints, ComponentType, DesignGraph, LayoutGraph, ValidationReport,
@@ -406,6 +407,7 @@ async def _ensure_graph(ws: PipelineWorkspace, meta, project_id: str) -> DesignG
else None
),
pcb_path=ws.local_path("uploads/pcb.kicad_pcb"),
extra_pdf_dirs=default_pdf_roots(ws.local_dir, app_settings.data_dir),
)
graph_path.write_text(graph.model_dump_json(indent=2) + "\n")
ws._upload_file("design_graph.json")
@@ -21,6 +21,7 @@ from backend.periscopex.bom_summary import build_bom_summary
from backend.periscopex.constraints_lookup import build_constraints_map, load_datasheets
from backend.periscopex.derating import build_derating_table
from backend.periscopex.graph import build_graph
from backend.periscopex.ferrite_z import default_pdf_roots
from backend.periscopex.models import ComponentType
from backend.periscopex.parsers import parse_bom, parse_netlist_any
from backend.periscopex.resolve_passives import SkippedItem, load_patterns, resolve_mpn
@@ -1077,6 +1078,7 @@ async def _stage_graph_build(ctx: PipelineContext) -> None:
if ctx.meta.netlist_subdesigns is not None else None
),
pcb_path=ctx.ws.local_path("uploads/pcb.kicad_pcb"),
extra_pdf_dirs=default_pdf_roots(ctx.ws.local_dir, settings.data_dir),
)
ctx.ws.local_path("design_graph.json").write_text(ctx.graph.model_dump_json(indent=2) + "\n")
_write_layout_graph(ctx.ws, ctx.project_id)
@@ -1459,6 +1461,7 @@ async def run_regen_pipeline(
if meta.netlist_subdesigns is not None else None
),
pcb_path=ws.local_path("uploads/pcb.kicad_pcb"),
extra_pdf_dirs=default_pdf_roots(ws.local_dir, settings.data_dir),
)
ws.local_path("design_graph.json").write_text(graph.model_dump_json(indent=2) + "\n")
_write_layout_graph(ws, project_id)
@@ -11,6 +11,8 @@ from __future__ import annotations
import logging
from pathlib import Path
from backend.config import settings as app_settings
from backend.periscopex.ferrite_z import default_pdf_roots
from backend.periscopex.functional_groups import build_placement_plan
from backend.periscopex.graph import build_graph
from backend.periscopex.models import ComponentConstraints, DesignGraph, LayoutGraph
@@ -193,6 +195,7 @@ async def _ensure_graph(ws: PipelineWorkspace, meta, project_id: str) -> DesignG
else None
),
pcb_path=ws.local_path("uploads/pcb.kicad_pcb"),
extra_pdf_dirs=default_pdf_roots(ws.local_dir, app_settings.data_dir),
)
graph_path.write_text(graph.model_dump_json(indent=2) + "\n")
ws._upload_file("design_graph.json")
@@ -2,6 +2,13 @@
What's new in Periscope.
## 2.60.3 — 2026-09-21 — Ferrite Z from library MPN datasheet
graph_build looks up the bead MPN PDF in project `uploads/datasheets` and `library/datasheets` (named files and blob refs). Z is filled only when the PDF states impedance (e.g. 120 Ω @ 100 MHz). Missing Z stays omitted; graph_build does not abort.
- [Fixed] HubAudio-style beads recover `impedance_ohm` from the manufacturer PDF, not from inventing Z or stuffing ohms into `value_henries`.
- [New] Tests: MPN + datasheet PDF fills Z; MPN + PDF without Z does not invent and still loads.
## 2.60.2 — 2026-09-20 — Ferrite Z from datasheet, no graph_build abort
Beads with an MPN pull `impedance_ohm` from the datasheet or a quoted `N ohm @ freq` string. Missing Z after that is INSUFFICIENT, not a crash. Ohms are never written to `value_henries`. Leftover dependency validator no longer requires Z.
+106
View File
@@ -147,3 +147,109 @@ def test_bead_mpn_datasheet_without_z_does_not_invent():
assert filled.impedance_ohm is None
model = ComponentModel(mpn="BLM21PG121SN1D", specs=filled)
assert model.specs.impedance_ohm is None
def _write_text_pdf(path: Path, text: str) -> None:
"""Write a one-page PDF pypdf can extract without PyMuPDF."""
escaped = (
text.replace("\\", "\\\\").replace("(", "\\(").replace(")", "\\)")
)
lines = escaped.split("\n")
ops = ["BT /F1 12 Tf 72 720 Td"]
for i, line in enumerate(lines):
if i:
ops.append("0 -16 Td")
ops.append(f"({line}) Tj")
ops.append("ET")
stream = "\n".join(ops).encode("latin-1", "replace")
objects = [
b"1 0 obj << /Type /Catalog /Pages 2 0 R >> endobj\n",
b"2 0 obj << /Type /Pages /Kids [3 0 R] /Count 1 >> endobj\n",
(
b"3 0 obj << /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] "
b"/Contents 4 0 R /Resources << /Font << /F1 5 0 R >> >> >> endobj\n"
),
b"4 0 obj << /Length %d >> stream\n" % len(stream) + stream + b"\nendstream endobj\n",
b"5 0 obj << /Type /Font /Subtype /Type1 /BaseFont /Helvetica >> endobj\n",
]
header = b"%PDF-1.4\n"
body = header
offsets = [0]
for obj in objects:
offsets.append(len(body))
body += obj
xref_pos = len(body)
xref = [b"xref\n0 6\n0000000000 65535 f \n"]
for off in offsets[1:]:
xref.append(f"{off:010d} 00000 n \n".encode())
trailer = (
b"trailer << /Root 1 0 R /Size 6 >>\nstartxref\n"
+ str(xref_pos).encode()
+ b"\n%%EOF\n"
)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(body + b"".join(xref) + trailer)
_BEAD_MODEL = {
"mpn": "BLM21PG121SN1D",
"specs": {
"specs_type": "inductor",
"component_subtype": "passive.ferrite_bead",
"value_formatted": "FB",
"current_rating_a": "3A",
"dcr_ohms": 0.03,
},
}
def test_bead_mpn_datasheet_pdf_fills_z(tmp_path: Path):
models = tmp_path / "models"
models.mkdir()
(models / "BLM21PG121SN1D.json").write_text(json.dumps(_BEAD_MODEL) + "\n")
_write_text_pdf(
tmp_path / "uploads" / "datasheets" / "BLM21PG121SN1D.pdf",
"BLM21PG121SN1D\nImpedance (at 100MHz / 20C) 120 ohm\nRated current 3A",
)
loaded = _load_component_models(models)
assert loaded["BLM21PG121SN1D"].impedance_ohm == 120.0
assert loaded["BLM21PG121SN1D"].value_henries is None
persisted = json.loads((models / "BLM21PG121SN1D.json").read_text())
assert persisted["specs"]["impedance_ohm"] == 120.0
assert persisted["specs"]["value_henries"] is None
assert "100" in (persisted["specs"].get("value_formatted") or "")
def test_bead_mpn_datasheet_pdf_without_z_does_not_invent(tmp_path: Path):
models = tmp_path / "models"
models.mkdir()
(models / "BLM21PG121SN1D.json").write_text(json.dumps(_BEAD_MODEL) + "\n")
_write_text_pdf(
tmp_path / "uploads" / "datasheets" / "BLM21PG121SN1D.pdf",
"BLM21PG121SN1D\nRated current 3 A\nDCR 0.03 ohm max\nPackage 0805",
)
loaded = _load_component_models(models)
assert "BLM21PG121SN1D" in loaded
assert loaded["BLM21PG121SN1D"].impedance_ohm is None
assert loaded["BLM21PG121SN1D"].value_henries is None
def test_bead_mpn_library_blob_fills_z(tmp_path: Path):
models = tmp_path / "proj" / "models"
models.mkdir(parents=True)
(models / "BLM21PG121SN1D.json").write_text(json.dumps(_BEAD_MODEL) + "\n")
blob = tmp_path / "library" / "datasheets" / "blobs" / "deadbeef.pdf"
_write_text_pdf(
blob,
"Impedance (at 100MHz / 20C) 120 ohm\nRated current 3A",
)
ref = tmp_path / "library" / "datasheets" / "refs" / "BLM21PG121SN1D.json"
ref.parent.mkdir(parents=True, exist_ok=True)
ref.write_text(json.dumps({
"hash": "deadbeef",
"blob_key": "library/datasheets/blobs/deadbeef.pdf",
"mpn": "BLM21PG121SN1D",
}) + "\n")
loaded = _load_component_models(models, extra_pdf_dirs=[tmp_path])
assert loaded["BLM21PG121SN1D"].impedance_ohm == 120.0
assert loaded["BLM21PG121SN1D"].value_henries is None