HartOMat/scripts/compare_live_cad_parity.py

#!/usr/bin/env python3
"""Low-RAM live CAD parity gate for manifest/model part-key consistency."""

from __future__ import annotations

import argparse
import json
import struct
import sys
from collections import Counter

import requests


DEFAULT_HOST = "http://localhost:8888"
DEFAULT_EMAIL = "admin@hartomat.com"
DEFAULT_PASSWORD = "Admin1234!"
DEFAULT_TIMEOUT = 60


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "Verify that the live CAD scene manifest and served GLB expose the same "
            "renderable part-key set, without duplicates or missing assignments."
        )
    )
    parser.add_argument("--host", default=DEFAULT_HOST, help="Backend base URL.")
    parser.add_argument("--email", default=DEFAULT_EMAIL, help="Login email.")
    parser.add_argument("--password", default=DEFAULT_PASSWORD, help="Login password.")
    parser.add_argument("--cad-id", required=True, help="CAD file id to inspect.")
    parser.add_argument(
        "--timeout",
        type=int,
        default=DEFAULT_TIMEOUT,
        help="HTTP timeout in seconds.",
    )
    parser.add_argument(
        "--allow-extra-non-mesh-keys",
        action="store_true",
        help=(
            "Only enforce mesh-node parity. Kept for forward compatibility if the "
            "manifest ever intentionally contains helper-only entries."
        ),
    )
    return parser.parse_args()


def login(session: requests.Session, *, host: str, email: str, password: str, timeout: int) -> None:
    response = session.post(
        f"{host}/api/auth/login",
        json={"email": email, "password": password},
        timeout=timeout,
    )
    response.raise_for_status()
    payload = response.json()
    session.headers["Authorization"] = f"Bearer {payload['access_token']}"


def fetch_manifest_part_keys(session: requests.Session, *, host: str, cad_id: str, timeout: int) -> set[str]:
    response = session.get(f"{host}/api/cad/{cad_id}/scene-manifest", timeout=timeout)
    response.raise_for_status()
    payload = response.json()
    return {
        str(part["part_key"]).strip()
        for part in payload.get("parts", [])
        if part.get("part_key")
    }


def fetch_glb_payload(session: requests.Session, *, host: str, cad_id: str, timeout: int) -> dict:
    response = session.get(f"{host}/api/cad/{cad_id}/model", timeout=timeout)
    response.raise_for_status()
    data = response.content
    if len(data) < 20:
        raise RuntimeError("GLB payload is too small to contain a JSON chunk header")
    json_len, json_type = struct.unpack_from("<II", data, 12)
    if json_type != 0x4E4F534A:
        raise RuntimeError(f"Unexpected GLB JSON chunk type: {json_type}")
    return json.loads(data[20 : 20 + json_len])


def build_report(manifest_part_keys: set[str], glb_payload: dict) -> dict:
    mesh_nodes = [node for node in glb_payload.get("nodes", []) if "mesh" in node]
    live_part_keys = [
        node.get("extras", {}).get("partKey")
        for node in mesh_nodes
        if node.get("extras", {}).get("partKey")
    ]
    live_part_keys = [str(part_key).strip() for part_key in live_part_keys if str(part_key).strip()]
    unique_live_part_keys = set(live_part_keys)
    duplicate_live_part_keys = {
        key: count for key, count in Counter(live_part_keys).items() if count > 1
    }
    missing_manifest_part_keys = sorted(manifest_part_keys - unique_live_part_keys)
    extra_live_part_keys = sorted(unique_live_part_keys - manifest_part_keys)
    return {
        "manifest_parts": len(manifest_part_keys),
        "mesh_nodes": len(mesh_nodes),
        "live_part_keys": len(live_part_keys),
        "unique_live_part_keys": len(unique_live_part_keys),
        "missing_manifest_part_keys": len(missing_manifest_part_keys),
        "extra_live_part_keys": len(extra_live_part_keys),
        "duplicate_live_part_keys": len(duplicate_live_part_keys),
        "sample_missing": missing_manifest_part_keys[:20],
        "sample_extra": extra_live_part_keys[:20],
        "sample_dupes": list(sorted(duplicate_live_part_keys.items())[:20]),
    }


def evaluate_report(report: dict, *, allow_extra_non_mesh_keys: bool) -> list[str]:
    failures: list[str] = []
    if report["mesh_nodes"] != report["manifest_parts"]:
        failures.append(
            f"mesh_nodes={report['mesh_nodes']} does not match manifest_parts={report['manifest_parts']}"
        )
    if report["live_part_keys"] != report["mesh_nodes"]:
        failures.append(
            f"live_part_keys={report['live_part_keys']} does not match mesh_nodes={report['mesh_nodes']}"
        )
    if report["unique_live_part_keys"] != report["manifest_parts"]:
        failures.append(
            "unique_live_part_keys does not match manifest_parts"
        )
    if report["missing_manifest_part_keys"] != 0:
        failures.append(
            f"missing_manifest_part_keys={report['missing_manifest_part_keys']}"
        )
    if not allow_extra_non_mesh_keys and report["extra_live_part_keys"] != 0:
        failures.append(
            f"extra_live_part_keys={report['extra_live_part_keys']}"
        )
    if report["duplicate_live_part_keys"] != 0:
        failures.append(
            f"duplicate_live_part_keys={report['duplicate_live_part_keys']}"
        )
    return failures


def main() -> int:
    args = parse_args()
    session = requests.Session()
    login(
        session,
        host=args.host,
        email=args.email,
        password=args.password,
        timeout=args.timeout,
    )
    manifest_part_keys = fetch_manifest_part_keys(
        session,
        host=args.host,
        cad_id=args.cad_id,
        timeout=args.timeout,
    )
    glb_payload = fetch_glb_payload(
        session,
        host=args.host,
        cad_id=args.cad_id,
        timeout=args.timeout,
    )
    report = build_report(manifest_part_keys, glb_payload)
    failures = evaluate_report(
        report,
        allow_extra_non_mesh_keys=args.allow_extra_non_mesh_keys,
    )
    print(json.dumps(report, indent=2, sort_keys=True))
    if failures:
        for failure in failures:
            print(f"[cad-parity] {failure}", file=sys.stderr)
        return 1
    return 0


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