364 lines
16 KiB
Python
364 lines
16 KiB
Python
"""
|
|
Main mesh processing workflow for CAE Mesh Generator
|
|
|
|
This module implements the complete mesh generation workflow including
|
|
geometry import, named selections, mesh controls, generation, and quality check.
|
|
"""
|
|
import logging
|
|
from typing import Dict, Any, Optional, Callable
|
|
from datetime import datetime
|
|
from enum import Enum
|
|
|
|
from backend.pymechanical.session_manager import ANSYSSessionManager
|
|
from backend.utils.state_manager import state_manager
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class ProcessingStep(Enum):
|
|
"""Processing step enumeration"""
|
|
INITIALIZING = "initializing"
|
|
STARTING_SESSION = "starting_session"
|
|
IMPORTING_GEOMETRY = "importing_geometry"
|
|
VALIDATING_GEOMETRY = "validating_geometry"
|
|
CREATING_NAMED_SELECTIONS = "creating_named_selections"
|
|
APPLYING_MESH_CONTROLS = "applying_mesh_controls"
|
|
GENERATING_MESH = "generating_mesh"
|
|
CHECKING_QUALITY = "checking_quality"
|
|
FINALIZING = "finalizing"
|
|
COMPLETED = "completed"
|
|
FAILED = "failed"
|
|
|
|
class MeshProcessingResult:
|
|
"""Result container for mesh processing workflow"""
|
|
def __init__(self):
|
|
self.success = False
|
|
self.current_step = ProcessingStep.INITIALIZING
|
|
self.progress_percentage = 0.0
|
|
self.started_at = None
|
|
self.completed_at = None
|
|
self.total_time = 0.0
|
|
self.error_message = None
|
|
self.warnings = []
|
|
|
|
# Step results
|
|
self.geometry_result = None
|
|
self.named_selections_result = None
|
|
self.mesh_controls_result = None
|
|
self.mesh_generation_result = None
|
|
self.quality_check_result = None
|
|
|
|
# Final mesh info
|
|
self.element_count = 0
|
|
self.node_count = 0
|
|
self.quality_score = 0.0
|
|
self.quality_status = "UNKNOWN"
|
|
self.mesh_image_path = None # Path to the exported mesh visualization image
|
|
|
|
def process_blade_mesh(
|
|
file_path: str,
|
|
progress_callback: Optional[Callable[[float, str, ProcessingStep], None]] = None
|
|
) -> MeshProcessingResult:
|
|
"""
|
|
Main mesh processing function for blade geometry
|
|
|
|
This function implements the complete workflow from geometry import to
|
|
quality checking, integrating all mesh generation components.
|
|
|
|
Args:
|
|
file_path: Path to the STEP file
|
|
progress_callback: Optional callback for progress updates (progress%, message, step)
|
|
|
|
Returns:
|
|
MeshProcessingResult with complete processing results
|
|
"""
|
|
result = MeshProcessingResult()
|
|
result.started_at = datetime.now()
|
|
session_manager = None
|
|
|
|
def update_progress(percentage: float, message: str, step: ProcessingStep = None):
|
|
"""Update progress and call callback if provided"""
|
|
result.progress_percentage = percentage
|
|
if step:
|
|
result.current_step = step
|
|
|
|
if progress_callback:
|
|
try:
|
|
progress_callback(percentage, message, result.current_step)
|
|
except Exception as e:
|
|
logger.warning(f"Progress callback error: {str(e)}")
|
|
|
|
logger.info(f"Progress: {percentage:.1f}% - {message} (Step: {result.current_step.value})")
|
|
|
|
try:
|
|
# Step 1: Initialize session
|
|
update_progress(5.0, "Initializing ANSYS session...", ProcessingStep.STARTING_SESSION)
|
|
|
|
session_manager = ANSYSSessionManager()
|
|
|
|
if not session_manager.start_session():
|
|
raise Exception("Failed to start ANSYS session")
|
|
|
|
update_progress(10.0, "ANSYS session started successfully")
|
|
|
|
# Step 2: Import geometry
|
|
update_progress(15.0, "Importing geometry from STEP file...", ProcessingStep.IMPORTING_GEOMETRY)
|
|
|
|
if not session_manager.import_geometry(file_path):
|
|
raise Exception("Failed to import geometry")
|
|
|
|
update_progress(25.0, "Geometry imported successfully")
|
|
|
|
# Step 3: Validate geometry
|
|
update_progress(30.0, "Validating imported geometry...", ProcessingStep.VALIDATING_GEOMETRY)
|
|
|
|
geometry_validation = session_manager.validate_geometry()
|
|
result.geometry_result = geometry_validation
|
|
|
|
if not geometry_validation.get('valid', False):
|
|
raise Exception(f"Geometry validation failed: {geometry_validation.get('error', 'Unknown error')}")
|
|
|
|
update_progress(35.0, f"Geometry validated: {geometry_validation.get('surface_count', 0)} surfaces")
|
|
|
|
# Step 4: Create named selections
|
|
update_progress(40.0, "Creating named selections for blade features...", ProcessingStep.CREATING_NAMED_SELECTIONS)
|
|
|
|
named_selections_result = session_manager.create_named_selections()
|
|
result.named_selections_result = named_selections_result
|
|
|
|
if not named_selections_result.get('success', False):
|
|
result.warnings.append(f"Named selections creation had issues: {named_selections_result.get('error', 'Unknown error')}")
|
|
logger.warning("Named selections creation failed, continuing without them")
|
|
else:
|
|
selections_count = named_selections_result.get('total_selections', 0)
|
|
update_progress(50.0, f"Named selections created: {selections_count} selections")
|
|
|
|
# Step 5: Apply mesh controls
|
|
update_progress(55.0, "Applying mesh controls and sizing...", ProcessingStep.APPLYING_MESH_CONTROLS)
|
|
|
|
# Get named selections for mesh controls
|
|
named_selections = []
|
|
if named_selections_result.get('success'):
|
|
named_selections = named_selections_result.get('selections_created', [])
|
|
|
|
mesh_controls_result = session_manager.apply_mesh_controls(named_selections)
|
|
result.mesh_controls_result = mesh_controls_result
|
|
|
|
if not mesh_controls_result.get('success', False):
|
|
result.warnings.append(f"Mesh controls application had issues: {mesh_controls_result.get('error', 'Unknown error')}")
|
|
logger.warning("Mesh controls application failed, using defaults")
|
|
else:
|
|
controls_count = len(mesh_controls_result.get('controls_applied', []))
|
|
update_progress(65.0, f"Mesh controls applied: {controls_count} controls")
|
|
|
|
# Step 6: Generate mesh
|
|
update_progress(70.0, "Generating mesh...", ProcessingStep.GENERATING_MESH)
|
|
|
|
# Set up progress callback for mesh generation
|
|
def mesh_progress_callback(mesh_progress: float, mesh_message: str):
|
|
# Map mesh generation progress (0-100) to overall progress (70-85)
|
|
overall_progress = 70.0 + (mesh_progress / 100.0) * 15.0
|
|
update_progress(overall_progress, f"Mesh generation: {mesh_message}")
|
|
|
|
mesh_generation_result = session_manager.generate_mesh(progress_callback=mesh_progress_callback)
|
|
result.mesh_generation_result = mesh_generation_result
|
|
|
|
if not mesh_generation_result.get('success', False):
|
|
raise Exception(f"Mesh generation failed: {mesh_generation_result.get('error_message', 'Unknown error')}")
|
|
|
|
result.element_count = mesh_generation_result.get('element_count', 0)
|
|
result.node_count = mesh_generation_result.get('node_count', 0)
|
|
|
|
update_progress(85.0, f"Mesh generated: {result.element_count} elements, {result.node_count} nodes")
|
|
|
|
# Step 7: Check mesh quality
|
|
update_progress(90.0, "Checking mesh quality...", ProcessingStep.CHECKING_QUALITY)
|
|
|
|
quality_check_result = session_manager.check_mesh_quality()
|
|
result.quality_check_result = quality_check_result
|
|
|
|
if quality_check_result.get('success', False):
|
|
result.quality_score = quality_check_result.get('quality_score', 0.0)
|
|
result.quality_status = quality_check_result.get('overall_status', 'UNKNOWN')
|
|
|
|
quality_issues = len(quality_check_result.get('critical_issues', []))
|
|
if quality_issues > 0:
|
|
result.warnings.append(f"Mesh quality check found {quality_issues} critical issues")
|
|
|
|
update_progress(95.0, f"Quality check completed: {result.quality_status} (Score: {result.quality_score:.1f})")
|
|
else:
|
|
result.warnings.append(f"Quality check failed: {quality_check_result.get('error', 'Unknown error')}")
|
|
result.quality_status = "CHECK_FAILED"
|
|
update_progress(95.0, "Quality check failed, but mesh generation completed")
|
|
|
|
# Step 8: Finalize
|
|
update_progress(98.0, "Finalizing results...", ProcessingStep.FINALIZING)
|
|
|
|
# Calculate total processing time
|
|
result.completed_at = datetime.now()
|
|
result.total_time = (result.completed_at - result.started_at).total_seconds()
|
|
|
|
# Step 6: Export visualization image
|
|
update_progress(95.0, "Exporting mesh visualization...", ProcessingStep.FINALIZING)
|
|
logger.info("Step 6: Exporting mesh visualization...")
|
|
|
|
try:
|
|
from backend.utils.visualization_exporter import VisualizationExporter, VisualizationSettings
|
|
import os
|
|
|
|
# Use absolute path to ensure correct location
|
|
base_dir = os.path.dirname(os.path.dirname(os.path.dirname(__file__))) # Go up to project root
|
|
viz_output_dir = os.path.join(base_dir, "frontend", "static", "visualizations")
|
|
|
|
# Create visualization exporter
|
|
viz_exporter = VisualizationExporter(
|
|
mechanical_session=session_manager.session if session_manager else None,
|
|
output_dir=viz_output_dir
|
|
)
|
|
|
|
# Export mesh image
|
|
viz_settings = VisualizationSettings(
|
|
width=1280,
|
|
height=720,
|
|
image_format="PNG",
|
|
camera_view="isometric",
|
|
show_edges=True
|
|
)
|
|
|
|
viz_result = viz_exporter.export_mesh_image(
|
|
filename="current_mesh_preview.png",
|
|
settings=viz_settings
|
|
)
|
|
|
|
if viz_result.success:
|
|
result.mesh_image_path = viz_result.image_path
|
|
logger.info(f"✓ Mesh visualization exported: {viz_result.image_path}")
|
|
else:
|
|
logger.warning(f"Mesh visualization export failed: {viz_result.error_message}")
|
|
result.warnings.append(f"Visualization export failed: {viz_result.error_message}")
|
|
|
|
except Exception as viz_error:
|
|
logger.warning(f"Visualization export error: {str(viz_error)}")
|
|
result.warnings.append(f"Visualization export error: {str(viz_error)}")
|
|
|
|
# Mark as successful
|
|
result.success = True
|
|
result.current_step = ProcessingStep.COMPLETED
|
|
|
|
update_progress(100.0, f"Mesh processing completed successfully in {result.total_time:.1f} seconds", ProcessingStep.COMPLETED)
|
|
|
|
logger.info(f"✓ Blade mesh processing completed successfully:")
|
|
logger.info(f" - Elements: {result.element_count}")
|
|
logger.info(f" - Nodes: {result.node_count}")
|
|
logger.info(f" - Quality Score: {result.quality_score:.1f}")
|
|
logger.info(f" - Quality Status: {result.quality_status}")
|
|
logger.info(f" - Processing Time: {result.total_time:.1f} seconds")
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
# Handle processing failure
|
|
logger.error(f"Mesh processing failed: {str(e)}")
|
|
|
|
result.success = False
|
|
result.current_step = ProcessingStep.FAILED
|
|
result.error_message = str(e)
|
|
result.completed_at = datetime.now()
|
|
|
|
if result.started_at:
|
|
result.total_time = (result.completed_at - result.started_at).total_seconds()
|
|
|
|
update_progress(0.0, f"Processing failed: {str(e)}", ProcessingStep.FAILED)
|
|
|
|
return result
|
|
|
|
finally:
|
|
# Clean up session
|
|
if session_manager:
|
|
try:
|
|
session_manager.close_session()
|
|
session_manager.cleanup_temp_files()
|
|
logger.info("✓ Session cleanup completed")
|
|
except Exception as cleanup_error:
|
|
logger.warning(f"Session cleanup error: {str(cleanup_error)}")
|
|
|
|
def process_blade_mesh_with_state_updates(
|
|
file_path: str
|
|
) -> MeshProcessingResult:
|
|
"""
|
|
Process blade mesh with automatic state manager updates
|
|
|
|
This function wraps the main processing function and automatically
|
|
updates the global state manager with progress information.
|
|
|
|
Args:
|
|
file_path: Path to the STEP file
|
|
|
|
Returns:
|
|
MeshProcessingResult with complete processing results
|
|
"""
|
|
def progress_callback(percentage: float, message: str, step: ProcessingStep):
|
|
"""Update state manager with progress"""
|
|
try:
|
|
# Update processing status in state manager
|
|
processing_status = state_manager.get_processing_status()
|
|
processing_status.status = step.value
|
|
processing_status.progress_percentage = percentage
|
|
processing_status.current_operation = message
|
|
processing_status.last_updated = datetime.now()
|
|
|
|
# Update state manager
|
|
state_manager.update_processing_status(processing_status)
|
|
|
|
except Exception as e:
|
|
logger.warning(f"State update error: {str(e)}")
|
|
|
|
logger.info(f"Starting blade mesh processing with state updates: {file_path}")
|
|
|
|
# Perform mesh processing with state updates
|
|
result = process_blade_mesh(
|
|
file_path=file_path,
|
|
progress_callback=progress_callback
|
|
)
|
|
|
|
# Update final state
|
|
try:
|
|
if result.success:
|
|
# Create mesh result for state manager
|
|
from backend.models.data_models import MeshResult
|
|
|
|
mesh_result = MeshResult(
|
|
element_count=result.element_count,
|
|
node_count=result.node_count,
|
|
generation_time=result.total_time,
|
|
quality_score=result.quality_score,
|
|
quality_status=result.quality_status,
|
|
mesh_file_path=None, # Could be added later for file output
|
|
created_at=result.completed_at,
|
|
# Add exported files information
|
|
exported_files=getattr(result.mesh_generation_result, 'exported_files', {}),
|
|
export_success=getattr(result.mesh_generation_result, 'export_success', False),
|
|
export_errors=getattr(result.mesh_generation_result, 'export_errors', [])
|
|
)
|
|
|
|
state_manager.set_mesh_result(mesh_result)
|
|
|
|
# Update processing status to completed
|
|
processing_status = state_manager.get_processing_status()
|
|
processing_status.status = "completed"
|
|
processing_status.progress_percentage = 100.0
|
|
processing_status.current_operation = "Mesh processing completed successfully"
|
|
processing_status.completed_at = result.completed_at
|
|
state_manager.update_processing_status(processing_status)
|
|
|
|
else:
|
|
# Update processing status to failed
|
|
processing_status = state_manager.get_processing_status()
|
|
processing_status.status = "failed"
|
|
processing_status.error_message = result.error_message
|
|
processing_status.completed_at = result.completed_at
|
|
state_manager.update_processing_status(processing_status)
|
|
|
|
except Exception as state_error:
|
|
logger.error(f"State manager update failed: {str(state_error)}")
|
|
|
|
return result |