Embed catalog meshes as base64 manifest entries (#456)

Author: shayancoin

backend/api/main.py

@@ -26,6 +26,7 @@
 )
 from api.routes import router as api_router
 from api.routes_catalog import router as catalog_router
+from api.routes_catalog import router_mesh as catalog_mesh_router
 from api.routes_cnc import router as cnc_router
 from api.routes_materials import router as materials_router
 from api.routes_modules import router as modules_router
@@ -120,6 +121,8 @@ async def dispatch(self, request: Request, call_next):  # type: ignore[override]
 
 # Include API router
 app.include_router(api_router)
+app.include_router(catalog_router)
+app.include_router(catalog_mesh_router)
 app.include_router(materials_router)
 app.include_router(modules_router)
 app.include_router(pricing_router)

backend/api/routes_catalog.py

@@ -2,18 +2,28 @@
 
 from __future__ import annotations
 
+import base64
+import binascii
 import json
 import os
+import struct
 from functools import lru_cache
 from pathlib import Path
-from typing import Any
+from typing import Any, Dict, Optional
 
 from fastapi import APIRouter, HTTPException
+from fastapi.responses import Response
 
 router = APIRouter(prefix="/api/catalog", tags=["catalog"])
+router_mesh = APIRouter(prefix="/api/catalog", tags=["catalog"])
 
 _DEFAULT_CATALOG_PATH = Path(__file__).resolve().parents[1] / "catalog" / "option_graph.json"
 CATALOG_PATH = Path(os.environ.get("CPQ_CATALOG_PATH") or _DEFAULT_CATALOG_PATH)
+_DEFAULT_MESH_MANIFEST_PATH = (
+    Path(__file__).resolve().parents[2] / "frontend" / "public" / "models" / "manifest.json"
+)
+MESH_MANIFEST_PATH = Path(os.environ.get("CPQ_MESH_MANIFEST_PATH") or _DEFAULT_MESH_MANIFEST_PATH)
+JSON_CHUNK = 0x4E4F534A
 
 
 @lru_cache(maxsize=1)
@@ -27,8 +37,144 @@ def _load_catalog(path: Path) -> Any:
         raise HTTPException(status_code=500, detail="catalog file is invalid") from exc
 
 
+@lru_cache(maxsize=1)
+def _load_mesh_manifest(path: Path) -> Dict[str, Any]:
+    try:
+        with path.open("r", encoding="utf-8") as fh:
+            return json.load(fh)
+    except FileNotFoundError as exc:  # pragma: no cover - protective guard
+        raise HTTPException(status_code=500, detail="mesh manifest not found") from exc
+    except json.JSONDecodeError as exc:  # pragma: no cover - protective guard
+        raise HTTPException(status_code=500, detail="mesh manifest is invalid") from exc
+
+
+def _find_manifest_entry(manifest: Dict[str, Any], variant_id: str) -> Dict[str, Any]:
+    models = manifest.get("models")
+    if not isinstance(models, list):
+        raise HTTPException(status_code=500, detail="mesh manifest missing models")
+
+    primary_match: Optional[Dict[str, Any]] = None
+    for entry in models:
+        if not isinstance(entry, dict):
+            continue
+        entry_id = entry.get("id")
+        if isinstance(entry_id, str) and entry_id == variant_id:
+            return entry
+        variant = entry.get("variant")
+        if isinstance(variant, str) and variant == variant_id:
+            lod = entry.get("lod")
+            if lod in (None, 0):
+                return entry
+            primary_match = entry
+
+    if primary_match is not None:
+        return primary_match
+
+    for entry in models:
+        if not isinstance(entry, dict):
+            continue
+        entry_id = entry.get("id")
+        variant = entry.get("variant")
+        lod = entry.get("lod")
+        if isinstance(entry_id, str) and entry_id == variant_id:
+            return entry
+        if (
+            isinstance(variant, str)
+            and lod is not None
+            and f"{variant}@lod{lod}" == variant_id
+        ):
+            return entry
+
+    raise HTTPException(status_code=404, detail="mesh variant not found")
+
+
+def _decode_mesh_bytes(entry: Dict[str, Any]) -> bytes:
+    mesh_value = entry.get("meshBase64")
+    if not isinstance(mesh_value, str) or not mesh_value:
+        raise HTTPException(status_code=500, detail="mesh manifest missing mesh data")
+
+    try:
+        return base64.b64decode(mesh_value)
+    except binascii.Error as exc:  # pragma: no cover - protective guard
+        raise HTTPException(status_code=500, detail="mesh data is invalid") from exc
+
+
+def _parse_glb_document(data: bytes) -> Dict[str, Any]:
+    if len(data) < 12 or data[:4] != b"glTF":
+        raise HTTPException(status_code=500, detail="invalid GLB header")
+
+    total_length = struct.unpack_from("<I", data, 8)[0]
+    offset = 12
+    while offset + 8 <= len(data) and offset < total_length:
+        chunk_length, chunk_type = struct.unpack_from("<II", data, offset)
+        offset += 8
+        chunk = data[offset : offset + chunk_length]
+        offset += chunk_length
+        if chunk_type == JSON_CHUNK:
+            try:
+                return json.loads(chunk.decode("utf-8"))
+            except json.JSONDecodeError as exc:
+                raise HTTPException(status_code=500, detail="invalid GLB metadata") from exc
+
+    raise HTTPException(status_code=500, detail="GLB missing JSON chunk")
+
+
+def _extract_variant_extras(document: Dict[str, Any]) -> Dict[str, Any]:
+    nodes = document.get("nodes")
+    if isinstance(nodes, list):
+        for node in nodes:
+            if not isinstance(node, dict):
+                continue
+            extras = node.get("extras")
+            if isinstance(extras, dict) and "paform:variant" in extras:
+                return extras
+    raise HTTPException(status_code=500, detail="paform:variant extras missing")
+
+
 @router.get("/option-graph")
 def get_option_graph() -> Any:
     """Return the option graph catalog used by CPQ pricing flows."""
 
     return _load_catalog(CATALOG_PATH)
