crimpdeq-pcb

#!/usr/bin/env python3

import argparse
import csv
import re
import sys
from collections import OrderedDict
from pathlib import Path


TRUTHY_DNP = {"1", "true", "yes", "y", "x", "dnp", "do not populate", "fitted=no"}
FALSY_DNP = {"", "0", "false", "no", "n"}


def normalize_header(name: str) -> str:
    return name.lstrip("\ufeff").strip().lower().replace("_", " ")


def get_field(row: dict, *names: str) -> str:
    for wanted in names:
        wanted_norm = normalize_header(wanted)
        for key, value in row.items():
            if normalize_header(key) == wanted_norm:
                return (value or "").strip()
    return ""


def is_dnp(value: str) -> bool:
    normalized = value.strip().lower()
    if normalized in FALSY_DNP:
        return False
    if normalized in TRUTHY_DNP:
        return True
    return bool(normalized)


def split_references(value: str) -> list[str]:
    if not value.strip():
        return []
    return [part.strip() for part in value.split(",") if part.strip()]


def ref_sort_key(ref: str):
    return ref.strip()


def normalize_footprint(footprint: str) -> str:
    suffix = footprint.split(":")[-1].strip()
    metric_match = re.match(r"^[A-Za-z]+_(\d{4})_\d+Metric(?:_.*)?$", suffix)
    if metric_match:
        return metric_match.group(1)
    return suffix


def normalize_note(note: str) -> str:
    return re.sub(r"\s+", " ", note or "").strip()


def parse_quantity(row: dict, references: list[str]) -> int:
    qty = get_field(row, "Quantity", "Qty")
    if qty:
        try:
            return int(qty)
        except ValueError:
            pass
    return max(len(references), 1)


def sexpr_paren_delta(text: str) -> int:
    delta = 0
    in_string = False
    escape = False
    for ch in text:
        if escape:
            escape = False
            continue
        if in_string and ch == "\\":
            escape = True
            continue
        if ch == '"':
            in_string = not in_string
            continue
        if not in_string:
            if ch == "(":
                delta += 1
            elif ch == ")":
                delta -= 1
    return delta


def load_schematic_properties(path: Path) -> dict[str, dict[str, str]]:
    symbols: dict[str, dict[str, str]] = {}
    depth = 0
    in_symbol_block = False
    symbol_depth = 0
    block_lines: list[str] = []

    with path.open("r", encoding="utf-8") as infile:
        for line in infile:
            stripped = line.lstrip()
            if not in_symbol_block and depth == 1 and stripped.startswith("(symbol"):
                in_symbol_block = True
                symbol_depth = depth
                block_lines = [line]
            elif in_symbol_block:
                block_lines.append(line)

            depth += sexpr_paren_delta(line)

            if in_symbol_block and depth == symbol_depth:
                block = "".join(block_lines)
                props = dict(re.findall(r'\(property "([^"]+)" "([^"]*)"', block))
                ref = props.get("Reference", "").strip()
                if ref:
                    symbols[ref] = props
                in_symbol_block = False
                block_lines = []

    return symbols


def get_schematic_property_values(
    schematic_props: dict[str, dict[str, str]], references: list[str], property_name: str
) -> list[str]:
    values: list[str] = []
    seen = set()
    for ref in references:
        value = schematic_props.get(ref, {}).get(property_name, "").strip()
        if property_name == "AssemblyNote":
            value = normalize_note(value)
        if value and value not in seen:
            values.append(value)
            seen.add(value)
    return values


def main() -> int:
    parser = argparse.ArgumentParser(description="Convert a KiCad BOM into grouped assembly BOM format.")
    parser.add_argument("input", type=Path)
    parser.add_argument("output", type=Path)
    parser.add_argument(
        "--allow-missing-lcsc",
        action="store_true",
        help="Allow populated BOM entries without an LCSC Part # and leave the field empty.",
    )
    parser.add_argument(
        "--schematic",
        type=Path,
        help="Optional KiCad schematic used to backfill properties like LCSC and AssemblyNote by reference.",
    )
    args = parser.parse_args()

    grouped = OrderedDict()
    missing_lcsc = []
    schematic_props = load_schematic_properties(args.schematic) if args.schematic else {}

    with args.input.open("r", encoding="utf-8-sig", newline="") as infile:
        reader = csv.DictReader(infile)
        for row in reader:
            references = split_references(get_field(row, "Reference", "Designator"))
            value = get_field(row, "Value")
            footprint = normalize_footprint(get_field(row, "Footprint"))
            lcsc = get_field(row, "LCSC Part #", "LCSC")
            note = normalize_note(get_field(row, "Assembly Note", "AssemblyNote"))
            dnp = get_field(row, "DNP")

            if is_dnp(dnp):
                continue

            if schematic_props:
                if not lcsc:
                    lcsc_values = get_schematic_property_values(schematic_props, references, "LCSC")
                    if len(lcsc_values) == 1:
                        lcsc = lcsc_values[0]
                if not note:
                    note = " | ".join(get_schematic_property_values(schematic_props, references, "AssemblyNote"))

            if not lcsc:
                missing_lcsc.append(
                    {
                        "refs": references,
                        "value": value,
                        "footprint": footprint,
                    }
                )

            key = (footprint, value, lcsc, note)
            if key not in grouped:
                grouped[key] = {
                    "refs": [],
                    "quantity": 0,
                    "notes": [],
                }

            grouped[key]["refs"].extend(references)
            grouped[key]["quantity"] += parse_quantity(row, references)
            if note and note not in grouped[key]["notes"]:
                grouped[key]["notes"].append(note)

    if missing_lcsc:
        level = "warning" if args.allow_missing_lcsc else "error"
        print(f"{level}: missing LCSC Part # for populated BOM entries:", file=sys.stderr)
        for item in missing_lcsc:
            refs = ", ".join(item["refs"]) or "<unknown ref>"
            value = item["value"] or "<unknown value>"
            footprint = item["footprint"] or "<unknown footprint>"
            print(f"  - {refs}: {value} [{footprint}]", file=sys.stderr)
        if not args.allow_missing_lcsc:
            return 1

    rows = []
    for (footprint, value, lcsc, note), data in grouped.items():
        refs = sorted(dict.fromkeys(data["refs"]), key=ref_sort_key)
        notes = data["notes"] or ([note] if note else [])
        rows.append(
            {
                "Designator": ", ".join(refs),
                "Footprint": footprint,
                "Quantity": str(data["quantity"] or len(refs) or 1),
                "Value": value,
                "LCSC Part #": lcsc,
                "Assembly Note": " | ".join(notes),
                "_sort_refs": refs,
            }
        )

    rows.sort(key=lambda row: ref_sort_key(row["_sort_refs"][0]) if row["_sort_refs"] else ("", 0, ""))

    args.output.parent.mkdir(parents=True, exist_ok=True)
    with args.output.open("w", encoding="utf-8", newline="") as outfile:
        fieldnames = ["Designator", "Footprint", "Quantity", "Value", "LCSC Part #", "Assembly Note"]
        writer = csv.DictWriter(outfile, fieldnames=fieldnames)
        writer.writeheader()
        for row in rows:
            row.pop("_sort_refs", None)
            writer.writerow(row)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())