Hartmut/HartOMat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(P11+P12): codebase hygiene — CLAUDE.md rewrite, type safety, dead code removal

Browse Source 
- Rewrite CLAUDE.md to match current 8-service architecture (was 11, 5 deleted)
- Remove all as-any casts in OrderDetail.tsx (9 casts → 0)
- Add cad_parsed_objects/cad_part_materials to OrderItem interface
- Rename require_admin → require_global_admin across 6 router files (22 calls)
- Remove EXPORT_GLB_PRODUCTION enum + generate_gltf_production_task (dead code)
- Remove worker-thumbnail from ALLOWED_SERVICES, replace Flamenco link
- Delete obsolete PLAN.md (1455 lines) and PLAN_REFACTOR.md (1174 lines)
- Fix digit-only USD prim names with p_ prefix

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Hartmut Nörenberg
2026-03-13 07:22:04 +01:00
parent 3dcfa7c0bd
commit 577dd1ca7e
21 changed files with 303 additions and 3229 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/domains/pipeline/tasks/export_glb.py
+2 -279
View File
@@ -1,8 +1,8 @@
"""GLB/GLTF export tasks.
"""GLB/GLTF and USD export tasks.
Covers:
- generate_gltf_geometry_task — OCC STEP → geometry GLB (fast preview)
- generate_gltf_production_task — OCC STEP → production GLB (Blender PBR materials)
- generate_usd_master_task — OCC STEP → USD canonical scene (pxr authoring)
"""
import logging
@@ -251,283 +251,6 @@ def generate_gltf_geometry_task(self, cad_file_id: str):
_r.delete(_lock_key)
@celery_app.task(
bind=True,
name="app.tasks.step_tasks.generate_gltf_production_task",
queue="asset_pipeline",
max_retries=2,
)
def generate_gltf_production_task(self, cad_file_id: str, product_id: str | None = None) -> dict:
"""Generate a production GLB (Blender + PBR materials) from a geometry GLB via export_gltf.py.
1. Ensures a gltf_geometry MediaAsset exists (runs OCC export inline if not).
2. Resolves SCHAEFFLER material map for the CadFile's product.
3. Runs Blender headless with export_gltf.py → production GLB.
4. Stores result as gltf_production MediaAsset.
"""
import json as _json
import os as _os
import subprocess as _subprocess
import sys as _sys
import uuid as _uuid
from pathlib import Path as _Path
from sqlalchemy import create_engine as _ce, delete as _del, select as _sel, update as _upd
from sqlalchemy.orm import Session as _Session
from app.config import settings as app_settings
from app.domains.media.models import MediaAsset, MediaAssetType
from app.services.render_blender import find_blender, is_blender_available
pl = PipelineLogger(task_id=self.request.id)
pl.step_start("export_glb_production", {"cad_file_id": cad_file_id})
log_task_event(self.request.id, f"generate_gltf_production_task started for cad {cad_file_id}", "info")
# Resolve and log tenant context at task start (required for RLS)
from app.core.tenant_context import resolve_tenant_id_for_cad, set_tenant_context_sync
_tenant_id = resolve_tenant_id_for_cad(cad_file_id)
_sync_url = app_settings.database_url.replace("+asyncpg", "")
_eng = _ce(_sync_url)
# --- 1. Resolve STEP file path and system settings ---
from app.models.cad_file import CadFile as _CF
from app.models.system_setting import SystemSetting
with _Session(_eng) as _sess:
set_tenant_context_sync(_sess, _tenant_id)
_cad = _sess.execute(
_sel(_CF).where(_CF.id == _uuid.UUID(cad_file_id))
).scalar_one_or_none()
step_path_str = _cad.stored_path if _cad else None
cad_mesh_attributes: dict = (_cad.mesh_attributes or {}) if _cad else {}
settings_rows = _sess.execute(_sel(SystemSetting)).scalars().all()
sys_settings = {s.key: s.value for s in settings_rows}
if not step_path_str:
raise RuntimeError(f"CadFile {cad_file_id} not found in DB")
step_path = _Path(step_path_str)
if not step_path.exists():
raise RuntimeError(f"STEP file not found: {step_path}")
smooth_angle = float(sys_settings.get("blender_smooth_angle", "30"))
prod_linear = float(sys_settings.get("render_linear_deflection", "0.03"))
prod_angular = float(sys_settings.get("render_angular_deflection", "0.05"))
tessellation_engine = sys_settings.get("tessellation_engine", "occ")
scripts_dir = _Path(_os.environ.get("RENDER_SCRIPTS_DIR", "/render-scripts"))
occ_script = scripts_dir / "export_step_to_gltf.py"
if not occ_script.exists():
raise RuntimeError(f"export_step_to_gltf.py not found at {occ_script}")
prod_geom_glb = step_path.parent / f"{step_path.stem}_production_geom.glb"
python_bin = _sys.executable
sharp_threshold = float(sys_settings.get("sharp_edge_threshold", "20.0"))
# --- Geometry GLB selection strategy ---
# When GMSH is enabled, the geometry GLB (_geometry.glb) is already a conforming
# mesh with correct seam topology — GMSH quality comes from the algorithm, not density.
# Re-tessellating at finer production settings only wastes time and RAM on large assemblies.
# → For GMSH: reuse the existing _geometry.glb if it is newer than the STEP file.
# → For OCC: generate a separate _production_geom.glb at finer settings (density matters).
step_mtime = step_path.stat().st_mtime if step_path.exists() else 0
preview_glb = step_path.parent / f"{step_path.stem}_geometry.glb"
preview_glb_valid = (
preview_glb.exists()
and preview_glb.stat().st_size > 0
and preview_glb.stat().st_mtime >= step_mtime
)
prod_geom_cache_valid = (
prod_geom_glb.exists()
and prod_geom_glb.stat().st_size > 0
and prod_geom_glb.stat().st_mtime >= step_mtime
)
if tessellation_engine == "gmsh" and preview_glb_valid:
# Fast path: reuse geometry GLB — GMSH topology is already correct at preview quality
geom_glb_path = preview_glb
log_task_event(
self.request.id,
f"GMSH: reusing geometry GLB as Blender input ({preview_glb.stat().st_size // 1024}KB, "
f"no re-tessellation needed)",
"info",
)
elif prod_geom_cache_valid:
# Cache hit: production_geom.glb exists and is up-to-date
geom_glb_path = prod_geom_glb
log_task_event(
self.request.id,
f"Cache hit: reusing production geometry GLB ({prod_geom_glb.stat().st_size // 1024}KB)",
"info",
)
else:
# No usable cache: run tessellation from STEP.
# When GMSH is selected, force preview-quality settings (0.1mm / 0.1rad) even here.
# Fine production settings (e.g. 0.03mm) combined with GMSH OOM-kill on large assemblies
# because CharacteristicLengthMax becomes too small. GMSH quality is algorithmic
# (conforming seams) not density-based — a denser GMSH mesh adds no UV-unwrap benefit.
if tessellation_engine == "gmsh":
eff_linear = float(sys_settings.get("scene_linear_deflection", "0.1"))
eff_angular = float(sys_settings.get("scene_angular_deflection", "0.1"))
else:
eff_linear = prod_linear
eff_angular = prod_angular
occ_cmd = [
python_bin, str(occ_script),
"--step_path", str(step_path),
"--output_path", str(prod_geom_glb),
"--linear_deflection", str(eff_linear),
"--angular_deflection", str(eff_angular),
"--sharp_threshold", str(sharp_threshold),
"--tessellation_engine", tessellation_engine,
]
log_task_event(
self.request.id,
f"Tessellating STEP for production ({tessellation_engine}, "
f"linear={eff_linear}mm, angular={eff_angular}rad)",
"info",
)
try:
occ_result = _subprocess.run(occ_cmd, capture_output=True, text=True, timeout=600)
for line in occ_result.stdout.splitlines():
logger.info("[occ-prod] %s", line)
if occ_result.returncode != 0 or not prod_geom_glb.exists() or prod_geom_glb.stat().st_size == 0:
raise RuntimeError(
f"OCC export failed (exit {occ_result.returncode}): {occ_result.stderr[-500:]}"
)
except Exception as exc:
log_task_event(self.request.id, f"OCC re-export failed: {exc}", "error")
pl.step_error("export_glb_production", f"OCC re-export failed: {exc}", exc)
raise self.retry(exc=exc, countdown=30)
geom_glb_path = prod_geom_glb
# --- 2. Resolve material map from Product.cad_part_materials (SCHAEFFLER library names) ---
# cad_part_materials lives on Product (list[dict]), NOT on CadFile.
# We look up the Product that owns this CadFile (prefer product_id arg if given).
from app.services.material_service import resolve_material_map
from app.domains.products.models import Product as _Product
with _Session(_eng) as _sess:
set_tenant_context_sync(_sess, _tenant_id)
_prod_query = _sel(_Product).where(_Product.cad_file_id == _uuid.UUID(cad_file_id))
if product_id:
_prod_query = _prod_query.where(_Product.id == _uuid.UUID(product_id))
_product = _sess.execute(_prod_query).scalars().first()
raw_materials: list[dict] = _product.cad_part_materials if _product else []
# Convert list[{"part_name": X, "material": Y}] → dict[str, str] for resolve_material_map
raw_mat_map: dict[str, str] = {
m["part_name"]: m["material"]
for m in raw_materials
if m.get("part_name") and m.get("material")
}
mat_map = resolve_material_map(raw_mat_map)
logger.info(
"generate_gltf_production_task: resolved %d material(s) for cad %s (product: %s)",
len(mat_map), cad_file_id, _product.id if _product else "none",
)
# --- 3. Run Blender: apply materials + smooth shading + export production GLB ---
# Use get_material_library_path() which checks active AssetLibrary first,
# then falls back to the legacy material_library_path system setting.
from app.services.template_service import get_material_library_path
asset_library_blend = get_material_library_path() or ""
_eng.dispose()
output_path = step_path.parent / f"{step_path.stem}_production.glb"
export_script = scripts_dir / "export_gltf.py"
if not is_blender_available():
raise RuntimeError("Blender is not available — cannot generate production GLB")
if not export_script.exists():
raise RuntimeError(f"export_gltf.py not found at {export_script}")
blender_bin = find_blender()
cmd = [
blender_bin, "--background",
"--python", str(export_script),
"--",
"--glb_path", str(geom_glb_path),
"--output_path", str(output_path),
"--material_map", _json.dumps(mat_map),
"--smooth_angle", str(smooth_angle),
"--mesh_attributes", _json.dumps(cad_mesh_attributes),
]
if asset_library_blend:
cmd += ["--asset_library_blend", asset_library_blend]
log_task_event(
self.request.id,
f"Running Blender export_gltf.py — {len(mat_map)} material(s), smooth={smooth_angle}°",
"info",
)
try:
result = _subprocess.run(cmd, capture_output=True, text=True, timeout=300)
for line in result.stdout.splitlines():
logger.info("[export-gltf] %s", line)
if result.returncode != 0:
raise RuntimeError(
f"export_gltf.py exited {result.returncode}:\n{result.stderr[-500:]}"
)
except Exception as exc:
log_task_event(self.request.id, f"Blender production GLB failed: {exc}", "error")
pl.step_error("export_glb_production", f"Blender production GLB failed: {exc}", exc)
logger.error("generate_gltf_production_task Blender failed for cad %s: %s", cad_file_id, exc)
raise self.retry(exc=exc, countdown=30)
# Note: _production_geom.glb is intentionally kept on disk as a tessellation cache.
# It is reused on subsequent runs when the STEP file hasn't changed.
log_task_event(self.request.id, f"Production GLB exported: {output_path.name}", "done")
# --- 4. Store MediaAsset (upsert: update existing record to keep stable ID/URL) ---
# Updating in-place (not DELETE+INSERT) preserves the existing asset UUID so that
# any frontend page holding a stale download_url continues to resolve correctly.
_eng2 = _ce(_sync_url)
with _Session(_eng2) as _sess:
set_tenant_context_sync(_sess, _tenant_id)
_key = str(output_path)
_prefix = str(app_settings.upload_dir).rstrip("/") + "/"
if _key.startswith(_prefix):
_key = _key[len(_prefix):]
_file_size = output_path.stat().st_size if output_path.exists() else None
existing = _sess.execute(
_sel(MediaAsset).where(
MediaAsset.cad_file_id == _uuid.UUID(cad_file_id),
MediaAsset.asset_type == MediaAssetType.gltf_production,
)
).scalars().first()
if existing:
existing.storage_key = _key
existing.mime_type = "model/gltf-binary"
existing.file_size_bytes = _file_size
if product_id:
existing.product_id = _uuid.UUID(product_id)
_sess.commit()
asset_id = str(existing.id)
else:
asset = MediaAsset(
cad_file_id=_uuid.UUID(cad_file_id),
product_id=_uuid.UUID(product_id) if product_id else None,
asset_type=MediaAssetType.gltf_production,
storage_key=_key,
mime_type="model/gltf-binary",
file_size_bytes=_file_size,
)
_sess.add(asset)
_sess.commit()
asset_id = str(asset.id)
_eng2.dispose()
pl.step_done("export_glb_production", result={"glb_path": str(output_path), "asset_id": asset_id})
logger.info("generate_gltf_production_task: MediaAsset %s created for cad %s", asset_id, cad_file_id)
return {"glb_path": str(output_path), "asset_id": asset_id}
@celery_app.task(
bind=True,
name="app.tasks.step_tasks.generate_usd_master_task",