Altium - the Altium Bridge

#!/usr/bin/env python3
"""altium-bom: the BOM work Altium leaves to a spreadsheet, from the container.

  altium-bom read  --project <C:\\...\\board.PrjPcb> [--variant V] [--boards N] [--target BOX] [--out bom.json]
      altium_bom through adom-bridge: every sheet's placed parts and parameters merged with the board,
      grouped one line per MPN, checked; the files land beside the project on Windows (BOM\\) and the
      JSON here.
  altium-bom price bom.json [--boards N] [--out DIR]
      One adom-parts-search call per line (Mouser, DigiKey and JLCPCB in parallel; mind the shared daily
      quota): stock against what N boards need, unit price, lead time, lifecycle, the distributor
      numbers each vendor actually sells the MPN under. Writes price.md, price.csv and suggestions.json.
  altium-bom fill bom.json --suggestions suggestions.json --project <PrjPcb> [--apply] [--target BOX]
      The distributor numbers the BOM lacks (or has wrong), written back onto the schematic sheets with
      altium_sheet_params. Without --apply it only prints the plan.

Every command prints OK:/ERROR: first and Hint: lines after, like the other Adom CLIs.
"""
import argparse
import csv
import json
import os
import re
import subprocess
import sys
from pathlib import Path

VENDORS = ("mouser", "digikey", "jlcpcb")
FIELD = {"mouser": "mouser", "digikey": "digikey", "jlcpcb": "lcsc"}
PARAM = {"mouser": "Mouser Part #", "digikey": "DigiKey Part #", "jlcpcb": "LCSC Part #"}
BAD_LIFECYCLE = re.compile(r"obsolete|nrnd|not recommended|end of life|eol|discontinued|last time buy", re.I)


def ok(msg, hints=()):
    print("OK: " + msg)
    for h in hints:
        if h:
            print("Hint: " + h)


def err(msg, hints=()):
    print("ERROR: " + msg)
    for h in hints:
        if h:
            print("Hint: " + h)
    sys.exit(1)


def bridge(verb, args, target=None, thread=None, timeout=900):
    cmd = ["adom-bridge", "--ai-thread", thread or os.environ.get("ADOM_AI_THREAD", "altium-bom")]
    if target:
        cmd += ["--target", target]
    cmd += [verb, json.dumps(args)]
    try:
        out = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, stdin=subprocess.DEVNULL).stdout
    except subprocess.TimeoutExpired:
        err("%s did not answer within %d s" % (verb, timeout))
    except FileNotFoundError:
        err("adom-bridge is not on PATH", ["This runs in an Adom container with Adom Bridge connected to the Windows machine running Altium."])
    try:
        d = json.loads(out)
    except ValueError:
        err("%s answered something that is not JSON: %s" % (verb, out[:300]))
    o = d.get("output", d)
    if isinstance(o, str) and o.startswith("{"):
        try:
            o = json.loads(o)
        except ValueError:
            pass
    return o if isinstance(o, dict) else d


def money(s):
    if s in (None, ""):
        return None
    m = re.search(r"[\d.,]+", str(s))
    if not m:
        return None
    try:
        return float(m.group(0).replace(",", ""))
    except ValueError:
        return None


def norm(pn):
    return re.sub(r"[\s\-_/.]", "", (pn or "").upper())


def search(mpn):
    """adom-parts-search search <mpn>: {vendor: [hits]} (exact MPN matches first)."""
    try:
        r = subprocess.run(["adom-parts-search", "search", mpn, "--limit", "4"], capture_output=True, text=True, timeout=120, stdin=subprocess.DEVNULL)
    except FileNotFoundError:
        err("adom-parts-search is not installed", ["adom-wiki pkg install adom/adom-parts-search"])
    except subprocess.TimeoutExpired:
        return {"error": "timed out"}
    t = r.stdout
    try:
        j = json.loads(t[t.index("{"):])
    except ValueError:
        return {"error": (t or r.stderr)[:200]}
    out = {}
    for v in VENDORS:
        hits = (j.get(v) or {}).get("hits") or []
        out[v] = sorted(hits, key=lambda h: norm(h.get("mpn")) != norm(mpn))
        if (j.get(v) or {}).get("error"):
            out[v + "_error"] = j[v]["error"]
    return out


