from __future__ import annotations import re import uuid from bom_assistant.classification.classifier import classify, is_non_electronic from bom_assistant.classification.param_extractor import extract_params from bom_assistant.session.models import BomRow, RowState _DNP_KEYWORDS = re.compile( r"\b(dnp|do\s+not\s+place|do\s+not\s+populate|np|not\s+placed)\b", re.IGNORECASE, ) _DNP_TRUTHY = {"1", "true", "yes", "dnp", "x"} # Range designator: "C1-5", "R56-61" _RANGE_RE = re.compile(r"^([A-Za-z]+)(\d+)-(\d+)$") def normalize(mapped_rows: list[dict[str, str]]) -> list[BomRow]: rows: list[BomRow] = [] for raw in mapped_rows: row = _normalize_row(raw) if row is not None: rows.append(row) return rows def _normalize_row(raw: dict[str, str]) -> BomRow | None: value = raw.get("value", "").strip() # skip empty or KiCad placeholder if not value or value == "~": return None designators = _parse_designators(raw.get("designators", "")) quantity = _parse_quantity(raw.get("quantity", "")) footprint = raw.get("footprint", "").strip() or None # pull part_number / manufacturer from mapped columns if present part_number = raw.get("part_number", "").strip() or None manufacturer = raw.get("manufacturer", "").strip() or None # state determination state = _determine_state(value, footprint, quantity, raw) category = classify(designators, value) params = extract_params(value, footprint, category) # inject part_number / manufacturer from mapped columns if extractor didn't find them if part_number and not params.part_number: params.part_number = part_number if manufacturer and not params.manufacturer: params.manufacturer = manufacturer return BomRow( row_id=str(uuid.uuid4()), designators=designators, quantity=quantity, raw_value=value, footprint=footprint, category=category, normalized_params=params, state=state, ) def _parse_designators(s: str) -> list[str]: if not s: return [] # split on comma, semicolon, or whitespace parts = re.split(r"[,;\s]+", s.strip()) expanded: list[str] = [] for part in parts: part = part.strip() if not part: continue expanded.extend(_expand_range(part)) # deduplicate while preserving order seen: set[str] = set() result: list[str] = [] for d in expanded: if d not in seen: seen.add(d) result.append(d) return sorted(result, key=lambda x: (re.sub(r"\d", "", x), int(re.sub(r"\D", "", x) or 0))) def _expand_range(s: str) -> list[str]: m = _RANGE_RE.match(s) if m: prefix = m.group(1) start, end = int(m.group(2)), int(m.group(3)) if start <= end and (end - start) < 200: # sanity cap return [f"{prefix}{i}" for i in range(start, end + 1)] return [s] def _parse_quantity(s: str) -> int: s = s.strip() if not s: return 1 try: return max(0, int(float(s))) except (ValueError, TypeError): return 1 def _determine_state( value: str, footprint: str | None, quantity: int, raw: dict[str, str], ) -> RowState: if quantity == 0: return RowState.flagged combined = f"{value} {footprint or ''}" if _DNP_KEYWORDS.search(combined): return RowState.flagged # check explicit boolean DNP columns for col in ("DNP", "Exclude from BOM", "Mounting", "_raw_DNP", "_raw_Exclude from BOM", "_raw_Mounting"): col_val = raw.get(col, "").strip().lower() if col_val in _DNP_TRUTHY: return RowState.flagged if is_non_electronic(value): return RowState.flagged return RowState.pending