Hartmut
716451ff76
Migration 039: cad_files.mesh_attributes JSONB column. domains/products/tasks.py: extract_mesh_attributes Celery task using pythonOCC. still_render.py + turntable_render.py: _apply_mesh_attributes() sets auto-smooth based on curved_ratio and topology threshold from OCC analysis. render_blender.py: passes --mesh-attributes JSON arg to Blender subprocess. render_still_task: loads mesh_attributes from DB and passes to renderer. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
110 lines
3.6 KiB
Python
110 lines
3.6 KiB
Python
"""CAD processing tasks for the products domain."""
|
|
from __future__ import annotations
|
|
import asyncio
|
|
import logging
|
|
|
|
from celery import shared_task
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@shared_task(name="products.extract_mesh_attributes", queue="step_processing", bind=True)
|
|
def extract_mesh_attributes(self, cad_file_id: str) -> dict:
|
|
"""Extract mesh topology attributes from a STEP file using pythonOCC.
|
|
|
|
Stores result in cad_files.mesh_attributes JSONB.
|
|
Returns attribute dict.
|
|
"""
|
|
from app.database import AsyncSessionLocal
|
|
|
|
async def _run() -> dict:
|
|
from sqlalchemy import select, update as sql_update
|
|
from app.domains.products.models import CadFile
|
|
|
|
async with AsyncSessionLocal() as db:
|
|
result = await db.execute(select(CadFile).where(CadFile.id == cad_file_id))
|
|
cad_file = result.scalar_one_or_none()
|
|
if not cad_file:
|
|
logger.warning("CadFile %s not found", cad_file_id)
|
|
return {}
|
|
|
|
try:
|
|
attrs = _extract_from_step(cad_file.stored_path)
|
|
except Exception as exc:
|
|
logger.error("Mesh extraction failed for %s: %s", cad_file_id, exc)
|
|
attrs = {"error": str(exc)}
|
|
|
|
await db.execute(
|
|
sql_update(CadFile)
|
|
.where(CadFile.id == cad_file_id)
|
|
.values(mesh_attributes=attrs)
|
|
)
|
|
await db.commit()
|
|
return attrs
|
|
|
|
return asyncio.get_event_loop().run_until_complete(_run())
|
|
|
|
|
|
def _extract_from_step(step_path: str, threshold_deg: float = 30.0) -> dict:
|
|
"""Use pythonOCC to analyse STEP topology.
|
|
|
|
Returns face_groups, counts, and meta-information.
|
|
Gracefully returns {} when OCC is not installed.
|
|
"""
|
|
try:
|
|
from OCC.Core.STEPControl import STEPControl_Reader
|
|
from OCC.Core.TopAbs import TopAbs_EDGE, TopAbs_FACE
|
|
from OCC.Core.TopExp import TopExp_Explorer
|
|
from OCC.Core.BRepAdaptor import BRepAdaptor_Surface
|
|
from OCC.Core.GeomAbs import (
|
|
GeomAbs_Plane, GeomAbs_Cylinder, GeomAbs_Torus,
|
|
GeomAbs_Cone, GeomAbs_Sphere,
|
|
)
|
|
except ImportError:
|
|
logger.warning("pythonOCC not available — skipping mesh extraction")
|
|
return {}
|
|
|
|
reader = STEPControl_Reader()
|
|
if reader.ReadFile(step_path) != 1:
|
|
raise ValueError(f"STEP read failed: {step_path}")
|
|
reader.TransferRoots()
|
|
shape = reader.OneShape()
|
|
|
|
face_groups: dict[str, int] = {
|
|
"planar": 0, "cylindrical": 0, "toroidal": 0,
|
|
"conical": 0, "spherical": 0, "other": 0,
|
|
}
|
|
type_map = {
|
|
GeomAbs_Plane: "planar",
|
|
GeomAbs_Cylinder: "cylindrical",
|
|
GeomAbs_Torus: "toroidal",
|
|
GeomAbs_Cone: "conical",
|
|
GeomAbs_Sphere: "spherical",
|
|
}
|
|
|
|
face_explorer = TopExp_Explorer(shape, TopAbs_FACE)
|
|
face_count = 0
|
|
while face_explorer.More():
|
|
adaptor = BRepAdaptor_Surface(face_explorer.Current())
|
|
key = type_map.get(adaptor.GetType(), "other")
|
|
face_groups[key] += 1
|
|
face_count += 1
|
|
face_explorer.Next()
|
|
|
|
edge_explorer = TopExp_Explorer(shape, TopAbs_EDGE)
|
|
edge_count = 0
|
|
while edge_explorer.More():
|
|
edge_count += 1
|
|
edge_explorer.Next()
|
|
|
|
curved = face_groups["cylindrical"] + face_groups["toroidal"] + face_groups["spherical"]
|
|
|
|
return {
|
|
"face_count": face_count,
|
|
"edge_count": edge_count,
|
|
"face_groups": face_groups,
|
|
"curved_ratio": round(curved / max(face_count, 1), 3),
|
|
"sharp_angle_threshold_deg": threshold_deg,
|
|
"extraction_version": "1.0",
|
|
}
|