Reuse datasheet family aliases and skip the LLM for catalog passives.

Store BOM MPN and orderable code on the same PDF blob, derive TI datasheet slugs from package/Q1 codes, and map LCSC/DigiKey R/C/L parameters to specs without a model call.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
2026-08-28 02:11:49 +02:00
co-authored by Cursor
parent 27a3ce8bd0
commit dd46ce1da1
11 changed files with 437 additions and 30 deletions
+3 -1
View File
@@ -776,7 +776,9 @@ async def fetch_auto_datasheet(mpn: str, request: Request, lcsc: str | None = No
if result.source: if result.source:
headers["X-Datasheet-Source"] = result.source headers["X-Datasheet-Source"] = result.source
try: try:
proj_svc.remember_datasheet(get_storage(request), mpn, result.pdf_bytes) proj_svc.remember_datasheet(
get_storage(request), mpn, result.pdf_bytes, extra_mpns=result.alias_mpns,
)
except Exception: except Exception:
pass pass
return Response(content=result.pdf_bytes, media_type="application/pdf", headers=headers) return Response(content=result.pdf_bytes, media_type="application/pdf", headers=headers)
+52 -6
View File
@@ -56,11 +56,19 @@ class DatasheetHit:
error: str | None = None error: str | None = None
url: str | None = None url: str | None = None
source: str | None = None # "lcsc" | "ti" | "digikey" | ... source: str | None = None # "lcsc" | "ti" | "digikey" | ...
catalog_mpn: str | None = None # orderable code that actually matched
@property @property
def ok(self) -> bool: def ok(self) -> bool:
return self.pdf_bytes is not None return self.pdf_bytes is not None
@property
def alias_mpns(self) -> list[str]:
extra = (self.catalog_mpn or "").strip()
if extra and extra.upper() != (self.mpn or "").upper():
return [extra]
return []
def _alnum(mpn: str) -> str: def _alnum(mpn: str) -> str:
return re.sub(r"[^A-Z0-9]", "", mpn.upper()) return re.sub(r"[^A-Z0-9]", "", mpn.upper())
@@ -242,8 +250,16 @@ async def _from_lcsc(mpn: str, lcsc_id: str | None) -> DatasheetHit | None:
except Exception as exc: except Exception as exc:
log.info("LCSC PDF download failed for %s (%s): %s", mpn, url, exc) log.info("LCSC PDF download failed for %s (%s): %s", mpn, url, exc)
return DatasheetHit(mpn, error=f"LCSC download failed: {exc}", url=url, source="lcsc") return DatasheetHit(mpn, error=f"LCSC download failed: {exc}", url=url, source="lcsc")
catalog = (
product.get("productModel")
or product.get("productName")
or ""
)
log.info("Fetched datasheet for %s via LCSC (%d KB)", mpn, len(pdf) // 1024) log.info("Fetched datasheet for %s via LCSC (%d KB)", mpn, len(pdf) // 1024)
return DatasheetHit(mpn, pdf_bytes=pdf, url=url, source="lcsc") return DatasheetHit(
mpn, pdf_bytes=pdf, url=url, source="lcsc",
catalog_mpn=str(catalog) or None,
)
def _strip_packing_alnum(mpn: str) -> str | None: def _strip_packing_alnum(mpn: str) -> str | None:
@@ -255,17 +271,46 @@ def _strip_packing_alnum(mpn: str) -> str | None:
return None return None
_TI_PACKAGE_SUFFIXES = (
"dbvr", "dbvt", "dbv", "pwr", "pwt", "pw", "rger", "rget", "rge",
"dgsr", "dgsk", "dgs", "ydtr", "ydt", "dcnr", "dcnt", "dcn",
"rgtr", "rgtt", "rgt", "runr", "runt", "dckr", "dckt", "dck",
)
def _ti_slugs(mpn: str) -> list[str]: def _ti_slugs(mpn: str) -> list[str]:
"""Candidate TI datasheet slugs, most specific first.""" """Candidate TI datasheet slugs, most specific first."""
raw = mpn.lower().replace("/", "-").strip("-") raw = mpn.lower().replace("/", "-").strip("-")
slugs = [raw] slugs: list[str] = []
# Longest packing / orderable suffixes first so "sptr" is not clipped to "sp".
def add(value: str) -> None:
value = value.strip("-")
if value and value not in slugs:
slugs.append(value)
add(raw)
for suffix in ("-t/r", "/tr", "-tr", "-reel", "sptr", "ptr", "mtr", "tr"): for suffix in ("-t/r", "/tr", "-tr", "-reel", "sptr", "ptr", "mtr", "tr"):
if raw.endswith(suffix) and len(raw) > len(suffix) + 3: if raw.endswith(suffix) and len(raw) > len(suffix) + 3:
base = raw[: -len(suffix)].rstrip("-") add(raw[: -len(suffix)].rstrip("-"))
if base and base not in slugs:
slugs.append(base)
break break
for pkg in _TI_PACKAGE_SUFFIXES:
if raw.endswith(pkg) and len(raw) > len(pkg) + 4:
add(raw[: -len(pkg)])
break
# Orderable INA228AQDGSRQ1 → datasheet ina228-q1 / ina228
prefixes = sorted(_TI_PREFIXES, key=len, reverse=True)
for prefix in prefixes:
if not raw.startswith(prefix):
continue
rest = raw[len(prefix):]
m = re.match(r"(\d{2,})", rest)
if not m:
break
family = f"{prefix}{m.group(1)}"
add(family)
if "q1" in raw:
add(f"{family}-q1")
break
return slugs return slugs
@@ -303,6 +348,7 @@ async def _from_digikey(mpn: str) -> DatasheetHit | None:
if result.ok: if result.ok:
return DatasheetHit( return DatasheetHit(
mpn, pdf_bytes=result.pdf_bytes, url=result.url, source="digikey", mpn, pdf_bytes=result.pdf_bytes, url=result.url, source="digikey",
catalog_mpn=getattr(result, "catalog_mpn", None),
) )
return DatasheetHit(mpn, error=result.error, url=result.url, source="digikey") return DatasheetHit(mpn, error=result.error, url=result.url, source="digikey")
+78 -9
View File
@@ -21,6 +21,7 @@ from backend.services.storage import StorageBackend
BLOB_PREFIX = "library/datasheets/blobs/" BLOB_PREFIX = "library/datasheets/blobs/"
REF_PREFIX = "library/datasheets/refs/" REF_PREFIX = "library/datasheets/refs/"
ALIAS_KEY = "library/datasheets/aliases.json"
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -81,28 +82,96 @@ def store_datasheet_bytes(
storage: StorageBackend, storage: StorageBackend,
data: bytes, data: bytes,
mpn: str, mpn: str,
extra_mpns: list[str] | None = None,
) -> str: ) -> str:
"""Same as :func:`store_datasheet` but from in-memory bytes.""" """Same as :func:`store_datasheet` but from in-memory bytes.
``extra_mpns`` are additional catalog/orderable codes that should point
at the same blob (family MPN vs ``…-N16R16V``).
"""
md5 = compute_md5_from_bytes(data) md5 = compute_md5_from_bytes(data)
bk = blob_key(md5) bk = blob_key(md5)
if not storage.exists(bk): if not storage.exists(bk):
storage.write_bytes(bk, data) storage.write_bytes(bk, data)
storage.write_json(ref_key(mpn), {"hash": md5, "blob_key": bk, "mpn": mpn}) names = [mpn, *(extra_mpns or [])]
seen: set[str] = set()
for name in names:
name = (name or "").strip()
if not name:
continue
key = name.upper()
if key in seen:
continue
seen.add(key)
storage.write_json(ref_key(name), {"hash": md5, "blob_key": bk, "mpn": name})
_record_aliases(storage, mpn, extra_mpns or [])
return bk return bk
def _record_aliases(storage: StorageBackend, mpn: str, extra_mpns: list[str]) -> None:
from backend.services.datasheet_finder import _alnum, _MIN_FAMILY_LEN
names = [mpn, *extra_mpns]
compact = {n: _alnum(n) for n in names if n and n.strip()}
if len(set(compact.values())) < 2 and not extra_mpns:
return
table: dict[str, str] = {}
if storage.exists(ALIAS_KEY):
raw = storage.read_json(ALIAS_KEY)
table = dict(raw.get("aliases") or {})
canonical = extra_mpns[0].strip() if extra_mpns else mpn
for name, key in compact.items():
if len(key) >= _MIN_FAMILY_LEN:
table[key] = canonical
storage.write_json(ALIAS_KEY, {"aliases": table})
def resolve_datasheet(storage: StorageBackend, mpn: str) -> str | None: def resolve_datasheet(storage: StorageBackend, mpn: str) -> str | None:
"""Look up the blob key for an MPN via its ref file. """Look up the blob key for an MPN via its ref file.
Returns the blob key if the ref exists *and* the blob exists, else None. Tries spelling variants, then the shared alias table (family MPN →
orderable code stored in the library).
""" """
rk = ref_key(mpn) from backend.services.datasheet_finder import (
if not storage.exists(rk): _MIN_FAMILY_LEN,
_alnum,
mpn_query_variants,
)
def _from_ref(name: str) -> str | None:
rk = ref_key(name)
if not storage.exists(rk):
return None
ref = storage.read_json(rk)
bk = ref.get("blob_key")
if bk and storage.exists(bk):
return bk
return None return None
ref = storage.read_json(rk)
bk = ref.get("blob_key") for name in mpn_query_variants(mpn) or [mpn]:
if bk and storage.exists(bk): hit = _from_ref(name)
return bk if hit:
return hit
if not storage.exists(ALIAS_KEY):
return None
table = (storage.read_json(ALIAS_KEY) or {}).get("aliases") or {}
want = _alnum(mpn)
if not want:
return None
target = table.get(want)
if target:
hit = _from_ref(target)
if hit:
return hit
if len(want) >= _MIN_FAMILY_LEN:
for key, target in table.items():
if key.startswith(want) or (
want.startswith(key) and len(key) >= _MIN_FAMILY_LEN
):
hit = _from_ref(target)
if hit:
return hit
return None return None
+20 -11
View File
@@ -145,8 +145,8 @@ def _find_product_one(mpn: str, products: list[dict]) -> dict | None:
return exact or loose or family return exact or loose or family
async def _search_mpn(mpn: str) -> str | None: async def _search_mpn(mpn: str) -> tuple[str | None, str | None]:
"""Search DigiKey for an MPN and return the primary datasheet URL, or None.""" """Search DigiKey; return (datasheet_url, catalog_mpn)."""
tried: set[str] = set() tried: set[str] = set()
for keyword in mpn_query_variants(mpn): for keyword in mpn_query_variants(mpn):
key = keyword.upper() key = keyword.upper()
@@ -159,8 +159,8 @@ async def _search_mpn(mpn: str) -> str | None:
continue continue
url = _get_ds_url(product) url = _get_ds_url(product)
if url: if url:
return url return url, _get_mpn(product) or None
return None return None, None
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -205,11 +205,13 @@ class DatasheetFetchResult:
pdf_bytes: bytes | None = None, pdf_bytes: bytes | None = None,
error: str | None = None, error: str | None = None,
url: str | None = None, url: str | None = None,
catalog_mpn: str | None = None,
): ):
self.mpn = mpn self.mpn = mpn
self.pdf_bytes = pdf_bytes self.pdf_bytes = pdf_bytes
self.error = error self.error = error
self.url = url # DigiKey datasheet URL (present even when PDF download fails) self.url = url
self.catalog_mpn = catalog_mpn
@property @property
def ok(self) -> bool: def ok(self) -> bool:
@@ -228,7 +230,7 @@ async def fetch_datasheet(mpn: str) -> DatasheetFetchResult:
return DatasheetFetchResult(mpn, error="DigiKey API not configured") return DatasheetFetchResult(mpn, error="DigiKey API not configured")
try: try:
url = await _search_mpn(mpn) url, catalog_mpn = await _search_mpn(mpn)
except httpx.HTTPStatusError as e: except httpx.HTTPStatusError as e:
logger.warning("DigiKey search failed for %s: %s", mpn, e) logger.warning("DigiKey search failed for %s: %s", mpn, e)
return DatasheetFetchResult(mpn, error=f"DigiKey search failed ({e.response.status_code})") return DatasheetFetchResult(mpn, error=f"DigiKey search failed ({e.response.status_code})")
@@ -244,20 +246,27 @@ async def fetch_datasheet(mpn: str) -> DatasheetFetchResult:
pdf_bytes = await _download_pdf(url) pdf_bytes = await _download_pdf(url)
except httpx.HTTPStatusError as e: except httpx.HTTPStatusError as e:
logger.warning("Datasheet download blocked for %s (%s): %s", mpn, url, e) logger.warning("Datasheet download blocked for %s (%s): %s", mpn, url, e)
return DatasheetFetchResult(mpn, error=f"Download blocked ({e.response.status_code})", url=url) return DatasheetFetchResult(
mpn, error=f"Download blocked ({e.response.status_code})", url=url,
catalog_mpn=catalog_mpn,
)
except ValueError as e: except ValueError as e:
logger.warning("Invalid PDF for %s (%s): %s", mpn, url, e) logger.warning("Invalid PDF for %s (%s): %s", mpn, url, e)
return DatasheetFetchResult(mpn, error=str(e), url=url) return DatasheetFetchResult(mpn, error=str(e), url=url, catalog_mpn=catalog_mpn)
except httpx.TimeoutException: except httpx.TimeoutException:
logger.warning("Datasheet download timed out for %s (%s)", mpn, url) logger.warning("Datasheet download timed out for %s (%s)", mpn, url)
return DatasheetFetchResult(mpn, error="Download timed out", url=url) return DatasheetFetchResult(mpn, error="Download timed out", url=url, catalog_mpn=catalog_mpn)
except Exception as e: except Exception as e:
msg = str(e) or type(e).__name__ msg = str(e) or type(e).__name__
logger.warning("Datasheet download failed for %s (%s): %s", mpn, url, msg) logger.warning("Datasheet download failed for %s (%s): %s", mpn, url, msg)
return DatasheetFetchResult(mpn, error=f"Download failed: {msg}", url=url) return DatasheetFetchResult(
mpn, error=f"Download failed: {msg}", url=url, catalog_mpn=catalog_mpn,
)
logger.info("Fetched datasheet for %s (%d KB)", mpn, len(pdf_bytes) // 1024) logger.info("Fetched datasheet for %s (%d KB)", mpn, len(pdf_bytes) // 1024)
return DatasheetFetchResult(mpn, pdf_bytes=pdf_bytes, url=url) return DatasheetFetchResult(
mpn, pdf_bytes=pdf_bytes, url=url, catalog_mpn=catalog_mpn,
)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
+16
View File
@@ -883,6 +883,22 @@ async def auto_resolve_specs(
""" """
tax_dir = taxonomy_dir or settings.taxonomy_dir tax_dir = taxonomy_dir or settings.taxonomy_dir
from backend.services.passive_from_distributor import specs_from_distributor
if component_type == "passive":
direct = specs_from_distributor(
mpn=mpn,
params=digikey_params,
category=digikey_category,
description=digikey_description,
)
if direct is not None:
import logging as _logging
_logging.getLogger(__name__).info(
"Auto-resolved %s from distributor params (no LLM)", mpn,
)
return direct
# Auto-generate type-level specs if none exist # Auto-generate type-level specs if none exist
if not has_specs(component_type, tax_dir): if not has_specs(component_type, tax_dir):
try: try:
@@ -0,0 +1,199 @@
"""Map distributor parameters / LCSC descriptions to typed passive specs.
Used so capacitor / resistor / inductor rows skip the LLM when the catalog
already states value, voltage, tolerance, and package.
"""
from __future__ import annotations
import re
from backend.pinscopex.models import ComponentModel, SimpleComponentSpecs
from backend.pinscopex.resolve_passives import simple_to_typed_passive_specs
_CAP = re.compile(
r"(?P<num>\d+(?:\.\d+)?)\s*(?P<mul>[pnuμµmk])?\s*[fF]\b",
)
_RES = re.compile(
r"(?P<num>\d+(?:\.\d+)?)\s*(?P<mul>[pnuμµmkM])?\s*(?:ohms?|Ω|R)\b",
re.IGNORECASE,
)
_IND = re.compile(
r"(?P<num>\d+(?:\.\d+)?)\s*(?P<mul>[pnuμµmk])?\s*H\b",
)
_TOL = re.compile(r"±\s*(?P<num>\d+(?:\.\d+)?)\s*%")
_VOLT = re.compile(r"(?P<num>\d+(?:\.\d+)?)\s*V\b")
_PKG = re.compile(r"\b(?P<pkg>0201|0402|0603|0805|1206|1210|1812|2220|2512)\b")
_DIEL = re.compile(r"\b(?P<diel>C0G|NP0|X5R|X6S|X7R|X7S|X8R|Y5V|Z5U)\b", re.I)
_MUL = {
"p": 1e-12, "n": 1e-9, "u": 1e-6, "μ": 1e-6, "µ": 1e-6,
"m": 1e-3, "k": 1e3, "K": 1e3, "M": 1e6,
}
def _spice(num: str, mul: str | None, unit: str) -> str:
n = float(num)
factor = _MUL.get((mul or ""), 1.0)
value = n * factor
if unit == "F":
if value >= 1e-6:
return f"{value * 1e6:g}uF"
if value >= 1e-9:
return f"{value * 1e9:g}nF"
return f"{value * 1e12:g}pF"
if unit == "ohm":
if value >= 1e6:
return f"{value / 1e6:g}Mohm"
if value >= 1e3:
return f"{value / 1e3:g}kohm"
return f"{value:g}ohm"
if unit == "H":
if value >= 1e-3:
return f"{value * 1e3:g}mH"
if value >= 1e-6:
return f"{value * 1e6:g}uH"
return f"{value * 1e9:g}nH"
return f"{value:g}{unit}"
def _param_map(params: list[dict[str, str]]) -> dict[str, str]:
out: dict[str, str] = {}
for p in params:
name = (p.get("name") or "").strip().lower()
value = (p.get("value") or "").strip()
if name and value:
out[name] = value
return out
def _first(pmap: dict[str, str], *needles: str) -> str | None:
for key, val in pmap.items():
for needle in needles:
if needle in key:
return val
return None
def _classify(category: str, description: str, pmap: dict[str, str]) -> str | None:
blob = f"{category} {description} {' '.join(pmap.values())}".lower()
if "ferrite" in blob or "bead" in blob:
return None # typed model wants henries; leave to the LLM
if "capacitor" in blob or "mlcc" in blob or "ceramic" in blob:
diel = _DIEL.search(description) or _DIEL.search(" ".join(pmap.values()))
if diel or "ceramic" in blob or "mlcc" in blob:
return "passive.capacitor.ceramic"
if "tantalum" in blob:
return "passive.capacitor.tantalum"
if "electrolytic" in blob or "aluminum" in blob:
return "passive.capacitor.electrolytic"
return "passive.capacitor.ceramic"
if "resistor" in blob:
if "thin film" in blob:
return "passive.resistor.thin_film"
if "thick film" in blob:
return "passive.resistor.thick_film"
return "passive.resistor"
if "inductor" in blob or "choke" in blob:
return "passive.inductor"
return None
def specs_from_distributor(
*,
mpn: str,
params: list[dict[str, str]],
category: str,
description: str,
) -> ComponentModel | None:
"""Return a ComponentModel when value + type can be parsed without an LLM."""
pmap = _param_map(params)
text = " ".join(
[description, category, *pmap.values()],
)
subtype = _classify(category, description, pmap)
if not subtype:
# Infer from parsed units in the description alone
if _CAP.search(text) and not _RES.search(text):
subtype = "passive.capacitor.ceramic"
elif _RES.search(text) and "capacitor" not in text.lower():
subtype = "passive.resistor"
elif _IND.search(text):
subtype = "passive.inductor"
else:
return None
values: dict[str, str] = {}
cap = _first(pmap, "capacitance") or (
_CAP.search(text).group(0) if _CAP.search(text) else None
)
res = _first(pmap, "resistance") or (
_RES.search(text).group(0) if _RES.search(text) else None
)
ind = _first(pmap, "inductance") or (
_IND.search(text).group(0) if _IND.search(text) else None
)
if subtype.startswith("passive.capacitor"):
if not cap:
return None
m = _CAP.search(cap) or _CAP.search(text)
if not m:
return None
values["value_farads"] = _spice(m.group("num"), m.group("mul"), "F")
values["value_formatted"] = values["value_farads"]
volt = _first(pmap, "voltage") or (
f"{_VOLT.search(text).group('num')}V" if _VOLT.search(text) else None
)
if volt:
values["voltage_rating_v"] = volt if volt.lower().endswith("v") else f"{volt}V"
diel = _first(pmap, "temperature coefficient", "dielectric")
dm = _DIEL.search(diel or "") or _DIEL.search(text)
if dm:
values["dielectric"] = dm.group("diel").upper().replace("NP0", "C0G")
elif subtype.startswith("passive.resistor"):
if not res:
return None
m = _RES.search(res) or _RES.search(text)
if not m:
return None
values["value_ohms"] = _spice(m.group("num"), m.group("mul"), "ohm")
values["value_formatted"] = values["value_ohms"]
power = _first(pmap, "power")
if power:
values["power_rating_w"] = power
elif subtype.startswith("passive.inductor"):
if not ind:
return None
m = _IND.search(ind) or _IND.search(text)
if not m:
return None
values["value_henries"] = _spice(m.group("num"), m.group("mul"), "H")
values["value_formatted"] = values["value_henries"]
else:
return None
tol = _first(pmap, "tolerance")
tm = _TOL.search(tol or "") or _TOL.search(text)
if tm:
values["tolerance"] = f"±{tm.group('num')}%"
elif tol:
values["tolerance"] = tol
pkg = _first(pmap, "package", "case", "size")
pm = _PKG.search(pkg or "") or _PKG.search(text)
if pm:
values["package"] = pm.group("pkg")
elif pkg and len(pkg) <= 12:
values["package"] = pkg
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)
+3 -1
View File
@@ -704,7 +704,9 @@ async def _ensure_local_datasheet(
except Exception: except Exception:
logger.exception("Failed to persist auto-fetched datasheet for %s", mpn) logger.exception("Failed to persist auto-fetched datasheet for %s", mpn)
try: try:
store_datasheet_bytes(ctx.storage, hit.pdf_bytes, mpn) store_datasheet_bytes(
ctx.storage, hit.pdf_bytes, mpn, extra_mpns=hit.alias_mpns,
)
except Exception: except Exception:
logger.exception("Failed to library-store auto-fetched datasheet for %s", mpn) logger.exception("Failed to library-store auto-fetched datasheet for %s", mpn)
logger.info( logger.info(
+4 -2
View File
@@ -636,12 +636,14 @@ def save_datasheet(
return key return key
def remember_datasheet(storage: StorageBackend, mpn: str, data: bytes) -> None: def remember_datasheet(
storage: StorageBackend, mpn: str, data: bytes, extra_mpns: list[str] | None = None,
) -> None:
"""Write a datasheet into the shared library without failing the caller.""" """Write a datasheet into the shared library without failing the caller."""
try: try:
from backend.services.datasheet_store import store_datasheet_bytes from backend.services.datasheet_store import store_datasheet_bytes
store_datasheet_bytes(storage, data, mpn) store_datasheet_bytes(storage, data, mpn, extra_mpns=extra_mpns)
except Exception: except Exception:
log.exception("Failed to store datasheet for %s in the shared library", mpn) log.exception("Failed to store datasheet for %s in the shared library", mpn)
+8
View File
@@ -92,6 +92,14 @@ def test_ti_slugs_include_family():
assert "mspm0g3507s" not in slugs assert "mspm0g3507s" not in slugs
def test_ti_slugs_from_orderable_code():
slugs = _ti_slugs("INA228AQDGSRQ1")
assert "ina228-q1" in slugs
assert "ina228" in slugs
slugs = _ti_slugs("SN74AXC1T45DBVR")
assert "sn74axc1t45" in slugs
def test_find_datasheet_uses_lcsc_then_skips_empty(monkeypatch): def test_find_datasheet_uses_lcsc_then_skips_empty(monkeypatch):
async def fake_lcsc(mpn, lcsc_id=None): async def fake_lcsc(mpn, lcsc_id=None):
return DatasheetHit( return DatasheetHit(
+12
View File
@@ -19,6 +19,18 @@ def _client(tmp_path) -> TestClient:
return TestClient(app) return TestClient(app)
def test_library_alias_resolves_family_mpn(storage):
from backend.services.datasheet_store import resolve_datasheet, store_datasheet_bytes
store_datasheet_bytes(
storage, PDF, "ESP32-S31-WROOM-3",
extra_mpns=["ESP32-S31-WROOM-3-N16R16V"],
)
assert resolve_datasheet(storage, "ESP32-S31-WROOM-3")
assert resolve_datasheet(storage, "ESP32-S31-WROOM-3-N16R16V")
assert proj_svc.library_has_datasheet(storage, "ESP32-S31-WROOM-3")
def test_save_datasheet_also_stores_in_library(storage): def test_save_datasheet_also_stores_in_library(storage):
meta = proj_svc.create_project(storage, "local", "board") meta = proj_svc.create_project(storage, "local", "board")
key = proj_svc.save_datasheet(storage, "local", meta.id, "CH340E", PDF) key = proj_svc.save_datasheet(storage, "local", meta.id, "CH340E", PDF)
+42
View File
@@ -0,0 +1,42 @@
from backend.services.passive_from_distributor import specs_from_distributor
def test_lcsc_description_capacitor():
model = specs_from_distributor(
mpn="CL21B225KPFNNNE",
params=[{"name": "Package / Case", "value": "0805"}],
category="Capacitors / Ceramic Capacitors",
description="2.2uF ±10% 10V X7R 0805",
)
assert model is not None
specs = model.specs
assert specs.component_subtype == "passive.capacitor.ceramic"
assert abs(specs.value_farads - 2.2e-6) < 1e-12
assert specs.dielectric == "X7R"
assert specs.package == "0805"
def test_digikey_params_resistor():
model = specs_from_distributor(
mpn="RC0805FR-0710KL",
params=[
{"name": "Resistance", "value": "10 kOhms"},
{"name": "Tolerance", "value": "±1%"},
{"name": "Power (Watts)", "value": "0.125W"},
{"name": "Package / Case", "value": "0805 (2012 Metric)"},
],
category="Resistors / Chip Resistor - Surface Mount",
description="RES SMD 10K OHM 1% 1/8W 0805",
)
assert model is not None
assert model.specs.value_ohms == 10000
assert model.specs.package == "0805"
def test_skips_when_no_value():
assert specs_from_distributor(
mpn="MYSTERY",
params=[{"name": "Manufacturer", "value": "Murata"}],
category="",
description="some module",
) is None