shibosoftwaredev/f1c100s-linux-nema17-stepper-controller

scripts/export-assembly.py

#!/usr/bin/env python3
"""Create a self-consistent Gerber, drill, BOM and CPL fabrication package."""

import csv
import hashlib
import io
import json
import os
from pathlib import Path
import subprocess
import tempfile
import zipfile
from fabrication_handoff import promote_handoff


ROOT = Path(__file__).resolve().parents[1]
DIST = ROOT / "dist" / "index"
FINAL_FAB = ROOT / "fabrication"
REPORTS = ROOT / "reports"
CIRCUIT_PATH = DIST / "circuit.json"

# Never overwrite a fabrication handoff using an incomplete routing build.
preflight_bytes = CIRCUIT_PATH.read_bytes()
preflight = json.loads(preflight_bytes)
assert not [e for e in preflight if e["type"].endswith("_error")]
assert next(e for e in preflight if e["type"]=="pcb_board")["num_layers"] == 4
for via in [e for e in preflight if e["type"]=="pcb_via"]:
    assert via["hole_diameter"]==.3 and via["outer_diameter"]==.45
    assert set(via["layers"])=={"top","inner1","inner2","bottom"}
preflight_stock=json.loads((REPORTS/"jlcpcb-stock-audit.json").read_text())
assert preflight_stock["circuitSha256"] == hashlib.sha256(preflight_bytes).hexdigest()
assert all(p["status"]=="in-stock" for p in preflight_stock["parts"]), "JLCPCB orderable assembly stock is insufficient; do not release an assembly handoff"
continuity=json.loads((REPORTS/"connectivity-audit.json").read_text())
assert continuity['circuitSha256']==hashlib.sha256(preflight_bytes).hexdigest()
assert continuity['sourceFailures']==0 and continuity['disconnectedTerminalIslands']==0
assert continuity['emittedCopperOnly'] is True
preflight_sha = hashlib.sha256(preflight_bytes).hexdigest()
bom_validation = json.loads((REPORTS / "bom-validation.json").read_text())
assert bom_validation["circuitSha256"] == preflight_sha
assert not bom_validation["findings"], "Catalog value/voltage/package checks must pass before export"
drill_validation = json.loads((REPORTS / "drill-clearance-regression.json").read_text())
assert drill_validation["circuitSha256"] == preflight_sha
assert drill_validation["currentFindings"] == 0, "Inherited drill findings are blockers, not waivers"
additional_dfm = json.loads((REPORTS / "additional-dfm-review.json").read_text())
assert additional_dfm["circuitSha256"] == preflight_sha
assert not additional_dfm["findings"], "Independent pad/PTH DFM checks must pass before export"
assert additional_dfm["silkscreenBelowPublished1mmHeight"] == 0, "Manufacturing legend cleanup must pass before export"
solder_process = json.loads((REPORTS / "solder-process-audit.json").read_text())
assert solder_process["circuitSha256"] == preflight_sha, "Fresh solder-process review required"
assert solder_process["approvedForAssembly"] is True, "Ordinary via-in-pad and thermal/opposite-face solder process remain unresolved; do not release an assembly handoff"


def normalize_csv(data: bytes):
    text = data.decode().replace("\\r\\n", "\n").replace("\r\n", "\n")
    rows = list(csv.DictReader(io.StringIO(text)))
    assert rows, "empty assembly CSV"
    out = io.StringIO(newline="")
    writer = csv.DictWriter(out, fieldnames=list(rows[0]))
    writer.writeheader()
    writer.writerows(rows)
    return out.getvalue().encode(), rows


# Build in an isolated directory. Nothing in the previous handoff is removed
# while export, archive validation or assembly validation can still fail.
staging = tempfile.TemporaryDirectory(prefix="nema17-fabrication-", dir=ROOT)
FAB = Path(staging.name) / "fabrication"
FAB.mkdir()
ARCHIVE = FAB / "board-gerbers.zip"
REPORTS.mkdir(exist_ok=True)

# tsci resolves output paths relative to the input circuit JSON directory.
result = subprocess.run(
    [str(ROOT/"node_modules"/".bin"/"tsci"), "export", "dist/index/circuit.json", "--format", "gerbers",
     "--output", os.path.relpath(ARCHIVE, DIST)],
    cwd=ROOT, text=True, capture_output=True, env={**os.environ,"TMPDIR":"/tmp"},
)
if result.returncode:
    raise SystemExit(result.stdout + result.stderr)

circuit_bytes = CIRCUIT_PATH.read_bytes()
assert circuit_bytes == preflight_bytes, "Circuit changed during export; retain previous handoff"
circuit = json.loads(circuit_bytes)
sources = {item["source_component_id"]: item for item in circuit
           if item["type"] == "source_component"}