+
+
+@router_mesh.get("/{variant_id}/mesh")
+def get_catalog_mesh(variant_id: str) -> Dict[str, Any]:
+    """Return the GLB payload and variant metadata for the requested catalog entry."""
+
+    manifest = _load_mesh_manifest(MESH_MANIFEST_PATH)
+    entry = _find_manifest_entry(manifest, variant_id)
+    data = _decode_mesh_bytes(entry)
+
+    document = _parse_glb_document(data)
+    extras = _extract_variant_extras(document)
+    variant_meta = extras.get("paform:variant")
+    if not isinstance(variant_meta, dict):
+        raise HTTPException(status_code=500, detail="mesh missing paform:variant metadata")
+
+    payload: Dict[str, Any] = {
+        "id": entry.get("id"),
+        "variant": variant_meta.get("code") or entry.get("variant") or variant_id,
+        "lod": entry.get("lod"),
+        "sha256": entry.get("sha256"),
+        "bytes": entry.get("bytes") or len(data),
+        "mesh": entry.get("meshBase64"),
+        "metadata": variant_meta,
+        "extras": extras,
+    }
+
+    return payload
+
+
+@router_mesh.get("/{variant_id}/mesh.glb")
+def download_catalog_mesh(variant_id: str) -> Response:
+    """Return the raw GLB binary for the requested catalog entry."""
+
+    manifest = _load_mesh_manifest(MESH_MANIFEST_PATH)
+    entry = _find_manifest_entry(manifest, variant_id)
+    data = _decode_mesh_bytes(entry)
+
+    filename = f"{(entry.get('id') or entry.get('variant') or variant_id)}.glb"
+    headers = {"Content-Disposition": f"attachment; filename=\"{filename}\""}
+
+    return Response(content=data, media_type="model/gltf-binary", headers=headers)

backend/tests/test_routes_catalog.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import base64
+
+from fastapi.testclient import TestClient
+
+from api.main import app
+
+
+def test_get_catalog_mesh_returns_glb_and_metadata() -> None:
+    client = TestClient(app)
+
+    response = client.get("/api/catalog/base_600/mesh")
+    assert response.status_code == 200
+
+    payload = response.json()
+    assert payload["variant"] == "base_600"
+    assert payload["id"] == "base_600"
+    assert isinstance(payload.get("mesh"), str)
+
+    binary = base64.b64decode(payload["mesh"]) if payload.get("mesh") else b""
+    assert len(binary) == payload["bytes"]
+
+    metadata = payload.get("metadata") or {}
+    assert isinstance(metadata, dict)
+    width_param = metadata.get("parameters", {}).get("width")
+    assert isinstance(width_param, dict)
+    assert width_param.get("initial") == 800
+    viewer_transform = metadata.get("viewerTransform")
+    assert isinstance(viewer_transform, dict)
+    assert viewer_transform.get("scaleParameter", {}).get("axis") == "x"
+
+
+def test_get_catalog_mesh_glb_downloads_binary() -> None:
+    client = TestClient(app)
+
+    response = client.get("/api/catalog/base_600/mesh.glb")
+    assert response.status_code == 200
+    assert response.headers["content-type"].startswith("model/gltf-binary")
+    assert response.headers["content-disposition"].endswith("base_600.glb\"")
+    assert len(response.content) > 1000
+
+
+def test_get_catalog_mesh_missing_variant_returns_404() -> None:
+    client = TestClient(app)
+
+    response = client.get("/api/catalog/unknown_variant/mesh")
+    assert response.status_code == 404