AnasSarkiz/muse-voice-remote

scripts/download-assembly-models.py

"""Download original selected JLCPCB CAD; retain URLs and verify cached bytes."""
import concurrent.futures
import datetime
import hashlib
import json
from pathlib import Path
import urllib.request

root = Path(__file__).resolve().parent.parent
destination = root / "models/jlcpcb"
destination.mkdir(parents=True, exist_ok=True)
entries = json.loads((root / "vendor/jlcpcb/import-manifest.json").read_text())["imports"]
report_path = root / "docs/assembly-model-downloads.json"
previous = {e["lcsc"]: e for e in json.loads(report_path.read_text())["models"]} if report_path.exists() else {}

def fetch(entry):
    result = {"lcsc": entry["lcsc"], "models": {}}
    for extension, key in [("obj", "objUrl"), ("step", "stepUrl")]:
        url = entry["cad"][key]
        path = destination / f'{entry["lcsc"]}.{extension}'
        try:
            old = previous.get(entry["lcsc"], {}).get("models", {}).get(extension, {})
            if not path.exists() or hashlib.sha256(path.read_bytes()).hexdigest() != old.get("sha256"):
                with urllib.request.urlopen(url, timeout=50) as response:
                    payload = response.read(32 * 1024 * 1024 + 1)
                if len(payload) > 32 * 1024 * 1024 or b"<html" in payload[:100].lower():
                    raise ValueError("Invalid CAD response")
                if extension == "step" and not payload.lstrip().startswith(b"ISO-10303-21;"):
                    raise ValueError("Not a STEP file")
                if extension == "obj" and not any(line.startswith(b"v ") for line in payload.splitlines()):
                    raise ValueError("Not an OBJ mesh")
                path.write_bytes(payload)
            payload = path.read_bytes()
            result["models"][extension] = {"url": url, "path": str(path.relative_to(root)), "bytes": len(payload),
                "sha256": hashlib.sha256(payload).hexdigest(), "status": "downloaded"}
            if extension == "obj":
                vertices = [list(map(float, line.split()[1:4])) for line in payload.decode().splitlines() if line.startswith("v ")]
                result["obj_bounds"] = {"min": [min(v[i] for v in vertices) for i in range(3)],
                    "max": [max(v[i] for v in vertices) for i in range(3)], "vertex_count": len(vertices)}
        except Exception as error:
            result["models"][extension] = {"url": url, "status": "failed", "error": str(error)}
    print(entry["lcsc"], ", ".join(k + ":" + v["status"] for k, v in result["models"].items()), flush=True)
    return result

with concurrent.futures.ThreadPoolExecutor(max_workers=6) as pool:
    results = list(pool.map(fetch, entries))
report_path.write_text(json.dumps({"checked_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
    "source": "Original JLCPCB/EasyEDA URLs retained by imported components", "models": results}, indent=2) + "\n")
raise SystemExit(any(m["status"] != "downloaded" for e in results for m in e["models"].values()))