placed = {sources[item["source_component_id"]]["name"]: item for item in circuit
          if item["type"] == "pcb_component" and item["source_component_id"] in sources}
expected = {name for name, item in placed.items()
            if sources[item["source_component_id"]].get("ftype") != "simple_test_point"
            and not item.get("do_not_place", False)}
excluded = set(placed) - expected

with zipfile.ZipFile(ARCHIVE) as archive:
    entries = {name: archive.read(name) for name in archive.namelist()}

bom_data, bom_rows = normalize_csv(entries["bom.csv"])
pnp_data, pnp_rows = normalize_csv(entries["pick_and_place.csv"])
bom_designators = {row["Designator"] for row in bom_rows}
pnp_designators = {row["Designator"] for row in pnp_rows}
assert bom_designators == expected, f"BOM mismatch: missing={expected - bom_designators}, extra={bom_designators - expected}"
assert pnp_designators == expected, f"CPL mismatch: missing={expected - pnp_designators}, extra={pnp_designators - expected}"
assert all(row["Layer"] in {"top", "bottom"} for row in pnp_rows), "unexpected assembly layer"
assert all(row["JLCPCB Part #"].startswith("C") for row in bom_rows), "missing JLCPCB part number"

entries["bom.csv"] = bom_data
entries["pick_and_place.csv"] = pnp_data
with zipfile.ZipFile(ARCHIVE, "w", zipfile.ZIP_DEFLATED) as archive:
    for name, data in entries.items():
        archive.writestr(name, data)

for name, data in entries.items():
    if name.endswith((".gbr", ".drl")):
        assert Path(name).name == name, "Unexpected nested fabrication export path"
        (FAB / name).write_bytes(data)
(FAB / "assembly-bom.csv").write_bytes(bom_data)
(FAB / "assembly-pick-and-place.csv").write_bytes(pnp_data)

vias = [item for item in circuit if item["type"] == "pcb_via"]
errors = [item for item in circuit if item["type"].endswith("_error")]
stock = json.loads((REPORTS / "jlcpcb-stock-audit.json").read_text())
circuit_sha = hashlib.sha256(circuit_bytes).hexdigest()
assert not errors
assert all(via["hole_diameter"] == 0.3 and via["outer_diameter"] == 0.45 for via in vias)
assert stock["circuitSha256"] == circuit_sha
assert all(part["status"] == "in-stock" for part in stock["parts"])

audit = {
    "circuitJsonSha256": circuit_sha,
    "gerbersSha256": hashlib.sha256(ARCHIVE.read_bytes()).hexdigest(),
    "drcErrors": len(errors),
    "assemblyComponents": len(expected),
    "assemblyTopCount": sum(row["Layer"] == "top" for row in pnp_rows),
    "assemblyBottomCount": sum(row["Layer"] == "bottom" for row in pnp_rows),
    "assemblyBottomComponents": sorted(
        row["Designator"] for row in pnp_rows if row["Layer"] == "bottom"
    ),
    "excludedBareTestFeatures": sorted(excluded),
    "pcbVias": len(vias),
    "viaHoleDiameterMm": 0.3,
    "viaPadDiameterMm": 0.45,
    "jlcpcbUniqueParts": stock["uniquePartCount"],
    "jlcpcbStockIssues": sum(part["status"] != "in-stock" for part in stock["parts"]),
    "assemblyOrderCaveats": [
        {key:part.get(key) for key in ["code","designators","orderableStock","minimumAssemblyQuantity","assemblyLossQuantity"]}
        for part in stock["parts"]
        if part["orderableStock"] < (part.get("minimumAssemblyQuantity") or part["quantityPerBoard"])
    ],
    "generatedFiles": sorted(name for name in entries if name.endswith((".gbr", ".drl"))),
    "assemblyNote": "Review polarity and rotation in the JLCPCB placement preview before payment; this is required for every PCBA order.",
}
audit_text = json.dumps(audit, indent=2) + "\n"
(FAB / "fabrication-audit.json").write_text(audit_text)
backup = promote_handoff(FAB, FINAL_FAB)
staging.cleanup()
# The authoritative audit travels inside the promoted handoff. Keep the
# conventional report copy in sync using an atomic single-file replacement.
report_tmp = REPORTS / "fabrication-audit.json.tmp"
report_tmp.write_text(audit_text)
report_tmp.replace(REPORTS / "fabrication-audit.json")
if backup:
    print(f"Previous fabrication handoff preserved at {backup}")
print(json.dumps(audit, indent=2))