Rewrite leftover analysis modules imported by src.

validate/validation_tools alias native lookup and review tools. Passive
resolve, derating, BOM summary, pin-mux, LED current, and net-function
tokens are original Periscope code. finding_engine and pcb_* unchanged.
This commit is contained in:
2026-09-20 19:14:20 +02:00
parent 3d7fa21eb6
commit 82ffa755cd
9 changed files with 1037 additions and 0 deletions
@@ -0,0 +1,60 @@
"""BOM summary rows grouped by MPN. Collation only."""
from __future__ import annotations
from backend.periscopex.models import ComponentType, DesignGraph
from backend.periscopex.utils import natural_sort_key
def build_bom_summary(
graph: DesignGraph,
datasheet_mpns: set[str] | None = None,
descriptions: dict[str, str] | None = None,
) -> list[dict]:
by_key: dict[str, list] = {}
for comp in graph.components.values():
key = comp.mpn if comp.mpn else f"__no_mpn__{comp.value}__{comp.component_type}"
by_key.setdefault(key, []).append(comp)
rows: list[dict] = []
for comps in by_key.values():
first = comps[0]
designators = sorted([c.reference for c in comps], key=natural_sort_key)
specs_dict = None
if first.specs:
if hasattr(first.specs, "values"):
raw = {k: v for k, v in first.specs.values.items() if v is not None}
else:
raw = first.specs.model_dump(exclude={"specs_type"})
raw = {
k: v for k, v in raw.items()
if v is not None and k not in (
"value_ohms", "value_farads", "value_henries", "impedance_ohm",
)
}
specs_dict = raw or None
has_ds = bool(first.mpn and datasheet_mpns is not None and first.mpn in datasheet_mpns)
description = None
if descriptions is not None and first.mpn and first.component_type == ComponentType.IC:
description = descriptions.get(first.mpn)
rows.append({
"mpn": first.mpn,
"designators": designators,
"value": first.value,
"category": first.component_subtype,
"specs": specs_dict,
"description": description,
"has_datasheet": has_ds,
})
def sort_key(row: dict) -> tuple:
cat = row["category"] or ""
if cat.startswith("ic"):
group = 0
elif cat.startswith("passive"):
group = 1
else:
group = 2
return (group, cat, row["mpn"] or "")
rows.sort(key=sort_key)
return rows
@@ -0,0 +1,151 @@
"""Capacitor voltage derating + DC-bias C_eff stima (not a vendor lot curve)."""
from __future__ import annotations
import re
from backend.periscopex.models import ComponentType, DesignGraph, NetType
from backend.periscopex.resolve_passives import _format_value
from backend.periscopex.utils import natural_sort_key
_CERAMIC = {"X7R", "X5R", "C0G", "NP0", "Y5V", "X7S", "X6S", "X8R", "C0G (NP0)"}
_CURVES: dict[str, list[tuple[float, float]]] = {
"c0g": [(0.0, 1.0), (1.2, 1.0)],
"x7r": [(0.0, 1.0), (0.25, 0.90), (0.50, 0.70), (0.75, 0.45), (1.0, 0.30), (1.2, 0.22)],
"x5r": [(0.0, 1.0), (0.25, 0.82), (0.50, 0.55), (0.75, 0.32), (1.0, 0.18), (1.2, 0.12)],
"y5v": [(0.0, 1.0), (0.25, 0.50), (0.50, 0.20), (0.80, 0.12), (1.0, 0.10)],
}
def _lerp(curve: list[tuple[float, float]], x: float) -> float:
if x <= curve[0][0]:
return curve[0][1]
for (x0, y0), (x1, y1) in zip(curve, curve[1:]):
if x <= x1:
if x1 == x0:
return y1
return y0 + (y1 - y0) * (x - x0) / (x1 - x0)
return curve[-1][1]
def _bias_family(dielectric: str | None) -> str | None:
if not dielectric:
return None
u = dielectric.upper()
if "C0G" in u or "NP0" in u or "NPO" in u:
return "c0g"
if "Y5V" in u:
return "y5v"
if "X5R" in u or "X6S" in u:
return "x5r"
if "X7R" in u or "X7S" in u or "X8R" in u:
return "x7r"
return None
def dc_bias_remaining(
dielectric: str | None, v_op: float | None, rated_v: float | None,
) -> float | None:
family = _bias_family(dielectric)
if family is None or v_op is None or rated_v is None or rated_v <= 0:
return None
return _lerp(_CURVES[family], max(0.0, v_op) / rated_v)
def _parse_voltage_rating(s: str | None) -> float | None:
if not s:
return None
m = re.match(r"([\d.]+)", s)
return float(m.group(1)) if m else None
def _dielectric_category(subtype: str | None, dielectric: str | None) -> str | None:
if subtype:
low = subtype.lower()
if "tantalum" in low:
return "tantalum"
if "electrolytic" in low:
return "electrolytic"
if "ceramic" in low:
return "ceramic"
if dielectric:
upper = dielectric.upper().strip()
if upper in _CERAMIC or any(d in upper for d in _CERAMIC):
return "ceramic"
low = dielectric.lower()
if "tantalum" in low or low == "ta":
return "tantalum"
if "electrolytic" in low or low == "al":
return "electrolytic"
return "ceramic"
def _stress(op: float | None, rated: float | None) -> str:
if op is None or rated is None or rated <= 0:
return "UNKNOWN"
ratio = op / rated
if ratio > 1.0:
return "RISK"
if ratio > 0.8:
return "MARGIN"
return "PASS"
def build_derating_table(graph: DesignGraph) -> list[dict]:
rows: list[dict] = []
for comp in graph.components.values():
if comp.component_type != ComponentType.CAPACITOR:
continue
rated_v = value_fmt = dielectric = c_nom = None
if comp.specs and hasattr(comp.specs, "voltage_rating_v"):
rated_v = _parse_voltage_rating(comp.specs.voltage_rating_v)
value_fmt = getattr(comp.specs, "value_formatted", None)
dielectric = getattr(comp.specs, "dielectric", None)
c_nom = getattr(comp.specs, "value_farads", None)
op_voltage: float | None = None
op_source: str | None = None
for net_name in comp.pins.values():
net = graph.nets.get(net_name)
if net and net.voltage is not None and net.voltage > 0:
if op_voltage is None or net.voltage > op_voltage:
op_voltage = net.voltage
op_source = net_name
seen: set[str] = set()
connected: list[tuple[str, float | None, NetType | None]] = []
for net_name in comp.pins.values():
if net_name in seen:
continue
seen.add(net_name)
net = graph.nets.get(net_name)
connected.append((net_name, net.voltage if net else None, net.net_type if net else None))
net_plus = net_minus = None
if len(connected) == 1:
net_plus = connected[0][0]
elif len(connected) >= 2:
by_v = sorted(connected, key=lambda c: (
c[2] != NetType.GROUND, c[1] is not None, c[1] or 0,
))
net_minus = by_v[0][0]
net_plus = by_v[-1][0]
factor = dc_bias_remaining(dielectric, op_voltage, rated_v)
c_eff = (c_nom * factor) if (c_nom is not None and factor is not None) else None
rows.append({
"designator": comp.reference,
"mpn": comp.mpn,
"value_formatted": value_fmt,
"rated_voltage_v": rated_v,
"operating_voltage_v": op_voltage,
"operating_voltage_source": op_source,
"net_plus": net_plus,
"net_minus": net_minus,
"dielectric_category": _dielectric_category(comp.component_subtype, dielectric),
"dielectric": dielectric,
"c_nominal_f": c_nom,
"dc_bias_factor": factor,
"c_eff_f": c_eff,
"c_eff_formatted": _format_value(c_eff, "F") if c_eff is not None else None,
"dc_bias_model": "stima" if factor is not None else None,
"stress": _stress(op_voltage, rated_v),
})
rows.sort(key=lambda r: natural_sort_key(r["designator"]))
return rows
@@ -0,0 +1,264 @@
"""LED I_f check: Ohm's law on the graph vs datasheet channel rating."""
from __future__ import annotations
import re
from backend.periscopex.models import ComponentType, DesignGraph, Finding, NetType
from backend.periscopex.resolve_passives import _parse_spice_value
_COLOR = {
"R": "red", "RED": "red",
"G": "green", "GRN": "green", "GREEN": "green",
"B": "blue", "BLU": "blue", "BLUE": "blue",
}
def _num(v: object) -> float | None:
if v is None:
return None
if isinstance(v, (int, float)):
return float(v)
s = str(v).strip()
for cand in (s, *re.findall(r"[-+]?\d*\.?\d+\s*[a-zA-Zµ]*", s)):
cand = cand.strip()
if not cand:
continue
try:
return _parse_spice_value(cand)
except ValueError:
pass
m = re.match(r"^[-+]?\d*\.?\d+", cand)
if m:
try:
return float(m.group(0))
except ValueError:
pass
return None
def _parse_resistance(v: object) -> float | None:
if v is None:
return None
if isinstance(v, (int, float)):
return float(v)
t = str(v).strip().upper().replace("OHMS", "").replace("OHM", "").replace("Ω", "").replace(" ", "")
if not t:
return None
mult = {"R": 1.0, "K": 1e3, "M": 1e6, "G": 1e9}
m = re.match(r"^(\d+)([RKMG])(\d+)$", t)
if m:
return (float(m.group(1)) + float(f"0.{m.group(3)}")) * mult[m.group(2)]
m = re.match(r"^(\d*\.?\d+)([RKMG])$", t)
if m:
return float(m.group(1)) * mult[m.group(2)]
try:
return float(t)
except ValueError:
return None
def _spec(values: dict, *keys: str) -> float | None:
for k in keys:
if k in values:
n = _num(values[k])
if n is not None:
return n
return None
def _imax(values: dict) -> float | None:
i = _spec(values, "forward_current_per_channel_a", "forward_current_a",
"max_forward_current_a", "if_max_a")
if i is None:
return None
if i >= 1.0:
i = i / 1000.0
return i
def _vf(values: dict, color: str | None) -> float | None:
vf = _spec(values, f"forward_voltage_{color}_v") if color else None
if vf is None:
vf = _spec(values, "forward_voltage_v", "vf_v")
if vf is None:
cands = [c for c in (_spec(values, f"forward_voltage_{c}_v") for c in ("red", "green", "blue")) if c is not None]
vf = min(cands) if cands else None
if vf is not None and vf > 20:
vf = vf / 1000.0
return vf
def _net_voltage(graph: DesignGraph, net_name: str | None) -> float | None:
if not net_name:
return None
net = graph.nets.get(net_name)
return net.voltage if net else None
def _is_rail_net(graph: DesignGraph, net_name: str) -> bool:
net = graph.nets.get(net_name)
if not net:
return False
return net.net_type in (NetType.POWER, NetType.GROUND) or net.voltage is not None
def _series_resistor(graph: DesignGraph, net_name: str, exclude_ref: str):
net = graph.nets.get(net_name)
if not net or len(net.pins) != 2:
return None
for pc in net.pins:
if pc.component_ref == exclude_ref:
continue
c = graph.components.get(pc.component_ref)
if not c or c.component_type != ComponentType.RESISTOR:
continue
rval = getattr(c.specs, "value_ohms", None) if c.specs else None
if rval is None:
rval = _parse_resistance(c.value)
if rval is None or rval <= 0:
continue
far = next((n for n in c.pins.values() if n != net_name), None)
return (pc.component_ref, float(rval), far)
return None
def _leg_to_ic(graph: DesignGraph, net_name: str, exclude_ref: str) -> bool:
for r in graph.components_on_net(net_name):
if r == exclude_ref:
continue
c = graph.components.get(r)
if c and c.component_type == ComponentType.IC:
return True
return False
def _leg_color(pid: str, comp) -> str | None:
if pid.upper() in _COLOR:
return _COLOR[pid.upper()]
specs = comp.specs
pin = specs.pin_by_number(pid) if specs and hasattr(specs, "pin_by_number") else None
if pin:
for tok in re.split(r"[\s_/-]+", pin.name.upper()):
if tok in _COLOR:
return _COLOR[tok]
return None
def check_led_current(graph: DesignGraph) -> list[Finding]:
findings: list[Finding] = []
for ref in sorted(graph.components_by_subtype("discrete.led")):
comp = graph.components.get(ref)
if not comp or not comp.specs:
continue
values = getattr(comp.specs, "values", None)
if not values:
continue
imax = _imax(values)
if imax is None:
continue
finding = _check_led(graph, ref, comp, values, imax)
if finding is not None:
findings.append(finding)
return findings
def _check_led(graph, ref, comp, values, imax) -> Finding | None:
pins = comp.pins
pin_volts = [v for v in (_net_voltage(graph, n) for n in pins.values()) if v is not None]
if len(pins) <= 2:
leg = next(
((pid, net, _series_resistor(graph, net, ref))
for pid, net in pins.items() if _series_resistor(graph, net, ref)),
None,
)
if leg is None:
cand = next(((pid, net) for pid, net in pins.items() if not _is_rail_net(graph, net)), None)
legs = [(cand[0], cand[1], None)] if cand else []
else:
legs = [leg]
else:
legs = [
(pid, net, _series_resistor(graph, net, ref))
for pid, net in pins.items() if not _is_rail_net(graph, net)
]
worst = None
no_res = None
for pid, net, res in legs:
color = _leg_color(pid, comp)
vf = _vf(values, color)
cand = list(pin_volts)
if res and res[2]:
fv = _net_voltage(graph, res[2])
if fv is not None:
cand.append(fv)
vrail = max(cand) if cand else None
if res is None:
if no_res is None and vrail is not None and vrail > 0 and not _leg_to_ic(graph, net, ref):
no_res = (color, net, vrail, vf)
continue
rref, rval, _far = res
if vrail is None or vf is None or vrail <= vf or rval <= 0:
continue
i = (vrail - vf) / rval
if i > imax and (worst is None or i > worst[0]):
worst = (i, color, net, vrail, vf, rval, rref)
if worst is not None:
i, color, net, vrail, vf, rval, rref = worst
return _over(ref, comp, net, color, vrail, vf, rval, rref, imax, i)
if no_res is not None:
color, net, vrail, vf = no_res
return _missing_r(ref, comp, net, color, vrail, vf, imax)
return None
def _chan(color: str | None) -> str:
return f"{color} channel" if color else "LED"
def _over(ref, comp, net, color, vrail, vf, rval, rref, imax, i) -> Finding:
rmin = (vrail - vf) / imax
return Finding(
designator=ref, mpn=comp.mpn or "", aspect="led_current",
source="led_current_check", source_page=None, status="ERROR",
finding=(
f"{ref} {_chan(color)} forward current is ~{i * 1000:.0f} mA, "
f"exceeding its {imax * 1000:.0f} mA forward-current rating."
),
why=(
f"With the supply at {vrail:.1f} V and Vf≈{vf:.1f} V, series resistor "
f"{rref} ({rval:.0f} Ω) on net '{net}' passes "
f"~({vrail:.1f}{vf:.1f})/{rval:.0f} = {i * 1000:.0f} mA (worst case, "
f"0 V driver drop) — above the {imax * 1000:.0f} mA rating."
),
recommendation=(
f"Increase the series resistor to at least {rmin:.0f} Ω to keep the "
f"{_chan(color)} at or below {imax * 1000:.0f} mA."
),
reference=f"{comp.mpn or ref} LED specs",
)
def _missing_r(ref, comp, net, color, vrail, vf, imax) -> Finding:
rec = "Add a series current-limiting resistor, or confirm a constant-current driver."
if vf is not None and vrail > vf:
rec = (
f"Add a series resistor of at least {((vrail - vf) / imax):.0f} Ω "
f"(or confirm a constant-current driver)."
)
return Finding(
designator=ref, mpn=comp.mpn or "", aspect="led_current",
source="led_current_check", source_page=None, status="WARNING",
finding=(
f"Unverified: {ref} {_chan(color)} has no series current-limiting "
f"resistor on net '{net}'."
),
why=(
f"The {_chan(color)} on net '{net}' has no series resistor between the "
f"LED and the {vrail:.1f} V supply. If it is not driven by a "
f"constant-current source, forward current can exceed the "
f"{imax * 1000:.0f} mA rating."
),
recommendation=rec,
reference=f"{comp.mpn or ref} LED specs",
)
@@ -0,0 +1,81 @@
"""Canonical (peripheral, signal) tokens from net names and pin functions."""
from __future__ import annotations
import re
_FAMILIES = (
"LPUART", "USART", "UART", "I2C", "OCTOSPI", "QSPI", "SPI",
"FDCAN", "CAN", "SDMMC", "SDIO", "I2S", "SAI", "USB",
)
_FAMILY_ALT = "|".join(_FAMILIES)
_PERIPHERAL_RE = re.compile(rf"^({_FAMILY_ALT})(\d*)$")
_FUNCTION_RE = re.compile(rf"^({_FAMILY_ALT})(\d*)_(.+)$")
_SIGNALS = {"TX", "RX", "SDA", "SCL", "MOSI", "MISO", "SCK", "NSS", "DP", "DM"}
_SYNONYMS = {
"TXD": "TX", "RXD": "RX",
"SCLK": "SCK", "CLK": "SCK",
"SS": "NSS", "CS": "NSS", "NCS": "NSS", "STE": "NSS",
"DPLUS": "DP", "DMINUS": "DM",
"PICO": "MOSI", "COPI": "MOSI",
"POCI": "MISO", "CIPO": "MISO",
}
_COMPLEMENT = {
"TX": "RX", "RX": "TX",
"SDA": "SCL", "SCL": "SDA",
"MOSI": "MISO", "MISO": "MOSI",
"DP": "DM", "DM": "DP",
}
_CS_INDEXED = re.compile(r"^(N?CS|SS|STE)\d+$")
def _canon_signal(tok: str) -> str | None:
t = tok.upper()
m = _CS_INDEXED.match(t)
if m:
t = m.group(1)
t = _SYNONYMS.get(t, t)
return t if t in _SIGNALS else None
def _tokens(name: str) -> list[str]:
s = name.upper().lstrip("/")
s = s.replace("D+", "DP").replace("D-", "DM")
s = re.sub(r"[._/]", "-", s)
return [p for p in s.split("-") if p]
def parse_net_token(net_name: str) -> tuple[str, str] | None:
parts = _tokens(net_name)
for i in range(len(parts) - 1):
m = _PERIPHERAL_RE.match(parts[i])
if not m:
continue
sig = _canon_signal(parts[i + 1])
if sig is None:
continue
return (m.group(1) + m.group(2), sig)
return None
def normalize_functions(functions: list[str] | None) -> set[tuple[str, str]]:
out: set[tuple[str, str]] = set()
for f in functions or []:
for alt in f.upper().replace("D+", "DP").replace("D-", "DM").split("/"):
m = _FUNCTION_RE.match(alt.strip())
if not m:
continue
sig = _canon_signal(m.group(3))
if sig is None:
continue
out.add((m.group(1) + m.group(2), sig))
return out
def signals_for_peripheral(funcs: set[tuple[str, str]], peripheral: str) -> set[str]:
return {s for (p, s) in funcs if p == peripheral}
def complement(signal: str) -> str | None:
return _COMPLEMENT.get(signal)
@@ -0,0 +1,125 @@
"""Pin-mux feasibility: net-asserted function vs datasheet alt-function table.
Feasibility only. Inter-device same-peripheral links are skipped.
"""
from __future__ import annotations
from backend.periscopex.constraints_lookup import match_constraints as _match_constraints
from backend.periscopex.models import ComponentConstraints, ComponentType, DesignGraph, Finding
from backend.periscopex.pin_function_tokens import (
complement,
normalize_functions,
parse_net_token,
signals_for_peripheral,
)
def check_pin_mux_feasibility(
graph: DesignGraph,
constraints_map: dict[str, ComponentConstraints],
) -> list[Finding]:
findings: 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_num, net_name in comp.pins.items():
token = parse_net_token(net_name)
if token is None:
continue
peripheral, signal = token
pin = cons.pin_by_number(pin_num)
if pin is None or not pin.functions:
continue
exposed = signals_for_peripheral(normalize_functions(pin.functions), peripheral)
if not exposed or signal in exposed:
continue
if _peer_exposes(graph, constraints_map, net_name, ref, peripheral):
continue
findings.append(
_finding(ref, comp.mpn or "", pin_num, pin.name, net_name,
peripheral, signal, exposed, pin.functions)
)
return findings
def _peer_exposes(
graph: DesignGraph,
constraints_map: dict[str, ComponentConstraints],
net_name: str,
self_ref: str,
peripheral: str,
) -> bool:
net = graph.nets.get(net_name)
if not net:
return False
for pc in net.pins:
if pc.component_ref == self_ref:
continue
other = graph.components.get(pc.component_ref)
if not other or other.component_type != ComponentType.IC:
continue
ocons = _match_constraints(other.mpn or other.value, constraints_map)
if not ocons:
continue
opin = ocons.pin_by_number(pc.pin_number)
if opin is None or not opin.functions:
continue
if signals_for_peripheral(normalize_functions(opin.functions), peripheral):
return True
return False
def _finding(
ref: str, mpn: str, pin_num: str, pin_name: str, net_name: str,
peripheral: str, signal: str, exposed: set[str], functions: list[str],
) -> Finding:
listed = ", ".join(functions) if functions else "(none listed)"
other = complement(signal)
swap = bool(other and other in exposed)
rec = (
f"Move '{net_name}' to a pin whose alternate functions include "
f"{peripheral}_{signal}."
)
hint = ""
if swap:
hint = (
f" This pin's {peripheral} role is {peripheral}_{other} — the "
f"complement of {peripheral}_{signal} — so the {signal}/{other} "
f"nets are most likely swapped."
)
rec = (
f"Move '{net_name}' to a {peripheral}_{signal}-capable pin, or swap "
f"it with the paired {peripheral}_{other} net if that resolves both."
)
return Finding(
designator=ref,
mpn=mpn,
aspect="pin_mux",
source="pin_mux_check",
source_page=None,
status="ERROR",
finding=(
f"Net '{net_name}' assigns {ref} pin {pin_num} ({pin_name}) the "
f"{peripheral}_{signal} function, but this pin cannot be muxed as "
f"{peripheral}_{signal}."
),
why=(
f"The intended function {peripheral}_{signal} was inferred from the "
f"net name '{net_name}'. Per the datasheet alternate-function table, "
f"pin {pin_num} ({pin_name}) can be muxed as: {listed}. "
f"{peripheral}_{signal} is not in that list, so the silicon cannot "
f"route it here regardless of downstream wiring." + hint +
f" If '{net_name}' is not actually configured for {peripheral} in "
f"firmware (e.g. bit-banged GPIO, or a label carried over from the "
f"connected part), disregard this finding."
),
recommendation=rec,
reference=f"{mpn or ref} alternate-function table",
net=net_name,
pins=[f"{ref}.{pin_num}"],
rule_id="PE-MUX-001",
)
@@ -0,0 +1,301 @@
"""Passive MPN pattern matching and typed-spec conversion."""
from __future__ import annotations
import json
import re
from collections import defaultdict
from pathlib import Path
from backend.periscopex.models import (
CapacitorSpecs,
ComponentSpecs,
ComponentType,
InductorSpecs,
PassivePattern,
ResistorSpecs,
ResolvedPassive,
SimpleComponentSpecs,
ValueDecoder,
)
from backend.periscopex.parsers import parse_bom
def _multiplier(digit: str, letter_multipliers: dict[str, int | str]) -> float:
if digit in letter_multipliers:
val = letter_multipliers[digit]
if val == "decimal_point":
raise ValueError(f"Letter '{digit}' is a decimal-point marker, not a multiplier")
return 10.0 ** int(val)
return 10.0 ** int(digit)
def _decode_eia3_pf(digits: str) -> float:
return float(int(digits[:2]) * (10 ** int(digits[2])))
def _decode_r_notation(digits: str, decimal_letters: set[str]) -> float | None:
for letter in decimal_letters:
if letter in digits:
return float(digits.replace(letter, "."))
return None
def _decode_eia4_ohm(digits: str, tolerance_code: str, decoder: ValueDecoder) -> float:
if decoder.zero_code and digits == decoder.zero_code:
return 0.0
decimal_letters = {k for k, v in decoder.letter_multipliers.items() if v == "decimal_point"}
if decimal_letters:
r_val = _decode_r_notation(digits, decimal_letters)
if r_val is not None:
return r_val
cond = decoder.conditional_on or {}
high_tol = cond.get("high_tolerance", [])
layout = cond.get("high_tolerance_layout", {}) if tolerance_code in high_tol else cond.get("low_tolerance_layout", {})
sig_start = layout.get("significant_start", 0)
sig_count = layout.get("significant_count", 3)
mult_idx = layout.get("multiplier_index", 3)
sig = int(digits[sig_start:sig_start + sig_count])
return float(sig) * _multiplier(digits[mult_idx], decoder.letter_multipliers)
def _decode_letter_decimal(digits: str, decoder: ValueDecoder) -> float:
for letter, mult in decoder.letter_multipliers.items():
if letter in digits:
before, after = digits.split(letter, 1)
value = float(f"{before}.{after}") if after else float(before)
return value * float(mult)
return float(digits)
def decode_value(digits: str, decoder: ValueDecoder, tolerance_code: str | None = None) -> float:
if decoder.type == "eia3_pf":
pf = _decode_eia3_pf(digits)
return pf * 1e-12 if decoder.output_unit == "F" else pf
if decoder.type == "eia4_ohm_conditional":
return _decode_eia4_ohm(digits, tolerance_code or "", decoder)
if decoder.type == "letter_decimal_ohm":
return _decode_letter_decimal(digits, decoder)
raise ValueError(f"Unknown decoder type: {decoder.type}")
_SI_OHM = [(1e6, "Mohm"), (1e3, "kohm"), (1.0, "ohm"), (1e-3, "mohm")]
_SI_F = [(1e-3, "mF"), (1e-6, "uF"), (1e-9, "nF"), (1e-12, "pF"), (1e-15, "fF")]
def _format_value(value: float, unit: str) -> str:
if value == 0.0:
return f"0 {unit}"
prefixes = _SI_OHM if unit == "ohm" else _SI_F
for threshold, label in prefixes:
if abs(value) >= threshold * 0.999:
scaled = value / threshold
if scaled == int(scaled):
return f"{int(scaled)} {label}"
return f"{scaled:.2f}".rstrip("0").rstrip(".") + f" {label}"
return f"{value} {unit}"
def resolved_to_specs(resolved: ResolvedPassive) -> ComponentSpecs:
if resolved.component_type == ComponentType.RESISTOR:
return ResistorSpecs(
value_ohms=resolved.value, value_formatted=resolved.value_formatted,
tolerance=resolved.tolerance, package=resolved.package,
power_rating_w=resolved.power_rating,
)
if resolved.component_type == ComponentType.CAPACITOR:
return CapacitorSpecs(
value_farads=resolved.value, value_formatted=resolved.value_formatted,
tolerance=resolved.tolerance, package=resolved.package,
voltage_rating_v=resolved.voltage_rating, dielectric=resolved.dielectric,
)
if resolved.component_type == ComponentType.INDUCTOR:
return InductorSpecs(
value_henries=resolved.value, value_formatted=resolved.value_formatted,
tolerance=resolved.tolerance, package=resolved.package,
)
raise ValueError(f"Unsupported component type: {resolved.component_type}")
_SPICE = {"T": 1e12, "G": 1e9, "M": 1e6, "k": 1e3, "m": 1e-3, "u": 1e-6, "n": 1e-9, "p": 1e-12}
_UNITS = ("ohm", "F", "H", "V", "W", "A", "Hz")
def _parse_spice_value(s: str) -> float:
s = s.strip()
for sep in (" at ", " @ ", "@"):
idx = s.find(sep)
if idx > 0:
s = s[:idx].strip()
break
for suffix in _UNITS:
if s.endswith(suffix):
s = s[: -len(suffix)]
break
try:
return float(s)
except ValueError:
pass
for i in range(len(s) - 1, -1, -1):
ch = s[i]
if ch in _SPICE:
return float(s[:i] + s[i + 1 :]) * _SPICE[ch]
raise ValueError(f"Cannot parse SPICE value: {s!r}")
def simple_to_typed_passive_specs(simple: SimpleComponentSpecs) -> ComponentSpecs:
subtype = simple.component_subtype or ""
vals = simple.values
value_formatted = str(vals.get("value_formatted") or "")
tolerance = str(vals.get("tolerance")) if vals.get("tolerance") else None
package = str(vals.get("package")) if vals.get("package") else None
st = subtype or None
if subtype.startswith("passive.resistor") or subtype == "passive.resistor":
raw = vals.get("value_ohms")
if raw is None:
raise ValueError("Missing value_ohms in auto-resolved resistor specs")
ohms = _parse_spice_value(str(raw)) if isinstance(raw, str) else float(raw)
return ResistorSpecs(
component_subtype=st, value_ohms=ohms,
value_formatted=value_formatted or _format_value(ohms, "ohm"),
tolerance=tolerance, package=package,
power_rating_w=str(vals.get("power_rating_w")) if vals.get("power_rating_w") else None,
)
if subtype.startswith("passive.capacitor"):
raw = vals.get("value_farads")
if raw is None:
raise ValueError("Missing value_farads in auto-resolved capacitor specs")
farads = _parse_spice_value(str(raw)) if isinstance(raw, str) else float(raw)
return CapacitorSpecs(
component_subtype=st, value_farads=farads,
value_formatted=value_formatted or _format_value(farads, "F"),
tolerance=tolerance, package=package,
voltage_rating_v=str(vals.get("voltage_rating_v")) if vals.get("voltage_rating_v") else None,
dielectric=str(vals.get("dielectric")) if vals.get("dielectric") else None,
)
if subtype == "passive.ferrite_bead":
raw = vals.get("impedance_ohm") or vals.get("value_ohms")
if raw is None:
raise ValueError("Missing impedance_ohm in auto-resolved ferrite bead specs")
z = _parse_spice_value(str(raw)) if isinstance(raw, str) else float(raw)
dcr_raw = vals.get("dcr_ohms")
dcr = None
if dcr_raw is not None:
dcr = _parse_spice_value(str(dcr_raw)) if isinstance(dcr_raw, str) else float(dcr_raw)
return InductorSpecs(
component_subtype=st, value_henries=None,
value_formatted=value_formatted or _format_value(z, "ohm"),
tolerance=tolerance, package=package,
current_rating_a=str(vals.get("current_rating_a")) if vals.get("current_rating_a") else None,
dcr_ohms=dcr, impedance_ohm=z,
)
if subtype.startswith("passive.inductor"):
raw = vals.get("value_henries")
if raw is None:
raise ValueError("Missing value_henries in auto-resolved inductor specs")
h = _parse_spice_value(str(raw)) if isinstance(raw, str) else float(raw)
dcr_raw = vals.get("dcr_ohms")
dcr = None
if dcr_raw is not None:
dcr = _parse_spice_value(str(dcr_raw)) if isinstance(dcr_raw, str) else float(dcr_raw)
return InductorSpecs(
component_subtype=st, value_henries=h, value_formatted=value_formatted,
tolerance=tolerance, package=package,
current_rating_a=str(vals.get("current_rating_a")) if vals.get("current_rating_a") else None,
dcr_ohms=dcr,
)
raise ValueError(f"Unsupported passive subtype for conversion: {subtype!r}")
class SkippedItem:
__slots__ = ("identifier", "stage", "error")
def __init__(self, identifier: str, stage: str, error: str) -> None:
self.identifier = identifier
self.stage = stage
self.error = error
def to_dict(self) -> dict[str, str]:
return {"identifier": self.identifier, "stage": self.stage, "error": self.error}
def load_patterns(
patterns_dir: str | Path, skipped: list[SkippedItem] | None = None,
) -> list[PassivePattern]:
patterns_dir = Path(patterns_dir)
out: list[PassivePattern] = []
for f in sorted(patterns_dir.glob("*.json")):
try:
out.append(PassivePattern(**json.loads(f.read_text())))
except Exception as exc:
if skipped is not None:
skipped.append(SkippedItem(f.stem, "passive_pattern_load", str(exc)))
return out
def resolve_mpn(mpn: str, patterns: list[PassivePattern]) -> tuple[PassivePattern, dict[str, str]] | None:
for pat in patterns:
m = re.match(pat.regex, mpn)
if m:
return pat, m.groupdict()
return None
def resolve_bom(
bom_path: str | Path,
patterns_dir: str | Path = "component-patterns",
*,
reference_col: str = "Reference",
mpn_col: str = "Manufacturer Part Number",
skipped: list[SkippedItem] | None = None,
) -> list[ResolvedPassive]:
patterns = load_patterns(patterns_dir, skipped=skipped)
bom = parse_bom(bom_path, reference_col=reference_col, mpn_col=mpn_col)
mpn_refs: dict[str, list[str]] = defaultdict(list)
for ref, info in bom.items():
mpn = info.get("mpn")
if mpn:
mpn_refs[mpn].append(ref)
resolved: list[ResolvedPassive] = []
for mpn, refs in sorted(mpn_refs.items()):
match = resolve_mpn(mpn, patterns)
if match is None:
continue
try:
pat, groups = match
by_name = {f.name: f for f in pat.fields}
digits = groups.get("resistance") or groups.get("capacitance") or ""
tol_code = groups.get("tolerance", "")
value = decode_value(digits, pat.value_decoder, tol_code)
formatted = _format_value(value, pat.value_decoder.output_unit)
def _lookup(field: str, group: str) -> str | None:
fd = by_name.get(field)
code = groups.get(group, "")
if fd and fd.lookup:
return fd.lookup.get(code, code)
return code or None
raw_fields: dict[str, str] = {}
for fname, fval in groups.items():
fd = by_name.get(fname)
if fd and fd.lookup:
raw_fields[fname] = fd.lookup.get(fval, fval)
else:
raw_fields[fname] = fval
resolved.append(ResolvedPassive(
mpn=mpn, references=sorted(refs),
component_type=pat.component_type, component_subtype=pat.component_subtype,
manufacturer=pat.manufacturer, series=pat.series,
value=value, value_formatted=formatted,
tolerance=_lookup("tolerance", "tolerance"),
package=_lookup("size", "size") or groups.get("size"),
voltage_rating=_lookup("voltage", "voltage"),
power_rating=_lookup("wattage", "wattage"),
dielectric=_lookup("dielectric", "dielectric"),
raw_fields=raw_fields,
))
except Exception as exc:
if skipped is not None:
skipped.append(SkippedItem(mpn, "passive_resolve", str(exc)))
return resolved
@@ -0,0 +1,27 @@
"""Compatibility names for native datasheet lookup and review parse.
Live review is ``review_session`` / ``services.validation``. These aliases keep
tests and leftover callers on one API.
"""
from __future__ import annotations
from backend.periscopex.constraints_lookup import (
build_constraints_map as _build_constraints_map,
load_datasheets as _load_datasheets,
match_constraints as _match_constraints,
)
from backend.periscopex.review_parse import (
ReviewResult,
assign_finding_ids,
parse_submit_review as _parse_review,
)
__all__ = [
"ReviewResult",
"assign_finding_ids",
"_build_constraints_map",
"_load_datasheets",
"_match_constraints",
"_parse_review",
]
@@ -0,0 +1,7 @@
"""Compatibility dispatcher for native graph-query review tools."""
from __future__ import annotations
from backend.periscopex.review_tools import execute_tool
__all__ = ["execute_tool"]
+21
View File
@@ -0,0 +1,21 @@
"""Prove leftover analysis modules live in periscope/src."""
from __future__ import annotations
from pathlib import Path
def test_analysis_leftovers_are_src():
import backend.periscopex.validate as v
import backend.periscopex.validation_tools as vt
import backend.periscopex.resolve_passives as rp
import backend.periscopex.derating as d
import backend.periscopex.bom_summary as bs
import backend.periscopex.pin_mux_check as mux
import backend.periscopex.led_current_check as led
import backend.periscopex.pin_function_tokens as tok
for mod in (v, vt, rp, d, bs, mux, led, tok):
parts = Path(mod.__file__).resolve().parts
assert "src" in parts
assert "dependency" not in parts[-6:]