def cmd_read(a):
    if not a.project:
        err("--project <C:\\...\\board.PrjPcb> is required")
    args = {"projectPath": a.project, "boards": a.boards, "timeout": 600}
    if a.variant:
        args["variant"] = a.variant
    if a.output_dir:
        args["outputDir"] = a.output_dir
    r = bridge("altium_bom", args, a.target, timeout=900)
    if not r.get("success"):
        err("altium_bom: %s" % (r.get("error") or r.get("errorCode") or "failed"), [r.get("_hint")])
    out = Path(a.out or "bom.json")
    out.write_text(json.dumps(r, indent=1))
    c = r.get("checks") or {}
    gaps = ", ".join("%d without %s" % (len(v), k) for k, v in (c.get("missingPartNumbers") or {}).items() if v)
    ok("%d lines, %d fitted parts from %d sheet(s): %s" % (r.get("lineCount", 0), r.get("fittedParts", 0), len(r.get("sheets") or []), out),
       ["Files on the Windows machine: " + ", ".join(Path(f.replace("\\", "/")).name for f in r.get("files") or []),
        ("%d part(s) have no MPN: %s" % (len(c["missingMpn"]), ", ".join(c["missingMpn"][:12]))) if c.get("missingMpn") else "",
        ("Distributor numbers missing: " + gaps) if gaps else "",
        ("On the board but not on a sheet: " + ", ".join(c["boardNotOnSheet"][:12])) if c.get("boardNotOnSheet") else "",
        ("On a sheet but not on the board: " + ", ".join(c["sheetNotOnBoard"][:12])) if c.get("sheetNotOnBoard") else "",
        ("%d line(s) mix values for one part (checks.inconsistent)" % len(c["inconsistent"])) if c.get("inconsistent") else "",
        "Next: altium-bom price %s --boards %d" % (out, a.boards)])


