import hashlib import json import os import signal import shutil import struct import subprocess import sys import time from datetime import datetime, timezone from multiprocessing import Process from pathlib import Path from typing import Any from OCP.BRepMesh import BRepMesh_IncrementalMesh from OCP.Message import Message_ProgressRange from OCP.RWGltf import RWGltf_CafWriter from OCP.STEPCAFControl import STEPCAFControl_Reader from OCP.TColStd import TColStd_IndexedDataMapOfStringString from OCP.TCollection import TCollection_AsciiString, TCollection_ExtendedString from OCP.TDF import TDF_LabelSequence from OCP.TDocStd import TDocStd_Document from OCP.XCAFApp import XCAFApp_Application from OCP.XCAFDoc import XCAFDoc_DocumentTool, XCAFDoc_ShapeTool CONVERTER_NAME = "NodeDcBimConverter" CONVERTER_VERSION = "0.4.9" UPLOADS_DIR = Path(os.environ.get("NODEDC_BIM_CONVERTER_UPLOADS_DIR", "/beam/uploads")).resolve() POLL_INTERVAL_SECONDS = float(os.environ.get("NODEDC_BIM_CONVERTER_INTERVAL_SECONDS", "10")) PROCESS_ONCE = os.environ.get("NODEDC_BIM_CONVERTER_ONCE", "").lower() in {"1", "true", "yes"} REPROCESS_READY_ON_VERSION_CHANGE = os.environ.get( "NODEDC_BIM_CONVERTER_REPROCESS_READY_ON_VERSION_CHANGE", "" ).lower() in {"1", "true", "yes"} REPROCESS_FAILED_ON_VERSION_CHANGE = os.environ.get( "NODEDC_BIM_CONVERTER_REPROCESS_FAILED_ON_VERSION_CHANGE", "1" ).lower() in {"1", "true", "yes"} XKT_ENABLED = os.environ.get("NODEDC_BIM_CONVERTER_XKT_ENABLED", "1").lower() not in {"0", "false", "no"} XKT_COMMAND = os.environ.get("NODEDC_BIM_CONVERTER_XKT_COMMAND", "xeokit-convert") GLB_XKT_ENABLED = os.environ.get("NODEDC_BIM_CONVERTER_GLB_XKT_ENABLED", "1").lower() not in { "0", "false", "no", } GLB_DRACO_ENABLED = os.environ.get("NODEDC_BIM_CONVERTER_GLB_DRACO_ENABLED", "1").lower() not in { "0", "false", "no", } GLTF_TRANSFORM_COMMAND = os.environ.get("NODEDC_BIM_CONVERTER_GLTF_TRANSFORM_COMMAND", "gltf-transform") NODE_OPTIONS = os.environ.get("NODE_OPTIONS", "--max-old-space-size=8192") GLB_LOD_ENABLED = os.environ.get("NODEDC_BIM_CONVERTER_GLB_LOD_ENABLED", "1").lower() not in { "0", "false", "no", } STEP_MESH_LINEAR_DEFLECTION = float(os.environ.get("NODEDC_BIM_CONVERTER_STEP_LINEAR_DEFLECTION", "0.1")) STEP_MESH_ANGULAR_DEFLECTION = float(os.environ.get("NODEDC_BIM_CONVERTER_STEP_ANGULAR_DEFLECTION", "0.1")) PROCESSING_STALE_SECONDS = float(os.environ.get("NODEDC_BIM_CONVERTER_PROCESSING_STALE_SECONDS", "60")) CONVERSION_MAX_ATTEMPTS = int(os.environ.get("NODEDC_BIM_CONVERTER_MAX_ATTEMPTS", "3")) CONVERSION_TIMEOUT_SECONDS = float(os.environ.get("NODEDC_BIM_CONVERTER_TIMEOUT_SECONDS", "300")) GLB_DRACO_TIMEOUT_SECONDS = float( os.environ.get("NODEDC_BIM_CONVERTER_GLB_DRACO_TIMEOUT_SECONDS", str(CONVERSION_TIMEOUT_SECONDS)) ) GLB_LOD_TIMEOUT_SECONDS = float( os.environ.get("NODEDC_BIM_CONVERTER_GLB_LOD_TIMEOUT_SECONDS", str(max(CONVERSION_TIMEOUT_SECONDS, 900))) ) XKT_TIMEOUT_SECONDS = float( os.environ.get("NODEDC_BIM_CONVERTER_XKT_TIMEOUT_SECONDS", str(max(CONVERSION_TIMEOUT_SECONDS, 900))) ) XCAF_TIMEOUT_SECONDS = float(os.environ.get("NODEDC_BIM_CONVERTER_XCAF_TIMEOUT_SECONDS", str(CONVERSION_TIMEOUT_SECONDS))) CADQUERY_TIMEOUT_SECONDS = float( os.environ.get("NODEDC_BIM_CONVERTER_CADQUERY_TIMEOUT_SECONDS", str(CONVERSION_TIMEOUT_SECONDS)) ) STEP_EXTENSIONS = {".step", ".stp"} GLB_EXTENSIONS = {".glb"} GLB_LOD_PROFILES = [ { "id": "full", "label": "Полный", "suffix": "lod-full", "ratio": None, "error": None, }, { "id": "desktop", "label": "Средний", "suffix": "lod-desktop", "ratio": 0.35, "error": 0.005, }, { "id": "mobile", "label": "Мобильный", "suffix": "lod-mobile", "ratio": 0.12, "error": 0.02, }, ] def now_iso() -> str: return datetime.now(timezone.utc).isoformat() def src_from_path(path: Path) -> str: return f"uploads/{path.resolve().relative_to(UPLOADS_DIR).as_posix()}" def manifest_path(source_path: Path) -> Path: return source_path.with_name(f"{source_path.name}.beam.json") def asset_manifest_path(source_path: Path) -> Path | None: version_dir = source_path.parent asset_dir = version_dir.parent if not version_dir.name.startswith("ver_"): return None if asset_dir == UPLOADS_DIR or not asset_dir.is_dir(): return None return asset_dir / "asset.json" def version_asset_dir(source_path: Path) -> Path | None: version_dir = source_path.parent asset_dir = version_dir.parent if not version_dir.name.startswith("ver_"): return None if asset_dir == UPLOADS_DIR or not asset_dir.is_dir(): return None return asset_dir def read_json(path: Path) -> dict[str, Any]: try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return {} def write_json_atomic(path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) tmp_path = path.with_name(f"{path.name}.tmp") tmp_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") tmp_path.replace(path) def conversion_attempt_count(manifest: dict[str, Any]) -> int: try: return max(0, int(manifest.get("attempts") or 0)) except Exception: return 0 def draco_optimization_attempt_count(manifest: dict[str, Any]) -> int: optimization = manifest.get("dracoOptimization") if not isinstance(optimization, dict): return 0 try: return max(0, int(optimization.get("attempts") or 0)) except Exception: return 0 def xkt_conversion_attempt_count(manifest: dict[str, Any]) -> int: conversion = manifest.get("xktConversion") if not isinstance(conversion, dict): return 0 try: return max(0, int(conversion.get("attempts") or 0)) except Exception: return 0 def lod_generation_attempt_count(manifest: dict[str, Any]) -> int: generation = manifest.get("lodGeneration") if not isinstance(generation, dict): return 0 try: return max(0, int(generation.get("attempts") or 0)) except Exception: return 0 def update_asset_manifest_version(source_path: Path, version_payload: dict[str, Any]) -> None: asset_path = asset_manifest_path(source_path) if not asset_path: return asset = read_json(asset_path) versions = asset.get("versions") if not isinstance(versions, list): versions = [] version_id = version_payload.get("versionId") version_number = version_payload.get("version") next_versions = [] replaced = False for item in versions: if not isinstance(item, dict): continue is_match = (version_id and item.get("versionId") == version_id) or ( version_number and item.get("version") == version_number ) if is_match: next_versions.append({**item, **version_payload}) replaced = True else: next_versions.append(item) if not replaced: next_versions.append(version_payload) next_versions.sort(key=lambda item: int(item.get("version") or 0)) now = now_iso() write_json_atomic( asset_path, { **asset, "assetId": asset.get("assetId") or version_payload.get("assetId"), "projectId": asset.get("projectId") or version_payload.get("projectId"), "currentVersion": asset.get("currentVersion") or version_payload.get("version"), "currentVersionId": asset.get("currentVersionId") or version_payload.get("versionId"), "updatedAt": now, "versions": next_versions, }, ) def file_size_mb(path: Path) -> float: try: return path.stat().st_size / (1024 * 1024) except Exception: return 0.0 def calculate_file_sha256(path: Path) -> str | None: try: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() except Exception: return None def upload_src_exists(src: Any) -> bool: if not isinstance(src, str) or not src.startswith("uploads/"): return False try: return (UPLOADS_DIR / src.removeprefix("uploads/")).resolve().is_file() except Exception: return False def upload_path_from_src(src: Any) -> Path | None: if not isinstance(src, str) or not src.startswith("uploads/"): return None try: path = (UPLOADS_DIR / src.removeprefix("uploads/")).resolve() except Exception: return None try: path.relative_to(UPLOADS_DIR) except ValueError: return None return path if path.is_file() else None def source_path_from_manifest_file(manifest_file: Path, manifest: dict[str, Any]) -> Path | None: source_path = upload_path_from_src(manifest.get("sourceSrc") or manifest.get("downloadSrc")) if source_path: return source_path source_name = manifest_file.name.removesuffix(".beam.json") fallback_path = manifest_file.with_name(source_name) return fallback_path if fallback_path.is_file() else None def link_or_copy_file(source_path: Path, target_path: Path) -> None: if source_path.resolve() == target_path.resolve(): return target_path.parent.mkdir(parents=True, exist_ok=True) try: if target_path.exists(): target_path.unlink() os.link(source_path, target_path) except OSError: shutil.copy2(source_path, target_path) def copy_reusable_conversion_artifacts( source_path: Path, reusable_conversion: dict[str, Any], glb_path: Path, xkt_path: Path, metadata_path: Path, ) -> dict[str, Any]: artifact_source_path = upload_path_from_src(reusable_conversion.get("artifactSrc")) metadata_source_path = upload_path_from_src(reusable_conversion.get("metadataSrc")) fallback_source_path = upload_path_from_src(reusable_conversion.get("fallbackArtifactSrc")) artifact_type = reusable_conversion.get("artifactType") or reusable_conversion.get("targetFormat") or "xkt" artifact_target_path = xkt_path if artifact_type == "xkt" else glb_path if not artifact_source_path or not metadata_source_path: raise RuntimeError("Reusable conversion is missing artifact or metadata file.") if fallback_source_path: link_or_copy_file(fallback_source_path, glb_path) if artifact_source_path != fallback_source_path: link_or_copy_file(artifact_source_path, artifact_target_path) metadata = read_json(metadata_source_path) metadata.update( { "source": src_from_path(source_path), "artifact": src_from_path(artifact_target_path), "artifactType": artifact_type, "glbArtifact": src_from_path(glb_path) if glb_path.exists() else None, "xktArtifact": src_from_path(xkt_path) if artifact_type == "xkt" and xkt_path.exists() else None, "generatedAt": now_iso(), "reusedFrom": { "assetId": reusable_conversion.get("assetId"), "projectId": reusable_conversion.get("projectId"), "version": reusable_conversion.get("version"), "versionId": reusable_conversion.get("versionId"), "sourceSrc": reusable_conversion.get("sourceSrc"), }, } ) converter = metadata.get("converter") if isinstance(converter, dict): metadata["converter"] = { **converter, "name": CONVERTER_NAME, "version": CONVERTER_VERSION, "reusedArtifacts": True, } write_json_atomic(metadata_path, metadata) return { "artifactSrc": src_from_path(artifact_target_path), "artifactType": artifact_type, "componentCount": reusable_conversion.get("componentCount") or metadata.get("componentCount"), "fallbackArtifactSrc": src_from_path(glb_path) if glb_path.exists() else None, "fallbackArtifactType": "gltf" if glb_path.exists() else None, "metadataSrc": src_from_path(metadata_path), } def find_ready_conversion_with_same_hash(source_path: Path, manifest: dict[str, Any]) -> dict[str, Any] | None: sha256 = manifest.get("sha256") or calculate_file_sha256(source_path) if not sha256: return None version_id = manifest.get("versionId") version_number = manifest.get("version") source_manifest_path = manifest_path(source_path).resolve() current_asset_dir = version_asset_dir(source_path) for candidate_manifest_path in sorted(UPLOADS_DIR.rglob("*.beam.json")): if candidate_manifest_path.resolve() == source_manifest_path: continue candidate = read_json(candidate_manifest_path) if candidate.get("status") != "ready": continue candidate_sha256 = candidate.get("sha256") if not candidate_sha256: candidate_source_path = source_path_from_manifest_file(candidate_manifest_path, candidate) if candidate_source_path: candidate_sha256 = calculate_file_sha256(candidate_source_path) if candidate_sha256: candidate = {**candidate, "sha256": candidate_sha256} write_json_atomic(candidate_manifest_path, candidate) if candidate_source_path: update_asset_manifest_version(candidate_source_path, candidate) if candidate_sha256 != sha256: continue if candidate.get("sourceFormat") != "step": continue if candidate.get("artifactType") != "xkt" and candidate.get("targetFormat") != "xkt": continue if current_asset_dir and candidate_manifest_path.parent.parent == current_asset_dir: if candidate.get("versionId") == version_id or candidate.get("version") == version_number: continue if not candidate.get("artifactSrc") or not candidate.get("metadataSrc"): continue if not upload_src_exists(candidate.get("artifactSrc")) or not upload_src_exists(candidate.get("metadataSrc")): continue return candidate return None def hash_id(value: str) -> str: return hashlib.sha1(value.encode("utf-8")).hexdigest()[:16] def json_safe(value: Any) -> Any: if value is None or isinstance(value, (str, int, float, bool)): return value if isinstance(value, dict): return {str(k): json_safe(v) for k, v in value.items()} if isinstance(value, (list, tuple)): return [json_safe(v) for v in value] return str(value) def node_to_metadata(node: Any, parent_path: str = "") -> dict[str, Any]: name = node.name or "root" node_path = f"{parent_path}/{name}" if parent_path else name children = [node_to_metadata(child, node_path) for child in node.children] return { "id": hash_id(node_path), "name": name, "path": node_path, "hasShape": bool(node.obj), "metadata": json_safe(node.metadata or {}), "children": children, } def count_nodes(node: dict[str, Any]) -> int: return 1 + sum(count_nodes(child) for child in node.get("children", [])) def metadata_dict(value: Any) -> dict[str, Any]: safe_value = json_safe(value or {}) return safe_value if isinstance(safe_value, dict) else {"value": safe_value} def get_shape_type(value: Any) -> str | None: try: shape_type = value.ShapeType() return str(shape_type) if shape_type else None except Exception: return None def collect_solids(workplane: Any) -> list[Any]: solids: list[Any] = [] seen: set[int] = set() for value in workplane.vals(): candidates = [] try: candidates = list(value.Solids()) except Exception: candidates = [] if not candidates and get_shape_type(value) == "Solid": candidates = [value] for solid in candidates: key = hash(solid) if key in seen: continue seen.add(key) solids.append(solid) return solids def unique_assembly_name(parent: Any, requested_name: Any) -> str: base_name = str(requested_name or "component") if base_name not in parent.objects: return base_name index = 2 while True: candidate = f"{base_name}__{index:03d}" if candidate not in parent.objects: return candidate index += 1 def import_step_with_unique_component_names(source_path: Path, cq_module: Any) -> Any: original_add = cq_module.Assembly.add def add_with_unique_names(self: Any, arg: Any, *args: Any, **kwargs: Any) -> Any: if isinstance(arg, cq_module.Assembly): positional_name = args[0] if args else None requested_name = kwargs.get("name", positional_name) or arg.name unique_name = unique_assembly_name(self, requested_name) if unique_name != requested_name: arg.metadata = { **metadata_dict(arg.metadata), "originalName": str(requested_name), } if args: args = (unique_name, *args[1:]) else: kwargs = {**kwargs, "name": unique_name} return original_add(self, arg, *args, **kwargs) cq_module.Assembly.add = add_with_unique_names try: return cq_module.Assembly.importStep(str(source_path)) finally: cq_module.Assembly.add = original_add def import_step_assembly(source_path: Path) -> tuple[Any, str, str]: import cadquery as cq try: return cq.Assembly.importStep(str(source_path)), "assembly", getattr(cq, "__version__", "unknown") except ValueError as exc: error_message = str(exc) if "Unique name is required" in error_message and "already in the assembly" in error_message: return ( import_step_with_unique_component_names(source_path, cq), "assembly-unique-names", getattr(cq, "__version__", "unknown"), ) if "does not contain an assembly" not in error_message: raise step_model = cq.importers.importStep(str(source_path)) solids = collect_solids(step_model) assy = cq.Assembly(name=f"{source_path.stem}_root") if len(solids) > 1: for index, solid in enumerate(solids, start=1): assy.add(solid, name=f"{source_path.stem}_solid_{index:03d}") return assy, "split-solids", getattr(cq, "__version__", "unknown") assy.add(step_model, name=source_path.stem) return assy, "single-shape", getattr(cq, "__version__", "unknown") def read_glb_json(glb_path: Path) -> tuple[dict[str, Any], list[tuple[bytes, bytes]]]: data = glb_path.read_bytes() if len(data) < 20: raise ValueError("GLB is too small.") magic, version, declared_length = struct.unpack("<4sII", data[:12]) if magic != b"glTF" or version != 2: raise ValueError("Only binary glTF v2 files are supported.") if declared_length != len(data): raise ValueError("GLB declared length does not match file size.") gltf_json: dict[str, Any] | None = None chunks: list[tuple[bytes, bytes]] = [] offset = 12 while offset < len(data): if offset + 8 > len(data): raise ValueError("Invalid GLB chunk header.") chunk_length, chunk_type = struct.unpack(" None: json_bytes = json.dumps(gltf_json, ensure_ascii=False, separators=(",", ":")).encode("utf-8") json_padding = (4 - (len(json_bytes) % 4)) % 4 json_chunk = json_bytes + (b" " * json_padding) body = struct.pack(" tuple[dict[str, Any], int]: gltf_json, chunks = read_glb_json(glb_path) nodes = gltf_json.get("nodes", []) meshes = gltf_json.get("meshes", []) renamed = 0 for node in nodes: mesh_index = node.get("mesh") if not isinstance(mesh_index, int) or mesh_index < 0 or mesh_index >= len(meshes): continue mesh_name = meshes[mesh_index].get("name") node_name = node.get("name") if mesh_name and (not node_name or str(node_name).startswith("NAUO")): node["name"] = mesh_name renamed += 1 seen_names: dict[str, int] = {} for index, node in enumerate(nodes): base_name = str(node.get("name") or f"node_{index}") occurrence = seen_names.get(base_name, 0) + 1 seen_names[base_name] = occurrence if occurrence > 1: extras = node.get("extras") if not isinstance(extras, dict): extras = {} extras.setdefault("originalName", base_name) node["extras"] = extras node["name"] = f"{base_name}__{occurrence:03d}" renamed += 1 if renamed: write_glb_json(glb_path, gltf_json, chunks) return gltf_json, renamed def gltf_node_name(gltf_json: dict[str, Any], node: dict[str, Any], index: int) -> str: name = node.get("name") if name: return str(name) meshes = gltf_json.get("meshes", []) mesh_index = node.get("mesh") if isinstance(mesh_index, int) and 0 <= mesh_index < len(meshes): mesh_name = meshes[mesh_index].get("name") if mesh_name: return str(mesh_name) return f"node_{index}" def gltf_node_to_metadata(gltf_json: dict[str, Any], node_index: int, parent_path: str = "") -> dict[str, Any]: nodes = gltf_json.get("nodes", []) node = nodes[node_index] name = gltf_node_name(gltf_json, node, node_index) node_path = f"{parent_path}/{name}#{node_index}" if parent_path else f"{name}#{node_index}" children = [ gltf_node_to_metadata(gltf_json, child_index, node_path) for child_index in node.get("children", []) if isinstance(child_index, int) and 0 <= child_index < len(nodes) ] return { "id": hash_id(node_path), "name": name, "path": node_path, "hasShape": isinstance(node.get("mesh"), int), "metadata": {}, "children": children, } def gltf_component_tree(gltf_json: dict[str, Any], fallback_name: str) -> dict[str, Any]: nodes = gltf_json.get("nodes", []) scenes = gltf_json.get("scenes", []) scene_index = gltf_json.get("scene", 0) scene = scenes[scene_index] if isinstance(scene_index, int) and 0 <= scene_index < len(scenes) else {} root_indexes = [idx for idx in scene.get("nodes", []) if isinstance(idx, int) and 0 <= idx < len(nodes)] if len(root_indexes) == 1: return gltf_node_to_metadata(gltf_json, root_indexes[0]) children = [gltf_node_to_metadata(gltf_json, idx, fallback_name) for idx in root_indexes] return { "id": hash_id(fallback_name), "name": fallback_name, "path": fallback_name, "hasShape": False, "metadata": {}, "children": children, } def convert_step_xcaf_to_glb(source_path: Path, glb_path: Path) -> tuple[dict[str, Any], dict[str, Any]]: app = XCAFApp_Application.GetApplication_s() doc = TDocStd_Document(TCollection_ExtendedString("MDTV-XCAF")) app.NewDocument(TCollection_ExtendedString("MDTV-XCAF"), doc) print(f"[{CONVERTER_NAME}] STEPCAF read {source_path.name}", flush=True) reader = STEPCAFControl_Reader() reader.SetNameMode(True) reader.SetColorMode(True) reader.SetLayerMode(True) read_status = reader.ReadFile(str(source_path)) if "RetDone" not in str(read_status): raise ValueError(f"STEPCAF read failed: {read_status}") if not reader.Transfer(doc): raise ValueError("STEPCAF transfer failed.") shape_tool = XCAFDoc_DocumentTool.ShapeTool_s(doc.Main()) labels = TDF_LabelSequence() shape_tool.GetFreeShapes(labels) if labels.Length() == 0: raise ValueError("STEPCAF document has no free shapes.") print( f"[{CONVERTER_NAME}] meshing {source_path.name}: " f"{labels.Length()} free shape(s), " f"linear={STEP_MESH_LINEAR_DEFLECTION}, angular={STEP_MESH_ANGULAR_DEFLECTION}", flush=True, ) for index in range(1, labels.Length() + 1): shape = XCAFDoc_ShapeTool.GetShape_s(labels.Value(index)) if not shape.IsNull(): BRepMesh_IncrementalMesh( shape, STEP_MESH_LINEAR_DEFLECTION, False, STEP_MESH_ANGULAR_DEFLECTION, True, ) glb_path.parent.mkdir(parents=True, exist_ok=True) print(f"[{CONVERTER_NAME}] GLB export {source_path.name}", flush=True) writer = RWGltf_CafWriter(TCollection_AsciiString(str(glb_path)), True) file_info = TColStd_IndexedDataMapOfStringString() if not writer.Perform(doc, file_info, Message_ProgressRange()): raise ValueError("GLB export failed.") if not glb_path.exists() or glb_path.stat().st_size == 0: raise ValueError("GLB export produced no output.") gltf_json, renamed_nodes = normalize_glb_node_names(glb_path) tree = gltf_component_tree(gltf_json, source_path.stem) stats = { "freeShapeCount": labels.Length(), "gltfNodeCount": len(gltf_json.get("nodes", [])), "gltfMeshCount": len(gltf_json.get("meshes", [])), "renamedNodeCount": renamed_nodes, "fileSizeMb": round(file_size_mb(source_path), 2), "linearDeflection": STEP_MESH_LINEAR_DEFLECTION, "angularDeflection": STEP_MESH_ANGULAR_DEFLECTION, } return tree, stats def convert_step_cadquery_to_glb(source_path: Path, glb_path: Path) -> tuple[dict[str, Any], dict[str, Any]]: assy, import_strategy, cadquery_version = import_step_assembly(source_path) tree = node_to_metadata(assy) glb_path.parent.mkdir(parents=True, exist_ok=True) assy.export(str(glb_path), tolerance=STEP_MESH_LINEAR_DEFLECTION, angularTolerance=STEP_MESH_ANGULAR_DEFLECTION) stats = { "fallback": True, "fileSizeMb": round(file_size_mb(source_path), 2), "importStrategy": import_strategy, "cadqueryVersion": cadquery_version, "linearDeflection": STEP_MESH_LINEAR_DEFLECTION, "angularDeflection": STEP_MESH_ANGULAR_DEFLECTION, } return tree, stats def read_log_tail(path: Path, max_lines: int = 30) -> str: try: lines = path.read_text(encoding="utf-8", errors="replace").splitlines() except Exception: return "" return "\n".join(lines[-max_lines:]) def stage_exit_error(stage_name: str, exitcode: int) -> str: if exitcode < 0: signal_number = abs(exitcode) if signal_number == 9: return f"{stage_name} process was killed by signal 9; likely converter memory limit was exceeded." return f"{stage_name} process was killed by signal {signal_number}." if exitcode == 137: return f"{stage_name} process was killed with code 137; likely converter memory limit was exceeded." return f"{stage_name} subprocess exited with code {exitcode}." def is_resource_limit_error(error: Any) -> bool: message = str(error) return ( "signal 9" in message or "code -9" in message or "code 137" in message or "memory limit was exceeded" in message ) def conversion_failed_message(error: Any) -> str: if is_resource_limit_error(error): return "Не удалось подготовить просмотр и дерево компонентов: не хватило памяти конвертера." return "Не удалось подготовить просмотр и дерево компонентов." def convert_glb_to_xkt(glb_path: Path, xkt_path: Path) -> dict[str, Any]: if not XKT_ENABLED: return {"enabled": False} tmp_xkt_path = xkt_path.with_name(f"{xkt_path.name}.tmp") log_path = xkt_path.with_name(f"{xkt_path.name}.convert.log") if tmp_xkt_path.exists(): tmp_xkt_path.unlink() env = {**os.environ, "NODE_OPTIONS": NODE_OPTIONS} command = [ XKT_COMMAND, "-s", str(glb_path), "-f", "glb", "-o", str(tmp_xkt_path), "-t", "-n", "-e", "0", "-b", ] print(f"[{CONVERTER_NAME}] XKT export {glb_path.name}", flush=True) with log_path.open("w", encoding="utf-8", errors="replace") as log_file: try: result = subprocess.run( command, cwd=str(glb_path.parent), env=env, stdout=log_file, stderr=subprocess.STDOUT, text=True, timeout=XKT_TIMEOUT_SECONDS, check=False, ) except subprocess.TimeoutExpired as exc: try: tmp_xkt_path.unlink() except Exception: pass raise TimeoutError(f"XKT export timed out after {int(XKT_TIMEOUT_SECONDS)} seconds.") from exc if result.returncode != 0: tail = read_log_tail(log_path) raise RuntimeError(f"XKT export failed with code {result.returncode}: {tail}") if not tmp_xkt_path.exists() or tmp_xkt_path.stat().st_size == 0: raise RuntimeError("XKT export produced no output.") tmp_xkt_path.replace(xkt_path) try: log_path.unlink() except Exception: pass return { "enabled": True, "command": XKT_COMMAND, "nodeOptions": NODE_OPTIONS, "edgeBuffers": False, "textures": False, "normals": False, "compressedBuffers": True, "sizeBytes": xkt_path.stat().st_size, } def build_step_metadata( source_path: Path, glb_path: Path, xkt_path: Path, metadata_path: Path, *, tree: dict[str, Any], stats: dict[str, Any], import_strategy: str, ) -> dict[str, Any]: xkt_stats = convert_glb_to_xkt(glb_path, xkt_path) has_xkt = bool(xkt_stats.get("enabled") and xkt_path.exists()) preferred_artifact_path = xkt_path if has_xkt else glb_path preferred_artifact_type = "xkt" if has_xkt else "gltf" metadata = { "source": src_from_path(source_path), "artifact": src_from_path(preferred_artifact_path), "artifactType": preferred_artifact_type, "glbArtifact": src_from_path(glb_path), "xktArtifact": src_from_path(xkt_path) if has_xkt else None, "format": "step", "targetFormat": "xkt" if has_xkt else "glb", "importStrategy": import_strategy, "componentTree": tree, "componentCount": count_nodes(tree), "generatedAt": now_iso(), "converter": { "name": CONVERTER_NAME, "version": CONVERTER_VERSION, "engine": "occt-xcaf" if import_strategy == "occt-xcaf-gltf" else "cadquery", "cadqueryVersion": stats.get("cadqueryVersion"), }, "stats": stats, "xkt": xkt_stats, } write_json_atomic(metadata_path, metadata) return metadata def convert_step_xcaf_to_glb_process(source_path: str, glb_path: str, result_path: str) -> None: try: tree, stats = convert_step_xcaf_to_glb(Path(source_path), Path(glb_path)) write_json_atomic(Path(result_path), {"ok": True, "tree": tree, "stats": stats}) except Exception as exc: write_json_atomic(Path(result_path), {"ok": False, "error": str(exc)}) def convert_step_cadquery_to_glb_process(source_path: str, glb_path: str, result_path: str) -> None: try: stage_path = Path(__file__).with_name("cadquery_stage.py") completed = subprocess.run( [ sys.executable, str(stage_path), source_path, glb_path, result_path, str(STEP_MESH_LINEAR_DEFLECTION), str(STEP_MESH_ANGULAR_DEFLECTION), ], check=False, ) if completed.returncode != 0: result = read_json(Path(result_path)) raise RuntimeError(result.get("error") or f"CadQuery subprocess exited with code {completed.returncode}.") except Exception as exc: write_json_atomic(Path(result_path), {"ok": False, "error": str(exc)}) def run_cadquery_stage_with_timeout(result_path: Path, source_path: Path, glb_path: Path) -> dict[str, Any]: if result_path.exists(): result_path.unlink() stage_path = Path(__file__).with_name("cadquery_stage.py") command = [ sys.executable, str(stage_path), str(source_path), str(glb_path), str(result_path), str(STEP_MESH_LINEAR_DEFLECTION), str(STEP_MESH_ANGULAR_DEFLECTION), ] process = subprocess.Popen(command, start_new_session=True) try: if CADQUERY_TIMEOUT_SECONDS <= 0: process.wait() else: process.wait(timeout=CADQUERY_TIMEOUT_SECONDS) except subprocess.TimeoutExpired: try: os.killpg(process.pid, signal.SIGTERM) except Exception: process.terminate() try: process.wait(timeout=10) except subprocess.TimeoutExpired: try: os.killpg(process.pid, signal.SIGKILL) except Exception: process.kill() process.wait(timeout=10) raise TimeoutError(f"CadQuery timed out after {int(CADQUERY_TIMEOUT_SECONDS)} seconds.") result = read_json(result_path) try: result_path.unlink() except Exception: pass if process.returncode != 0: raise RuntimeError(result.get("error") or stage_exit_error("CadQuery", process.returncode)) if not result: raise RuntimeError("CadQuery produced no result.") if not result.get("ok"): raise RuntimeError(str(result.get("error") or "CadQuery failed.")) return result def run_step_glb_stage_with_timeout( stage_name: str, timeout_seconds: float, target: Any, result_path: Path, *args: str, ) -> dict[str, Any]: if result_path.exists(): result_path.unlink() if timeout_seconds <= 0: target(*args, str(result_path)) else: process = Process(target=target, args=(*args, str(result_path))) process.start() process.join(timeout_seconds) if process.is_alive(): process.terminate() process.join(10) if process.is_alive(): process.kill() process.join(10) raise TimeoutError(f"{stage_name} timed out after {int(timeout_seconds)} seconds.") if process.exitcode != 0: result = read_json(result_path) error = result.get("error") if isinstance(result, dict) else None raise RuntimeError(error or stage_exit_error(stage_name, process.exitcode)) result = read_json(result_path) try: result_path.unlink() except Exception: pass if not result: raise RuntimeError(f"{stage_name} produced no result.") if not result.get("ok"): raise RuntimeError(str(result.get("error") or f"{stage_name} failed.")) return result def remove_partial_artifacts(*paths: Path) -> None: for path in paths: try: if path.exists(): path.unlink() except Exception: pass def convert_step_assets(source_path: Path, glb_path: Path, xkt_path: Path, metadata_path: Path) -> dict[str, Any]: previous_metadata = read_json(metadata_path) if XKT_ENABLED and glb_path.exists() and previous_metadata.get("componentTree"): print(f"[{CONVERTER_NAME}] upgrading existing GLB to XKT {glb_path.name}", flush=True) try: gltf_json, renamed_nodes = normalize_glb_node_names(glb_path) except Exception as exc: print(f"[{CONVERTER_NAME}] existing GLB normalize failed for {glb_path}: {exc}", file=sys.stderr, flush=True) gltf_json = {} renamed_nodes = 0 previous_stats = previous_metadata.get("stats") if isinstance(previous_metadata.get("stats"), dict) else {} stats = { **previous_stats, "gltfNodeCount": len(gltf_json.get("nodes", [])) or previous_stats.get("gltfNodeCount"), "gltfMeshCount": len(gltf_json.get("meshes", [])) or previous_stats.get("gltfMeshCount"), "renamedNodeCount": (previous_stats.get("renamedNodeCount") or 0) + renamed_nodes, "reusedExistingGlb": True, } return build_step_metadata( source_path, glb_path, xkt_path, metadata_path, tree=previous_metadata["componentTree"], stats=stats, import_strategy=previous_metadata.get("importStrategy") or "existing-glb", ) import_strategy = "occt-xcaf-gltf" result_path = metadata_path.with_name(f"{metadata_path.name}.stage-result.json") try: result = run_step_glb_stage_with_timeout( "OCCT/XCAF", XCAF_TIMEOUT_SECONDS, convert_step_xcaf_to_glb_process, result_path, str(source_path), str(glb_path), ) tree = result["tree"] stats = result["stats"] except Exception as exc: print( f"[{CONVERTER_NAME}] OCCT/XCAF path failed for {source_path}: {exc}; " "falling back to CadQuery", file=sys.stderr, flush=True, ) remove_partial_artifacts(glb_path, xkt_path) result = run_cadquery_stage_with_timeout(result_path, source_path, glb_path) tree = result["tree"] stats = result["stats"] import_strategy = stats.get("importStrategy") or "cadquery" stats = { **stats, "fallbackFrom": "occt-xcaf-gltf", "fallbackReason": str(exc), } return build_step_metadata( source_path, glb_path, xkt_path, metadata_path, tree=tree, stats=stats, import_strategy=import_strategy, ) def should_process(source_path: Path, manifest: dict[str, Any], glb_path: Path, xkt_path: Path) -> bool: status = manifest.get("status") ready_artifact_path = xkt_path if XKT_ENABLED else glb_path if status == "ready" and (ready_artifact_path.exists() or upload_src_exists(manifest.get("artifactSrc"))): return REPROCESS_READY_ON_VERSION_CHANGE and manifest.get("converterVersion") != CONVERTER_VERSION if status == "failed": return ( REPROCESS_FAILED_ON_VERSION_CHANGE and manifest.get("converterVersion") != CONVERTER_VERSION and source_path.exists() ) if status == "processing": updated_at = manifest.get("updatedAt") if updated_at: try: updated = datetime.fromisoformat(str(updated_at).replace("Z", "+00:00")) if (datetime.now(timezone.utc) - updated).total_seconds() < PROCESSING_STALE_SECONDS: return False except Exception: pass return source_path.exists() def process_one(source_path: Path) -> bool: manifest_file = manifest_path(source_path) stem = source_path.stem glb_path = source_path.with_name(f"{stem}.glb") xkt_path = source_path.with_name(f"{stem}.xkt") metadata_path = source_path.with_name(f"{stem}.metadata.json") manifest = read_json(manifest_file) if not should_process(source_path, manifest, glb_path, xkt_path): return False preserved_fields = { key: manifest.get(key) for key in ( "assetId", "projectId", "version", "versionId", "originalFilename", "storedFilename", "size", "sha256", "uploadedAt", "downloadSrc", ) if manifest.get(key) is not None } base_manifest = { **preserved_fields, "createdAt": manifest.get("createdAt") or now_iso(), "componentTreeRequired": True, "converterName": CONVERTER_NAME, "converterVersion": CONVERTER_VERSION, "sourceFormat": "step", "sourceSrc": src_from_path(source_path), "targetFormat": "xkt" if XKT_ENABLED else "glb", } attempts = 0 if manifest.get("converterVersion") != CONVERTER_VERSION else conversion_attempt_count(manifest) if manifest.get("status") == "processing" and attempts >= CONVERSION_MAX_ATTEMPTS: failed_manifest = { **base_manifest, "attempts": attempts, "error": f"Converter process stopped before completion after {attempts} attempt(s).", "message": "Не удалось подготовить просмотр и дерево компонентов: превышен лимит попыток конвертации.", "status": "failed", "updatedAt": now_iso(), } write_json_atomic(manifest_file, failed_manifest) update_asset_manifest_version(source_path, failed_manifest) print( f"[{CONVERTER_NAME}] failed {source_path}: max attempts reached ({attempts})", file=sys.stderr, flush=True, ) return False reusable_conversion = find_ready_conversion_with_same_hash(source_path, manifest) if reusable_conversion: reused_artifacts = copy_reusable_conversion_artifacts( source_path, reusable_conversion, glb_path, xkt_path, metadata_path, ) ready_manifest = { **base_manifest, "attempts": attempts, **reused_artifacts, "message": "Просмотр и дерево компонентов готовы.", "reusedFromAssetId": reusable_conversion.get("assetId"), "reusedFromProjectId": reusable_conversion.get("projectId"), "reusedFromVersion": reusable_conversion.get("version"), "reusedFromVersionId": reusable_conversion.get("versionId"), "reusedFromSourceSrc": reusable_conversion.get("sourceSrc"), "status": "ready", "updatedAt": now_iso(), } write_json_atomic(manifest_file, ready_manifest) update_asset_manifest_version(source_path, ready_manifest) reuse_label = ( reusable_conversion.get("versionId") or reusable_conversion.get("version") or reusable_conversion.get("sourceSrc") or "ready-artifact" ) print( f"[{CONVERTER_NAME}] reused ready conversion {source_path.name} " f"from {reuse_label}", flush=True, ) return True attempt = attempts + 1 write_json_atomic( manifest_file, { **base_manifest, "attempts": attempt, "message": "Подготавливаем просмотр и дерево компонентов.", "status": "processing", "updatedAt": now_iso(), }, ) print(f"[{CONVERTER_NAME}] converting {source_path}", flush=True) try: metadata = convert_step_assets(source_path, glb_path, xkt_path, metadata_path) artifact_src = metadata.get("artifact") or src_from_path(xkt_path if xkt_path.exists() else glb_path) artifact_type = metadata.get("artifactType") or ("xkt" if xkt_path.exists() else "gltf") ready_manifest = { **base_manifest, "attempts": attempt, "artifactSrc": artifact_src, "artifactType": artifact_type, "fallbackArtifactSrc": src_from_path(glb_path), "fallbackArtifactType": "gltf", "componentCount": metadata.get("componentCount"), "metadataSrc": src_from_path(metadata_path), "message": "Просмотр и дерево компонентов готовы.", "status": "ready", "updatedAt": now_iso(), } write_json_atomic(manifest_file, ready_manifest) update_asset_manifest_version(source_path, ready_manifest) print(f"[{CONVERTER_NAME}] ready {artifact_src}", flush=True) return True except Exception as exc: failed_manifest = { **base_manifest, "attempts": attempt, "error": str(exc), "message": conversion_failed_message(exc), "status": "failed", "updatedAt": now_iso(), } write_json_atomic(manifest_file, failed_manifest) update_asset_manifest_version(source_path, failed_manifest) print(f"[{CONVERTER_NAME}] failed {source_path}: {exc}", file=sys.stderr, flush=True) return False def limited_process_output(value: Any, limit: int = 4000) -> str | None: if value is None: return None text = str(value).strip() if not text: return None return text if len(text) <= limit else f"{text[:limit]}..." def glb_draco_artifact_path(source_path: Path) -> Path: return source_path.with_name(f"{source_path.stem}.draco.glb") def glb_xkt_artifact_path(source_path: Path) -> Path: return source_path.with_name(f"{source_path.stem}.xkt") def glb_lod_artifact_path(source_path: Path, profile: dict[str, Any]) -> Path: return source_path.with_name(f"{source_path.stem}.{profile['suffix']}.glb") def lod_records_ready(lods: Any) -> bool: if not isinstance(lods, list) or len(lods) < len(GLB_LOD_PROFILES): return False by_id = { str(item.get("id")): item for item in lods if isinstance(item, dict) and item.get("id") } for profile in GLB_LOD_PROFILES: item = by_id.get(profile["id"]) if not item or not upload_src_exists(item.get("artifactSrc")): return False return True def should_optimize_glb(source_path: Path, manifest: dict[str, Any], artifact_path: Path) -> bool: if not GLB_DRACO_ENABLED or not source_path.exists() or not manifest: return False if source_path.name.endswith(".draco.glb"): return False source_format = str(manifest.get("sourceFormat") or "").replace(".", "").lower() if source_format != "glb": return False if manifest.get("status") not in {None, "ready"}: return False optimization = manifest.get("dracoOptimization") if not isinstance(optimization, dict): optimization = {} optimization_status = optimization.get("status") optimization_version = optimization.get("converterVersion") artifact_ready = upload_src_exists(manifest.get("artifactSrc")) if optimization_status == "ready" and artifact_ready: return REPROCESS_READY_ON_VERSION_CHANGE and optimization_version != CONVERTER_VERSION if optimization_status == "failed": attempts = draco_optimization_attempt_count(manifest) if attempts >= CONVERSION_MAX_ATTEMPTS: return REPROCESS_FAILED_ON_VERSION_CHANGE and optimization_version != CONVERTER_VERSION return True if optimization_status == "processing": updated_at = optimization.get("updatedAt") or manifest.get("updatedAt") if updated_at: try: updated = datetime.fromisoformat(str(updated_at).replace("Z", "+00:00")) if (datetime.now(timezone.utc) - updated).total_seconds() < PROCESSING_STALE_SECONDS: return False except Exception: pass return True def optimize_glb_one(source_path: Path) -> bool: manifest_file = manifest_path(source_path) if not manifest_file.exists(): return False manifest = read_json(manifest_file) artifact_path = glb_draco_artifact_path(source_path) if not should_optimize_glb(source_path, manifest, artifact_path): return False source_src = src_from_path(source_path) base_manifest = { **manifest, "createdAt": manifest.get("createdAt") or now_iso(), "downloadSrc": manifest.get("downloadSrc") or source_src, "sourceFormat": "glb", "sourceSrc": manifest.get("sourceSrc") or source_src, "status": "ready", } attempts = 0 optimization = manifest.get("dracoOptimization") if isinstance(optimization, dict) and optimization.get("converterVersion") == CONVERTER_VERSION: attempts = draco_optimization_attempt_count(manifest) if isinstance(optimization, dict) and optimization.get("status") == "processing" and attempts >= CONVERSION_MAX_ATTEMPTS: failed_manifest = { **base_manifest, "message": "Model is ready. Draco optimization did not finish; original GLB is used.", "dracoOptimization": { **optimization, "status": "failed", "attempts": attempts, "converterName": CONVERTER_NAME, "converterVersion": CONVERTER_VERSION, "error": f"GLB optimization stopped before completion after {attempts} attempt(s).", "updatedAt": now_iso(), }, "updatedAt": now_iso(), } write_json_atomic(manifest_file, failed_manifest) update_asset_manifest_version(source_path, failed_manifest) print( f"[{CONVERTER_NAME}] GLB Draco failed {source_path}: max attempts reached ({attempts})", file=sys.stderr, flush=True, ) return False attempt = attempts + 1 command = [GLTF_TRANSFORM_COMMAND, "draco", str(source_path), str(artifact_path.with_suffix(".tmp.glb"))] processing_manifest = { **base_manifest, "message": "Model is ready. Optimizing GLB geometry with Draco.", "dracoOptimization": { "status": "processing", "attempts": attempt, "converterName": CONVERTER_NAME, "converterVersion": CONVERTER_VERSION, "command": command, "updatedAt": now_iso(), }, "updatedAt": now_iso(), } write_json_atomic(manifest_file, processing_manifest) update_asset_manifest_version(source_path, processing_manifest) tmp_path = artifact_path.with_suffix(".tmp.glb") print(f"[{CONVERTER_NAME}] optimizing GLB with Draco {source_path}", flush=True) try: if tmp_path.exists(): tmp_path.unlink() run_env = os.environ.copy() run_env["NODE_OPTIONS"] = NODE_OPTIONS result = subprocess.run( command, cwd=str(Path(__file__).resolve().parent), env=run_env, text=True, capture_output=True, timeout=GLB_DRACO_TIMEOUT_SECONDS, check=False, ) if result.returncode != 0: raise RuntimeError( limited_process_output(result.stderr) or limited_process_output(result.stdout) or f"gltf-transform exited with code {result.returncode}" ) if not tmp_path.exists() or tmp_path.stat().st_size <= 0: raise RuntimeError("gltf-transform did not produce a Draco GLB artifact") tmp_path.replace(artifact_path) source_size = source_path.stat().st_size artifact_size = artifact_path.stat().st_size draco_artifact_src = src_from_path(artifact_path) existing_artifact_src = manifest.get("artifactSrc") existing_artifact_type = manifest.get("artifactType") or manifest.get("targetFormat") preserve_xkt_artifact = ( existing_artifact_type == "xkt" and isinstance(existing_artifact_src, str) and upload_src_exists(existing_artifact_src) ) preferred_artifact_src = existing_artifact_src if preserve_xkt_artifact else draco_artifact_src preferred_artifact_type = "xkt" if preserve_xkt_artifact else "gltf" ready_manifest = { **base_manifest, "artifactSha256": calculate_file_sha256(artifact_path), "artifactSrc": preferred_artifact_src, "artifactType": preferred_artifact_type, "dracoArtifactSrc": draco_artifact_src, "dracoArtifactType": "gltf", "fallbackArtifactSrc": source_src, "fallbackArtifactType": "gltf", "message": "Model is ready. GLB geometry optimized with Draco.", "targetFormat": "xkt" if preserve_xkt_artifact else "gltf", "dracoOptimization": { "status": "ready", "attempts": attempt, "artifactSrc": draco_artifact_src, "artifactType": "gltf", "converterName": CONVERTER_NAME, "converterVersion": CONVERTER_VERSION, "command": command, "sourceSize": source_size, "artifactSize": artifact_size, "compressionRatio": round(artifact_size / source_size, 6) if source_size else None, "stdout": limited_process_output(result.stdout), "stderr": limited_process_output(result.stderr), "updatedAt": now_iso(), }, "updatedAt": now_iso(), } write_json_atomic(manifest_file, ready_manifest) update_asset_manifest_version(source_path, ready_manifest) print( f"[{CONVERTER_NAME}] GLB Draco ready {draco_artifact_src} " f"({source_size} -> {artifact_size} bytes)", flush=True, ) return True except Exception as exc: if tmp_path.exists(): try: tmp_path.unlink() except Exception: pass failed_manifest = { **base_manifest, "message": "Model is ready. Draco optimization failed; original GLB is used.", "dracoOptimization": { "status": "failed", "attempts": attempt, "converterName": CONVERTER_NAME, "converterVersion": CONVERTER_VERSION, "command": command, "error": str(exc), "updatedAt": now_iso(), }, "updatedAt": now_iso(), } write_json_atomic(manifest_file, failed_manifest) update_asset_manifest_version(source_path, failed_manifest) print(f"[{CONVERTER_NAME}] GLB Draco failed {source_path}: {exc}", file=sys.stderr, flush=True) return False def should_convert_glb_to_xkt(source_path: Path, manifest: dict[str, Any], xkt_path: Path) -> bool: if not XKT_ENABLED or not GLB_XKT_ENABLED or not source_path.exists() or not manifest: return False if source_path.name.endswith(".draco.glb") or ".lod-" in source_path.name: return False source_format = str(manifest.get("sourceFormat") or "").replace(".", "").lower() if source_format != "glb": return False if manifest.get("status") not in {None, "ready"}: return False conversion = manifest.get("xktConversion") if not isinstance(conversion, dict): conversion = {} conversion_status = conversion.get("status") conversion_version = conversion.get("converterVersion") artifact_src = manifest.get("artifactSrc") artifact_type = manifest.get("artifactType") or manifest.get("targetFormat") artifact_ready = artifact_type == "xkt" and isinstance(artifact_src, str) and upload_src_exists(artifact_src) if conversion_status == "ready" and artifact_ready: return REPROCESS_READY_ON_VERSION_CHANGE and conversion_version != CONVERTER_VERSION if conversion_status == "ready" and xkt_path.exists() and not artifact_ready: return True if conversion_status == "failed": attempts = xkt_conversion_attempt_count(manifest) if attempts >= CONVERSION_MAX_ATTEMPTS: return REPROCESS_FAILED_ON_VERSION_CHANGE and conversion_version != CONVERTER_VERSION return True if conversion_status == "processing": updated_at = conversion.get("updatedAt") or manifest.get("updatedAt") if updated_at: try: updated = datetime.fromisoformat(str(updated_at).replace("Z", "+00:00")) if (datetime.now(timezone.utc) - updated).total_seconds() < PROCESSING_STALE_SECONDS: return False except Exception: pass return True return True def convert_glb_to_xkt_one(source_path: Path) -> bool: manifest_file = manifest_path(source_path) if not manifest_file.exists(): return False manifest = read_json(manifest_file) xkt_path = glb_xkt_artifact_path(source_path) if not should_convert_glb_to_xkt(source_path, manifest, xkt_path): return False source_src = src_from_path(source_path) base_manifest = { **manifest, "createdAt": manifest.get("createdAt") or now_iso(), "downloadSrc": manifest.get("downloadSrc") or source_src, "sourceFormat": "glb", "sourceSrc": manifest.get("sourceSrc") or source_src, "status": "ready", } attempts = 0 conversion = manifest.get("xktConversion") if isinstance(conversion, dict) and conversion.get("converterVersion") == CONVERTER_VERSION: attempts = xkt_conversion_attempt_count(manifest) if isinstance(conversion, dict) and conversion.get("status") == "processing" and attempts >= CONVERSION_MAX_ATTEMPTS: failed_manifest = { **base_manifest, "message": "Model is ready. XKT conversion did not finish; GLB fallback is used.", "xktConversion": { **conversion, "status": "failed", "attempts": attempts, "converterName": CONVERTER_NAME, "converterVersion": CONVERTER_VERSION, "error": f"XKT conversion stopped before completion after {attempts} attempt(s).", "updatedAt": now_iso(), }, "updatedAt": now_iso(), } write_json_atomic(manifest_file, failed_manifest) update_asset_manifest_version(source_path, failed_manifest) print( f"[{CONVERTER_NAME}] GLB XKT failed {source_path}: max attempts reached ({attempts})", file=sys.stderr, flush=True, ) return False attempt = attempts + 1 command = [ XKT_COMMAND, "-s", str(source_path), "-f", "glb", "-o", str(xkt_path.with_name(f"{xkt_path.name}.tmp")), "-t", "-n", "-e", "0", "-b", ] processing_manifest = { **base_manifest, "message": "Model is ready. Converting GLB to XKT viewer artifact.", "targetFormat": "xkt", "xktConversion": { "status": "processing", "attempts": attempt, "converterName": CONVERTER_NAME, "converterVersion": CONVERTER_VERSION, "command": command, "updatedAt": now_iso(), }, "updatedAt": now_iso(), } write_json_atomic(manifest_file, processing_manifest) update_asset_manifest_version(source_path, processing_manifest) print(f"[{CONVERTER_NAME}] converting GLB to XKT {source_path}", flush=True) try: xkt_stats = convert_glb_to_xkt(source_path, xkt_path) source_size = source_path.stat().st_size artifact_size = xkt_path.stat().st_size xkt_artifact_src = src_from_path(xkt_path) fallback_artifact_src = ( manifest.get("dracoArtifactSrc") or ( manifest.get("artifactSrc") if (manifest.get("artifactType") or manifest.get("targetFormat")) == "gltf" else None ) or source_src ) ready_manifest = { **base_manifest, "artifactSha256": calculate_file_sha256(xkt_path), "artifactSrc": xkt_artifact_src, "artifactType": "xkt", "fallbackArtifactSrc": fallback_artifact_src, "fallbackArtifactType": "gltf", "glbArtifactSrc": source_src, "targetFormat": "xkt", "xktArtifactSrc": xkt_artifact_src, "xkt": xkt_stats, "xktConversion": { "status": "ready", "attempts": attempt, "artifactSrc": xkt_artifact_src, "artifactType": "xkt", "converterName": CONVERTER_NAME, "converterVersion": CONVERTER_VERSION, "command": command, "sourceSize": source_size, "artifactSize": artifact_size, "compressionRatio": round(artifact_size / source_size, 6) if source_size else None, "updatedAt": now_iso(), }, "message": "Model is ready. XKT viewer artifact generated.", "updatedAt": now_iso(), } write_json_atomic(manifest_file, ready_manifest) update_asset_manifest_version(source_path, ready_manifest) print( f"[{CONVERTER_NAME}] GLB XKT ready {xkt_artifact_src} " f"({source_size} -> {artifact_size} bytes)", flush=True, ) return True except Exception as exc: tmp_path = xkt_path.with_name(f"{xkt_path.name}.tmp") if tmp_path.exists(): try: tmp_path.unlink() except Exception: pass failed_manifest = { **base_manifest, "message": "Model is ready. XKT conversion failed; GLB fallback is used.", "xktConversion": { "status": "failed", "attempts": attempt, "converterName": CONVERTER_NAME, "converterVersion": CONVERTER_VERSION, "command": command, "error": str(exc), "updatedAt": now_iso(), }, "updatedAt": now_iso(), } write_json_atomic(manifest_file, failed_manifest) update_asset_manifest_version(source_path, failed_manifest) print(f"[{CONVERTER_NAME}] GLB XKT failed {source_path}: {exc}", file=sys.stderr, flush=True) return False def should_generate_glb_lods(source_path: Path, manifest: dict[str, Any]) -> bool: if not GLB_LOD_ENABLED or not source_path.exists() or not manifest: return False if source_path.name.endswith(".draco.glb") or ".lod-" in source_path.name: return False source_format = str(manifest.get("sourceFormat") or "").replace(".", "").lower() if source_format != "glb": return False if manifest.get("status") not in {None, "ready"}: return False generation = manifest.get("lodGeneration") if not isinstance(generation, dict): return False generation_status = generation.get("status") generation_version = generation.get("converterVersion") if generation_status == "ready" and lod_records_ready(manifest.get("lods")): return REPROCESS_READY_ON_VERSION_CHANGE and generation_version != CONVERTER_VERSION if generation_status == "failed": attempts = lod_generation_attempt_count(manifest) if attempts >= CONVERSION_MAX_ATTEMPTS: return REPROCESS_FAILED_ON_VERSION_CHANGE and generation_version != CONVERTER_VERSION return True if generation_status == "processing": updated_at = generation.get("updatedAt") or manifest.get("updatedAt") if updated_at: try: updated = datetime.fromisoformat(str(updated_at).replace("Z", "+00:00")) if (datetime.now(timezone.utc) - updated).total_seconds() < PROCESSING_STALE_SECONDS: return False except Exception: pass return True return generation_status == "queued" def glb_lod_command(source_path: Path, target_path: Path, profile: dict[str, Any]) -> list[str]: if profile.get("ratio") is None: return [GLTF_TRANSFORM_COMMAND, "draco", str(source_path), str(target_path)] return [ GLTF_TRANSFORM_COMMAND, "optimize", str(source_path), str(target_path), "--compress", "draco", "--texture-compress", "false", "--simplify-ratio", str(profile["ratio"]), "--simplify-error", str(profile["error"]), ] def run_gltf_transform_lod(source_path: Path, target_path: Path, profile: dict[str, Any]) -> tuple[list[str], str, str]: tmp_path = target_path.with_suffix(".tmp.glb") if tmp_path.exists(): tmp_path.unlink() command = glb_lod_command(source_path, tmp_path, profile) run_env = os.environ.copy() run_env["NODE_OPTIONS"] = NODE_OPTIONS result = subprocess.run( command, cwd=str(Path(__file__).resolve().parent), env=run_env, text=True, capture_output=True, timeout=GLB_LOD_TIMEOUT_SECONDS, check=False, ) if result.returncode != 0: raise RuntimeError( limited_process_output(result.stderr) or limited_process_output(result.stdout) or f"gltf-transform exited with code {result.returncode}" ) if not tmp_path.exists() or tmp_path.stat().st_size <= 0: raise RuntimeError("gltf-transform did not produce a LOD GLB artifact") tmp_path.replace(target_path) return command, limited_process_output(result.stdout), limited_process_output(result.stderr) def generate_glb_lods_one(source_path: Path) -> bool: manifest_file = manifest_path(source_path) if not manifest_file.exists(): return False manifest = read_json(manifest_file) if not should_generate_glb_lods(source_path, manifest): return False source_src = src_from_path(source_path) base_manifest = { **manifest, "createdAt": manifest.get("createdAt") or now_iso(), "downloadSrc": manifest.get("downloadSrc") or source_src, "sourceFormat": "glb", "sourceSrc": manifest.get("sourceSrc") or source_src, "status": "ready", } attempts = 0 generation = manifest.get("lodGeneration") if isinstance(generation, dict) and generation.get("converterVersion") == CONVERTER_VERSION: attempts = lod_generation_attempt_count(manifest) if isinstance(generation, dict) and generation.get("status") == "processing" and attempts >= CONVERSION_MAX_ATTEMPTS: failed_manifest = { **base_manifest, "message": "Model is ready. LOD generation did not finish.", "lodGeneration": { **generation, "status": "failed", "attempts": attempts, "converterName": CONVERTER_NAME, "converterVersion": CONVERTER_VERSION, "error": f"LOD generation stopped before completion after {attempts} attempt(s).", "updatedAt": now_iso(), }, "updatedAt": now_iso(), } write_json_atomic(manifest_file, failed_manifest) update_asset_manifest_version(source_path, failed_manifest) print( f"[{CONVERTER_NAME}] GLB LOD failed {source_path}: max attempts reached ({attempts})", file=sys.stderr, flush=True, ) return False attempt = attempts + 1 processing_manifest = { **base_manifest, "message": "Model is ready. Generating GLB LOD levels.", "lodGeneration": { **(generation if isinstance(generation, dict) else {}), "status": "processing", "attempts": attempt, "converterName": CONVERTER_NAME, "converterVersion": CONVERTER_VERSION, "profiles": [ {key: profile[key] for key in ("id", "label", "ratio", "error") if key in profile} for profile in GLB_LOD_PROFILES ], "updatedAt": now_iso(), }, "updatedAt": now_iso(), } write_json_atomic(manifest_file, processing_manifest) update_asset_manifest_version(source_path, processing_manifest) source_size = source_path.stat().st_size records = [] commands = [] print(f"[{CONVERTER_NAME}] generating GLB LODs {source_path}", flush=True) try: for profile in GLB_LOD_PROFILES: artifact_path = glb_lod_artifact_path(source_path, profile) command, stdout, stderr = run_gltf_transform_lod(source_path, artifact_path, profile) commands.append({ "id": profile["id"], "command": command, "stdout": stdout, "stderr": stderr, }) artifact_size = artifact_path.stat().st_size records.append({ "id": profile["id"], "label": profile["label"], "artifactSrc": src_from_path(artifact_path), "type": "gltf", "ratio": profile.get("ratio"), "error": profile.get("error"), "size": artifact_size, "sha256": calculate_file_sha256(artifact_path), "sourceSize": source_size, "compressionRatio": round(artifact_size / source_size, 6) if source_size else None, "generatedAt": now_iso(), }) ready_manifest = { **base_manifest, "lods": records, "lodGeneration": { "status": "ready", "attempts": attempt, "converterName": CONVERTER_NAME, "converterVersion": CONVERTER_VERSION, "commands": commands, "sourceSize": source_size, "updatedAt": now_iso(), }, "message": "Model is ready. GLB LOD levels generated.", "updatedAt": now_iso(), } write_json_atomic(manifest_file, ready_manifest) update_asset_manifest_version(source_path, ready_manifest) print( f"[{CONVERTER_NAME}] GLB LOD ready {source_src}: " + ", ".join(f"{item['id']}={item['size']}" for item in records), flush=True, ) return True except Exception as exc: for profile in GLB_LOD_PROFILES: tmp_path = glb_lod_artifact_path(source_path, profile).with_suffix(".tmp.glb") if tmp_path.exists(): try: tmp_path.unlink() except Exception: pass failed_manifest = { **base_manifest, "message": "Model is ready. LOD generation failed.", "lodGeneration": { **(generation if isinstance(generation, dict) else {}), "status": "failed", "attempts": attempt, "converterName": CONVERTER_NAME, "converterVersion": CONVERTER_VERSION, "commands": commands, "error": str(exc), "updatedAt": now_iso(), }, "updatedAt": now_iso(), } write_json_atomic(manifest_file, failed_manifest) update_asset_manifest_version(source_path, failed_manifest) print(f"[{CONVERTER_NAME}] GLB LOD failed {source_path}: {exc}", file=sys.stderr, flush=True) return False def scan_once() -> int: if not UPLOADS_DIR.exists(): print(f"[{CONVERTER_NAME}] uploads dir does not exist: {UPLOADS_DIR}", flush=True) return 0 processed = 0 for source_path in sorted(UPLOADS_DIR.rglob("*")): if not source_path.is_file(): continue suffix = source_path.suffix.lower() did_process = False if suffix in STEP_EXTENSIONS: did_process = process_one(source_path) elif suffix in GLB_EXTENSIONS: did_process = convert_glb_to_xkt_one(source_path) did_process = optimize_glb_one(source_path) or did_process did_process = generate_glb_lods_one(source_path) or did_process if did_process: processed += 1 return processed def main() -> None: print(f"[{CONVERTER_NAME}] watching {UPLOADS_DIR}", flush=True) while True: scan_once() if PROCESS_ONCE: return time.sleep(POLL_INTERVAL_SECONDS) if __name__ == "__main__": main()