""" 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