def cmd_price(a):
    bom = json.loads(Path(a.bom).read_text())
    lines = bom.get("lines") or []
    boards = a.boards or bom.get("boards") or 1
    out_dir = Path(a.out or Path(a.bom).with_suffix("").name + "-price")
    out_dir.mkdir(parents=True, exist_ok=True)
    rows, suggestions, flags_all = [], [], {}
    total = {v: 0.0 for v in VENDORS}
    best_total, unpriced = 0.0, []
    for i, l in enumerate(lines, 1):
        mpn = l.get("mpn") or ""
        need = int(l.get("quantity") or 0) * boards
        refs = ",".join(l.get("designators") or [])
        row = {"line": i, "designators": refs, "value": l.get("value", ""), "mpn": mpn, "manufacturer": l.get("manufacturer", ""), "need": need}
        flags = []
        if not mpn:
            flags.append("NO_MPN")
            row["flags"] = flags
            rows.append(row)
            unpriced.append(refs)
            continue
        found = search(mpn)
        if found.get("error"):
            flags.append("SEARCH_FAILED")
        best = None
        for v in VENDORS:
            hits = found.get(v) or []
            exact = [h for h in hits if norm(h.get("mpn")) == norm(mpn)]
            h = exact[0] if exact else None
            have = l.get(FIELD[v]) or ""
            if h:
                price = money(h.get("price_100")) if need >= 100 and money(h.get("price_100")) else money(h.get("unit_price"))
                stock = h.get("stock")
                row[v] = {"pn": h.get("vendor_pn"), "stock": stock, "unit": price, "lead": h.get("lead_time") or "", "flags": h.get("flags") or []}
                if price is not None:
                    total[v] += price * need
                    if (stock or 0) >= need and (best is None or price < best[1]):
                        best = (v, price)
                life = " ".join(str(x) for x in (h.get("flags") or [])) + " " + str(h.get("lifecycle") or "")
                if BAD_LIFECYCLE.search(life):
                    flags.append("LIFECYCLE_%s" % v.upper())
                if have and norm(have) != norm(h.get("vendor_pn")):
                    # the BOM's number may be a reel vs cut-tape twin; say so, do not overwrite silently
                    flags.append("PN_DIFFERS_%s" % v.upper())
                    suggestions.append({"designators": l.get("designators"), "param": PARAM[v], "value": h.get("vendor_pn"), "was": have, "why": "%s sells %s as %s" % (v, mpn, h.get("vendor_pn")), "replace": True})
                elif not have:
                    suggestions.append({"designators": l.get("designators"), "param": PARAM[v], "value": h.get("vendor_pn"), "was": "", "why": "%s sells %s as %s" % (v, mpn, h.get("vendor_pn")), "replace": False})
            else:
                row[v] = None
        stock = {v: ((row.get(v) or {}).get("stock") or 0) for v in VENDORS}
        dist = max(stock["mouser"], stock["digikey"])
        if not any(row.get(v) for v in VENDORS):
            flags.append("NOT_FOUND")
        elif max(stock.values()) == 0:
            flags.append("NO_STOCK")
        elif max(stock.values()) < need:
            flags.append("LOW_STOCK")
        elif dist < need:
            # in stock only for JLCPCB assembly: buying it from a distributor means their lead time
            flags.append("JLCPCB_ONLY")
        if best:
            row["best"] = {"vendor": best[0], "unit": best[1], "extended": round(best[1] * need, 2)}
            best_total += best[1] * need
        else:
            unpriced.append(refs)
        row["flags"] = flags
        for f in flags:
            flags_all.setdefault(f, []).append(refs)
        rows.append(row)
    # the files
    md = ["# BOM pricing: %d line(s), %d board(s)" % (len(lines), boards), "",
          "Searched with adom-parts-search (Mouser, DigiKey, JLCPCB). Prices are the vendors' list prices at the time of the search; stock is a snapshot.", "",
          "| # | Designators | Value | MPN | Need | Mouser (stock, unit) | DigiKey (stock, unit) | JLCPCB (stock) | Best | Flags |", "|---|---|---|---|---|---|---|---|---|---|"]
    fmt = lambda c, show_price=True: "" if not c else "%s (%s%s)" % (c["pn"], c["stock"], (", $%.4g" % c["unit"]) if show_price and c.get("unit") is not None else "")
    for r in rows:
        md.append("| %d | %s | %s | %s | %d | %s | %s | %s | %s | %s |" % (r["line"], r["designators"], r["value"], r["mpn"], r["need"], fmt(r.get("mouser")), fmt(r.get("digikey")), fmt(r.get("jlcpcb"), False),
                  ("%s $%.2f" % (r["best"]["vendor"], r["best"]["extended"])) if r.get("best") else "", " ".join(r.get("flags") or [])))
    md += ["", "**Cheapest in-stock total:** $%.2f for %d board(s) (%s line(s) not priced: %s)" % (best_total, boards, len(unpriced), "; ".join(unpriced[:10]) or "none")]
    (out_dir / "price.md").write_text("\n".join(md) + "\n")
    with open(out_dir / "price.csv", "w", newline="") as f:
        w = csv.writer(f)
        w.writerow(["Line", "Designators", "Value", "MPN", "Manufacturer", "Need", "Mouser PN", "Mouser stock", "Mouser unit", "DigiKey PN", "DigiKey stock", "DigiKey unit", "LCSC", "JLCPCB stock", "Best vendor", "Best extended", "Flags"])
        for r in rows:
            g = lambda v, k: (r.get(v) or {}).get(k, "") if r.get(v) else ""
            w.writerow([r["line"], r["designators"], r["value"], r["mpn"], r["manufacturer"], r["need"], g("mouser", "pn"), g("mouser", "stock"), g("mouser", "unit"),
                        g("digikey", "pn"), g("digikey", "stock"), g("digikey", "unit"), g("jlcpcb", "pn"), g("jlcpcb", "stock"),
                        (r.get("best") or {}).get("vendor", ""), (r.get("best") or {}).get("extended", ""), " ".join(r.get("flags") or [])])
    (out_dir / "suggestions.json").write_text(json.dumps({"project": bom.get("projectPath"), "suggestions": suggestions}, indent=1))
    (out_dir / "price.json").write_text(json.dumps({"boards": boards, "rows": rows, "bestTotal": round(best_total, 2), "flags": flags_all}, indent=1))
    worry = {k: v for k, v in flags_all.items() if k in ("NO_STOCK", "LOW_STOCK", "NOT_FOUND", "NO_MPN", "JLCPCB_ONLY") or k.startswith("LIFECYCLE")}
    ok("%d line(s) priced for %d board(s): cheapest in-stock total $%.2f; %s" % (len(rows) - len(unpriced), boards, best_total, out_dir / "price.md"),
       ["%s: %s" % (k, "; ".join(v[:8])) for k, v in sorted(worry.items())] +
       ["%d distributor number(s) to add or correct: altium-bom fill %s --suggestions %s --project <PrjPcb> (dry run; --apply writes them onto the sheets)" % (len(suggestions), a.bom, out_dir / "suggestions.json") if suggestions else "",
        "A PN_DIFFERS flag is often a reel vs cut-tape twin of the same part: check before replacing (fill only replaces with --replace)."])


