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.7" 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_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") 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)) ) 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"} 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 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: result = subprocess.run( command, cwd=str(glb_path.parent), env=env, stdout=log_file, stderr=subprocess.STDOUT, text=True, check=False, ) 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 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 ready_manifest = { **base_manifest, "artifactSha256": calculate_file_sha256(artifact_path), "artifactSrc": src_from_path(artifact_path), "artifactType": "gltf", "fallbackArtifactSrc": source_src, "fallbackArtifactType": "gltf", "message": "Model is ready. GLB geometry optimized with Draco.", "targetFormat": "gltf", "dracoOptimization": { "status": "ready", "attempts": attempt, "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 {ready_manifest['artifactSrc']} " 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 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 = optimize_glb_one(source_path) 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()