diff --git a/periscope/src/backend/periscopex/bom_summary.py b/periscope/src/backend/periscopex/bom_summary.py new file mode 100644 index 0000000..fa35c6b --- /dev/null +++ b/periscope/src/backend/periscopex/bom_summary.py @@ -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 diff --git a/periscope/src/backend/periscopex/derating.py b/periscope/src/backend/periscopex/derating.py new file mode 100644 index 0000000..fa58a69 --- /dev/null +++ b/periscope/src/backend/periscopex/derating.py @@ -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 diff --git a/periscope/src/backend/periscopex/led_current_check.py b/periscope/src/backend/periscopex/led_current_check.py new file mode 100644 index 0000000..29d1773 --- /dev/null +++ b/periscope/src/backend/periscopex/led_current_check.py @@ -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", + ) diff --git a/periscope/src/backend/periscopex/pin_function_tokens.py b/periscope/src/backend/periscopex/pin_function_tokens.py new file mode 100644 index 0000000..46bcd82 --- /dev/null +++ b/periscope/src/backend/periscopex/pin_function_tokens.py @@ -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) diff --git a/periscope/src/backend/periscopex/pin_mux_check.py b/periscope/src/backend/periscopex/pin_mux_check.py new file mode 100644 index 0000000..e7f7165 --- /dev/null +++ b/periscope/src/backend/periscopex/pin_mux_check.py @@ -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", + ) diff --git a/periscope/src/backend/periscopex/resolve_passives.py b/periscope/src/backend/periscopex/resolve_passives.py new file mode 100644 index 0000000..429ecbd --- /dev/null +++ b/periscope/src/backend/periscopex/resolve_passives.py @@ -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 diff --git a/periscope/src/backend/periscopex/validate.py b/periscope/src/backend/periscopex/validate.py new file mode 100644 index 0000000..ec194e6 --- /dev/null +++ b/periscope/src/backend/periscopex/validate.py @@ -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", +] diff --git a/periscope/src/backend/periscopex/validation_tools.py b/periscope/src/backend/periscopex/validation_tools.py new file mode 100644 index 0000000..4be3888 --- /dev/null +++ b/periscope/src/backend/periscopex/validation_tools.py @@ -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"] diff --git a/tests/test_periscope_analysis_rewrite.py b/tests/test_periscope_analysis_rewrite.py new file mode 100644 index 0000000..992f8ac --- /dev/null +++ b/tests/test_periscope_analysis_rewrite.py @@ -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:]