def cmd_fill(a):
    sug = json.loads(Path(a.suggestions).read_text()).get("suggestions") or []
    if not a.replace:
        sug = [s for s in sug if not s.get("replace")]
    if not sug:
        ok("nothing to write", ["Every line already has the distributor numbers (run with --replace to also correct the PN_DIFFERS ones)."])
        return
    # one sheet_params call per designator set: all of that line's parameters together
    plan = {}
    for s in sug:
        key = ",".join(s["designators"])
        plan.setdefault(key, {})[s["param"]] = s["value"]
    for refs, params in plan.items():
        print("  %s: %s" % (refs, ", ".join("%s=%s" % kv for kv in params.items())))
    if not a.apply:
        ok("%d write(s) planned (dry run)" % len(plan), ["Run again with --apply to write them onto the sheets (each touched sheet is saved)."])
        return
    project = a.project or json.loads(Path(a.suggestions).read_text()).get("project")
    if not project:
        err("--project <PrjPcb> is required")
    done, failed = 0, []
    for refs, params in plan.items():
        r = bridge("altium_sheet_params", {"projectPath": project, "designators": refs.split(","), "params": params}, a.target, timeout=400)
        if r.get("success"):
            done += 1
        else:
            failed.append("%s (%s)" % (refs, r.get("error") or r.get("errorCode")))
    if failed:
        err("%d of %d write(s) failed: %s" % (len(failed), len(plan), "; ".join(failed[:6])), ["altium-bom read again shows what landed."])
    ok("%d line(s) updated on the sheets" % done, ["altium-bom read --project %s re-reads the BOM; Design > Update PCB Document carries the parameters to the board." % project])


def main():
    ap = argparse.ArgumentParser(prog="altium-bom", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    sub = ap.add_subparsers(dest="cmd", required=True)
    r = sub.add_parser("read")
    r.add_argument("--project")
    r.add_argument("--variant")
    r.add_argument("--boards", type=int, default=1)
    r.add_argument("--output-dir")
    r.add_argument("--target")
    r.add_argument("--out")
    p = sub.add_parser("price")
    p.add_argument("bom")
    p.add_argument("--boards", type=int)
    p.add_argument("--out")
    f = sub.add_parser("fill")
    f.add_argument("bom")
    f.add_argument("--suggestions", required=True)
    f.add_argument("--project")
    f.add_argument("--apply", action="store_true")
    f.add_argument("--replace", action="store_true")
    f.add_argument("--target")
    a = ap.parse_args()
    {"read": cmd_read, "price": cmd_price, "fill": cmd_fill}[a.cmd](a)


if __name__ == "__main__":
    main()