instant-bom/bom_assistant/suppliers/lcsc/scout_ground_truth.py

#!/usr/bin/env python3
"""
One-off: build a ground-truth dataset of (component -> real LCSC category)
from the two examples/full BOMs that happen to carry real LCSC part numbers.

Reads the raw BOM columns directly (not through the normal ingestion pipeline —
production BOMs won't have LCSC part numbers, so there's nothing to wire up
there). For each row, looks up the part's real category via product_lookup,
and writes one JSON line per row to ground_truth.ndjson:

    {"designator": "C38", "value": "330u, 63V", "footprint": "...",
     "part": "C437643", "category_id": 1140, "category_name": "Aluminum Electrolytic Capacitors",
     "parent_category_name": "Capacitors"}

Usage:
    python -m bom_assistant.suppliers.lcsc.scout_ground_truth
"""
from __future__ import annotations

import csv
import json
import re
from pathlib import Path

import openpyxl

from bom_assistant.suppliers.lcsc.product_lookup import lookup_product, sleep_between_requests

_EXAMPLES_DIR = Path(__file__).resolve().parents[3] / "examples" / "full"
_OUT_PATH = Path(__file__).parent / "ground_truth.ndjson"


def _rows_from_csv(path: Path) -> list[dict]:
    with path.open(encoding="utf-8-sig", newline="") as f:
        reader = csv.DictReader(f)
        return [
            {
                "source": path.name,
                "designators": row.get("Designator", ""),
                "value": row.get("Comment", ""),
                "footprint": row.get("Footprint", ""),
                "part": (row.get("LCSC Part #") or "").strip(),
            }
            for row in reader
            if (row.get("LCSC Part #") or "").strip()
        ]


def _rows_from_xlsx(path: Path) -> list[dict]:
    wb = openpyxl.load_workbook(path, data_only=True)
    ws = wb.active
    header = [str(c.value).strip() if c.value is not None else "" for c in next(ws.iter_rows(min_row=1, max_row=1))]
    idx = {name: i for i, name in enumerate(header)}

    out: list[dict] = []
    for row in ws.iter_rows(min_row=2, values_only=True):
        supplier = str(row[idx["Supplier"]] or "").strip() if "Supplier" in idx else ""
        part = str(row[idx["Supplier Part"]] or "").strip() if "Supplier Part" in idx else ""
        if supplier.upper() != "LCSC" or not part:
            continue
        out.append({
            "source": path.name,
            "designators": str(row[idx["Designator"]] or "") if "Designator" in idx else "",
            "value": str(row[idx["Value"]] or row[idx["Comment"]] or "") if "Value" in idx or "Comment" in idx else "",
            "footprint": str(row[idx["Footprint"]] or "") if "Footprint" in idx else "",
            "part": part,
        })
    return out


def _first_designator(designators: str) -> str:
    parts = re.split(r"[,;\s]+", designators.strip())
    return parts[0] if parts else ""


def main() -> None:
    sources = [
        (_EXAMPLES_DIR / "bom (1).csv", _rows_from_csv),
        (_EXAMPLES_DIR / "BOM_imx415_DigitalFPV (1).xlsx", _rows_from_xlsx),
    ]

    rows: list[dict] = []
    for path, reader in sources:
        if not path.exists():
            print(f"warn: missing {path}, skipping")
            continue
        found = reader(path)
        print(f"{path.name}: {len(found)} rows with an LCSC part #")
        rows.extend(found)

    print(f"\nLooking up {len(rows)} parts on lcsc.com …")
    written = 0
    with _OUT_PATH.open("w", encoding="utf-8") as out:
        for i, row in enumerate(rows, 1):
            part = row["part"]
            info = lookup_product(part)
            if info is None:
                print(f"  [{i}/{len(rows)}] {part}: lookup failed — skipping")
                sleep_between_requests()
                continue

            entry = {
                "source": row["source"],
                "designator": _first_designator(row["designators"]),
                "designators": row["designators"],
                "value": row["value"],
                "footprint": row["footprint"],
                "part": part,
                "category_id": info.category_id,
                "category_name": info.category_name,
                "parent_category_name": info.parent_category_name,
                "product_name": info.product_name,
            }
            out.write(json.dumps(entry, ensure_ascii=False) + "\n")
            written += 1
            print(f"  [{i}/{len(rows)}] {part} -> {info.category_name}")
            sleep_between_requests()

    print(f"\nWrote {written}/{len(rows)} ground-truth entries to {_OUT_PATH}")


if __name__ == "__main__":
    main()