michele/periscope
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Show cached passives in the library and match IC PDFs by family MPN.

Browse Source 
Decode EIA/Yageo/AVX chip codes into library specs, and find datasheets when the file is stored under the orderable catalog name.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
michele
2026-08-28 02:27:27 +02:00
co-authored by Cursor
parent 3be7d2fc21
commit 08cbbdc422
11 changed files with 534 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/services/datasheet_finder.py
+29
View File
@@ -19,6 +19,7 @@ from __future__ import annotations
import logging
import re
from dataclasses import dataclass
from pathlib import Path
import httpx
@@ -138,6 +139,34 @@ def mpn_catalog_match(query: str, candidate: str) -> bool:
return c.startswith(q) and len(c) > len(q)
def find_local_pdf(pdf_dir: Path, mpn: str) -> Path | None:
"""Find a datasheet PDF whose filename is this MPN or a catalog alias.
``ESP32-S31-WROOM-3`` matches ``ESP32-S31-WROOM-3-N16R16V.pdf`` and the
reverse — packing / flash-size suffixes, not sibling dies (CH340 vs CH340E).
"""
from backend.pinscopex.utils import safe_mpn
if not mpn or not pdf_dir.is_dir():
return None
for name in mpn_query_variants(mpn) or [mpn]:
hit = pdf_dir / f"{safe_mpn(name)}.pdf"
if hit.is_file():
return hit
want = _alnum(mpn)
if len(want) < _MIN_FAMILY_LEN:
return None
family_hit: Path | None = None
for hit in pdf_dir.glob("*.pdf"):
stem = hit.stem
if mpn_matches(mpn, stem) or mpn_catalog_match(mpn, stem) or mpn_catalog_match(stem, mpn):
got = _alnum(stem)
if got == want or mpn_matches(mpn, stem):
return hit
family_hit = family_hit or hit
return family_hit
def _pick_lcsc_product(mpn: str, products: list[dict]) -> dict | None:
exact: dict | None = None
loose: dict | None = None
backend/services/passive_from_mpn.py
+139
View File
@@ -0,0 +1,139 @@
"""Decode common chip R/C MPNs into typed specs without an LLM.
Only encodings that carry package + value (and voltage for capacitors when
the manufacturer puts it in the code) are accepted. Incomplete BOM-value
guesses stay out of the shared library.
"""
from __future__ import annotations
import re
from backend.pinscopex.models import ComponentModel, SimpleComponentSpecs
from backend.pinscopex.resolve_passives import simple_to_typed_passive_specs
from backend.services.passive_from_distributor import _spice
_SIZE = r"(?:0201|0402|0603|0805|1206|1210|1812|2010|2512)"
_DIEL = r"(?:C0G|NP0|CG|X8R|X7R|X7S|X6S|X5R|Y5V)"
_TOL = {
"B": "±0.1%",
"C": "±0.25%",
"D": "±0.5%",
"F": "±1%",
"G": "±2%",
"J": "±5%",
"K": "±10%",
"M": "±20%",
}
_AVX_V = {
"4": "4V",
"6": "6.3V",
"Z": "10V",
"Y": "16V",
"3": "25V",
"5": "50V",
"1": "100V",
"2": "200V",
"7": "500V",
}
_AVX_DIEL = {"A": "C0G", "C": "X7R", "D": "X5R", "Z": "Y5V"}
_DIEL_NORM = {"CG": "C0G", "NP0": "C0G"}
_WALSIN_CAP = re.compile(
rf"^({_SIZE})({_DIEL})(\d{{3}})([{''.join(_TOL)}])(\d{{3}}|[0-9]R[0-9])",
re.IGNORECASE,
)
_AVX_CAP = re.compile(
rf"^({_SIZE})([46ZY35127])([{''.join(_AVX_DIEL)}])(\d{{3}})([{''.join(_TOL)}])",
re.IGNORECASE,
)
_CHIP_R = re.compile(
rf"^(?:FRC)?({_SIZE})(?:W\d)?([{''.join('FJKG')}])(\d{{4}})",
re.IGNORECASE,
)
def _eia3_pf(digits: str) -> float:
return float(int(digits[:2]) * (10 ** int(digits[2])))
def _eia3_volts(code: str) -> str | None:
code = code.upper()
if "R" in code:
try:
return f"{float(code.replace('R', '.')):g}V"
except ValueError:
return None
if len(code) != 3 or not code.isdigit():
return None
volts = int(code[:2]) * (10 ** int(code[2]))
return f"{volts}V"
def _eia4_ohm(digits: str) -> float:
return float(int(digits[:3]) * (10 ** int(digits[3])))
def _model(mpn: str, subtype: str, values: dict[str, str]) -> ComponentModel | None:
specs = SimpleComponentSpecs(
specs_type="passive",
component_subtype=subtype,
values=values,
)
try:
typed = simple_to_typed_passive_specs(specs)
except (ValueError, TypeError):
return None
return ComponentModel(mpn=mpn, specs=typed)
def specs_from_mpn(mpn: str) -> ComponentModel | None:
"""Return a ComponentModel when the MPN itself encodes enough specs."""
raw = (mpn or "").strip()
if not raw:
return None
m = _WALSIN_CAP.match(raw)
if m:
size, diel, cap, tol, volt = m.groups()
pf = _eia3_pf(cap)
values = {
"value_farads": _spice(str(pf), "p", "F") if pf else None,
"package": size.upper(),
"dielectric": _DIEL_NORM.get(diel.upper(), diel.upper()),
"tolerance": _TOL[tol.upper()],
}
rated = _eia3_volts(volt)
if rated:
values["voltage_rating_v"] = rated
values["value_formatted"] = values["value_farads"]
if values["value_farads"]:
return _model(raw, "passive.capacitor.ceramic", values)
m = _AVX_CAP.match(raw)
if m:
size, vcode, diel, cap, tol = m.groups()
pf = _eia3_pf(cap)
values = {
"value_farads": _spice(str(pf), "p", "F"),
"value_formatted": _spice(str(pf), "p", "F"),
"package": size.upper(),
"dielectric": _AVX_DIEL[diel.upper()],
"tolerance": _TOL[tol.upper()],
"voltage_rating_v": _AVX_V[vcode.upper()],
}
return _model(raw, "passive.capacitor.ceramic", values)
m = _CHIP_R.match(raw)
if m:
size, tol, code = m.groups()
ohms = _eia4_ohm(code)
values = {
"value_ohms": _spice(str(ohms), None, "ohm"),
"value_formatted": _spice(str(ohms), None, "ohm"),
"package": size.upper(),
"tolerance": _TOL[tol.upper()],
}
return _model(raw, "passive.resistor.thick_film", values)
return None
backend/services/pipeline.py
+24 -10
View File
@@ -681,15 +681,16 @@ async def _ensure_local_datasheet(
"""
if pdf_path.is_file():
return True
from backend.services.datasheet_finder import find_datasheet, mpn_query_variants
from backend.services.datasheet_finder import find_datasheet, find_local_pdf, mpn_query_variants
alt = find_local_pdf(pdf_path.parent, mpn)
if alt is not None and alt.is_file():
if alt.resolve() != pdf_path.resolve():
pdf_path.parent.mkdir(parents=True, exist_ok=True)
pdf_path.write_bytes(alt.read_bytes())
return True
for name in mpn_query_variants(mpn) or [mpn]:
alt = pdf_path.parent / f"{safe_mpn(name)}.pdf"
if alt.is_file():
if alt.resolve() != pdf_path.resolve():
pdf_path.parent.mkdir(parents=True, exist_ok=True)
pdf_path.write_bytes(alt.read_bytes())
return True
lib_ds_key = proj_svc.library_has_datasheet(ctx.storage, name)
if lib_ds_key:
ctx.storage.download_to_local(lib_ds_key, pdf_path)
@@ -1116,6 +1117,16 @@ async def _catalog_resolve_unresolved_passives(
except Exception as e:
first_error = str(e)
if model is None:
from backend.services.passive_from_mpn import specs_from_mpn
model = specs_from_mpn(mpn)
if model is not None:
resolved_via = "mpn"
broker.publish(ctx.project_id, "step_update",
{"stage": "passive_extraction",
"substep": mpn, "status": "running",
"detail": "decoded from MPN"})
if model is None:
bom_value = ctx.passive_values.get(mpn, "").strip()
refs = ctx.passive_mpns.get(mpn, [])
@@ -1152,7 +1163,7 @@ async def _catalog_resolve_unresolved_passives(
model_path.write_text(model.model_dump_json(indent=2) + "\n")
model_key = f"{ctx.ws.prefix}/models/{safe}.json"
ctx.storage.upload_from_local(model_path, model_key)
if resolved_via in ("digikey", "lcsc"):
if resolved_via in ("digikey", "lcsc", "mpn"):
proj_svc.save_to_library(
ctx.storage, model_key, "passives", f"{safe}.json",
)
@@ -1161,6 +1172,7 @@ async def _catalog_resolve_unresolved_passives(
detail = {
"lcsc": "auto-resolved via LCSC",
"digikey": "auto-resolved via DigiKey",
"mpn": "decoded from MPN (saved to library)",
"value": "resolved from BOM value (not saved to library)",
}.get(resolved_via, "resolved")
broker.publish(ctx.project_id, "step_update",
@@ -1502,14 +1514,16 @@ async def _stage_validation(ctx: PipelineContext) -> None:
# Snapshot the full review queue so pause checkpoints can show what's left.
# Mirrors the filter in validate_design_async: ICs with a PDF available.
from backend.services.datasheet_finder import find_local_pdf
planned_refs: list[str] = []
for ref, comp in ctx.graph.components.items():
if comp.component_type != ComponentType.IC:
continue
mpn = comp.mpn or comp.value
mpn = (comp.mpn or "").strip() or (comp.value or "").strip()
if not mpn:
continue
if (ds_dir / f"{safe_mpn(mpn)}.pdf").is_file():
if find_local_pdf(ds_dir, mpn) is not None:
planned_refs.append(ref)
ctx.all_review_refs = sorted(planned_refs, key=natural_sort_key)
backend/services/projects.py
+36 -11
View File
@@ -775,6 +775,31 @@ def save_to_library(
return dst_key
def _specs_param_count(specs: dict) -> int:
if not isinstance(specs, dict):
return 0
values = specs.get("values")
if isinstance(values, dict):
return sum(1 for v in values.values() if v not in (None, "", []))
skip = {"specs_type", "component_subtype"}
return sum(
1 for k, v in specs.items()
if k not in skip and v not in (None, "", [])
)
def _catalog_model_row(data: dict, key: str, *, row_type: str) -> dict:
mpn = data.get("mpn", "") or key.rsplit("/", 1)[-1].replace(".json", "")
specs = data.get("specs", {}) or {}
return {
"mpn": mpn,
"type": row_type,
"specs_type": specs.get("specs_type", ""),
"subtype": specs.get("component_subtype", ""),
"param_count": _specs_param_count(specs),
}
def list_library_catalog(storage: StorageBackend) -> dict:
"""List ICs, passive patterns, discrete specs, and datasheet refs.
@@ -826,26 +851,24 @@ def list_library_catalog(storage: StorageBackend) -> dict:
continue
simple_models: list[dict] = []
passive_parts: list[dict] = []
seen_model_mpns: set[str] = set()
for prefix in ("library/models/", "library/passives/"):
for prefix, row_type, dest in (
("library/passives/", "passive_part", passive_parts),
("library/models/", "simple", simple_models),
):
for key in storage.list_prefix(prefix):
if not key.endswith(".json"):
continue
try:
data = storage.read_json(key)
mpn = data.get("mpn", "") or key.rsplit("/", 1)[-1].replace(".json", "")
row = _catalog_model_row(data, key, row_type=row_type)
mpn = row["mpn"]
if mpn in seen_model_mpns:
continue
seen_model_mpns.add(mpn)
specs = data.get("specs", {}) or {}
simple_models.append({
"mpn": mpn,
"type": "simple",
"specs_type": specs.get("specs_type", ""),
"subtype": specs.get("component_subtype", ""),
"param_count": len(specs.get("values", {}) or {}),
"has_datasheet": bool(resolve_datasheet(storage, mpn)),
})
row["has_datasheet"] = bool(resolve_datasheet(storage, mpn))
dest.append(row)
except Exception:
continue
@@ -871,11 +894,13 @@ def list_library_catalog(storage: StorageBackend) -> dict:
ics.sort(key=lambda r: r["mpn"].lower())
passives.sort(key=lambda r: r["mpn"].lower())
passive_parts.sort(key=lambda r: r["mpn"].lower())
simple_models.sort(key=lambda r: r["mpn"].lower())
datasheets.sort(key=lambda r: r["mpn"].lower())
return {
"ics": ics,
"passives": passives,
"passive_parts": passive_parts,
"simple": simple_models,
"datasheets": datasheets,
}
backend/services/validation.py
+22 -15
View File
@@ -571,27 +571,29 @@ def _find_pdf(
"""Find the datasheet PDF for an MPN. Checks local dir first,
then tries to download from the library.
"""
from backend.services.datasheet_finder import mpn_query_variants
from backend.services.datasheet_finder import find_local_pdf
from backend.pinscopex.utils import safe_mpn as _safe
names = mpn_query_variants(mpn) or [mpn]
for name in names:
local = pdf_dir / f"{_safe(name)}.pdf"
if local.is_file():
wanted = pdf_dir / f"{_safe(mpn)}.pdf"
if local != wanted and not wanted.is_file():
wanted.write_bytes(local.read_bytes())
return wanted
return local
mpn = (mpn or "").strip()
if not mpn:
return None
local = find_local_pdf(pdf_dir, mpn)
if local is not None and local.is_file():
wanted = pdf_dir / f"{_safe(mpn)}.pdf"
if local.resolve() != wanted.resolve() and not wanted.is_file():
wanted.write_bytes(local.read_bytes())
return wanted
return local
if storage:
from backend.services import projects as proj_svc
lib_key = proj_svc.library_has_datasheet(storage, mpn)
if lib_key:
local = pdf_dir / f"{_safe(mpn)}.pdf"
storage.download_to_local(lib_key, local)
if local.is_file():
return local
wanted = pdf_dir / f"{_safe(mpn)}.pdf"
storage.download_to_local(lib_key, wanted)
if wanted.is_file():
return wanted
return None
@@ -665,7 +667,12 @@ async def validate_design_async(
for ref, comp in sorted(graph.components.items()):
if comp.component_type != ComponentType.IC:
continue
mpn = comp.mpn or comp.value
mpn = (comp.mpn or "").strip() or (comp.value or "").strip()
if not mpn:
not_reviewed.append({"designator": ref, "reason": "no MPN in BOM"})
if on_progress:
await on_progress(ref, 0, "skipped", "no MPN in BOM")
continue
pdf = _find_pdf(mpn, pdf_dir_path, storage=storage)
if pdf:
ic_tasks.append((ref, str(pdf)))