AnsysLink/backend/pymechanical/session_manager.py
2025-08-11 13:58:59 +08:00

801 lines
30 KiB
Python

"""
ANSYS Mechanical session manager for CAE Mesh Generator
"""
import os
import time
import logging
from pathlib import Path
from typing import Optional, Dict, Any, List, Callable
from datetime import datetime
from config import ANSYS_CONFIG, TEMP_DIR
from backend.pymechanical.named_selection_manager import NamedSelectionManager
from backend.pymechanical.mesh_controller import MeshController
from backend.pymechanical.mesh_generator import MeshGenerator
from backend.pymechanical.mesh_quality_checker import MeshQualityChecker
from backend.utils.resource_manager import resource_manager, file_manager
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ANSYSSessionManager:
"""
ANSYS Mechanical session manager
This class manages ANSYS Mechanical sessions for mesh generation
using real PyMechanical integration.
"""
def __init__(self):
"""
Initialize session manager for real ANSYS Mechanical integration
"""
self.session = None
self.mechanical = None
self.is_session_active = False
self.current_geometry = None
self.session_start_time = None
self.temp_files = []
self.named_selection_manager = None
self.mesh_controller = None
self.mesh_generator = None
self.mesh_quality_checker = None
# Create temp directory if it doesn't exist
os.makedirs(TEMP_DIR, exist_ok=True)
logger.info("ANSYS Session Manager initialized")
def start_session(self, batch_mode: bool = True) -> bool:
"""
Start ANSYS Mechanical session
Args:
batch_mode: Whether to start in batch mode
Returns:
True if session started successfully, False otherwise
"""
try:
logger.info("Starting ANSYS Mechanical session...")
self.session_start_time = datetime.now()
# Real PyMechanical integration
try:
import ansys.mechanical.core as pymechanical
import os
logger.info("Launching ANSYS Mechanical...")
# Set up ANSYS environment
ansys_root = ANSYS_CONFIG.get('ansys_root', r'C:\Program Files\ANSYS Inc\v241')
os.environ['AWP_ROOT241'] = ansys_root
os.environ['ANSYS_ROOT'] = ansys_root
# Configure launch parameters
launch_options = {
'batch': batch_mode,
'cleanup_on_exit': True,
}
# Add version-specific configuration if available
if 'ansys_version' in ANSYS_CONFIG:
launch_options['version'] = ANSYS_CONFIG['ansys_version']
# Start Mechanical session
self.mechanical = pymechanical.launch_mechanical(**launch_options)
# Verify session is working
if self.mechanical:
# Test basic functionality
try:
# Check if session is alive
if hasattr(self.mechanical, 'is_alive') and self.mechanical.is_alive:
logger.info(f"ANSYS Mechanical session established and alive")
else:
logger.info(f"ANSYS Mechanical session established")
# Try to get product info
try:
product_info = self.mechanical.get_product_info()
logger.info(f"ANSYS Product info: {product_info}")
except Exception as e:
logger.debug(f"Could not get product info: {str(e)}")
self.session = self.mechanical
self.is_session_active = True
# Initialize named selection manager
self.named_selection_manager = NamedSelectionManager(self.mechanical)
# Initialize mesh controller
self.mesh_controller = MeshController(self.mechanical)
# Initialize mesh generator
self.mesh_generator = MeshGenerator(self.mechanical)
# Initialize mesh quality checker
self.mesh_quality_checker = MeshQualityChecker(self.mechanical)
logger.info("✓ Real ANSYS Mechanical session started successfully")
return True
except Exception as session_error:
logger.error(f"Failed to verify Mechanical session: {str(session_error)}")
# Try to close the session if it was partially started
try:
self.mechanical.exit()
except:
pass
return False
else:
logger.error("Failed to launch ANSYS Mechanical")
return False
except ImportError as e:
logger.error(f"PyMechanical not available: {str(e)}")
logger.error("Please ensure ANSYS Mechanical and PyMechanical are properly installed")
return False
except Exception as e:
logger.error(f"Failed to start ANSYS Mechanical session: {str(e)}")
logger.error("Please check ANSYS installation and license availability")
return False
except Exception as e:
logger.error(f"Session startup error: {str(e)}")
return False
def import_geometry(self, file_path: str) -> bool:
"""
Import geometry from CAD file
Args:
file_path: Path to the CAD file
Returns:
True if import successful, False otherwise
"""
try:
if not self.is_session_active:
logger.error("No active ANSYS session")
return False
if not os.path.exists(file_path):
logger.error(f"Geometry file not found: {file_path}")
return False
logger.info(f"Importing geometry from: {file_path}")
# Real PyMechanical geometry import
try:
logger.info("Importing geometry using PyMechanical...")
# Method 1: Try direct script execution
try:
# Create a script to import geometry using correct PyMechanical API
import_script = f'''
# Import geometry using PyMechanical API
geometry_import = Model.GeometryImportGroup.AddGeometryImport()
geometry_import.Import(r"{file_path}")
body_count = len(Model.Geometry.GetChildren(Ansys.Mechanical.DataModel.Enums.DataModelObjectCategory.Body, True))
print("Imported geometry with " + str(body_count) + " bodies")
'''
# Execute the script
result = self.mechanical.run_python_script(import_script)
logger.info(f"Import script result: {result}")
self.current_geometry = {
"file_path": file_path,
"imported_at": datetime.now(),
"import_method": "direct_script",
"import_result": result if result else "Import executed"
}
logger.info("✓ Real geometry import completed successfully")
return True
except Exception as script_error:
logger.warning(f"Direct script method failed: {str(script_error)}")
# Method 2: Try file upload approach
logger.info("Trying file upload approach...")
# Upload the file to the server first
uploaded_file = self.mechanical.upload(file_path)
logger.info(f"File uploaded to server: {uploaded_file}")
# Create import script using uploaded file with full path
project_dir = self.mechanical.run_python_script("ExtAPI.DataModel.Project.ProjectDirectory")
full_uploaded_path = f"{project_dir}/{uploaded_file}".replace("\\", "/")
upload_script = f'''
# Import uploaded geometry using PyMechanical API
geometry_import = Model.GeometryImportGroup.AddGeometryImport()
geometry_import.Import(r"{full_uploaded_path}")
body_count = len(Model.Geometry.GetChildren(Ansys.Mechanical.DataModel.Enums.DataModelObjectCategory.Body, True))
print("Imported uploaded geometry with " + str(body_count) + " bodies")
'''
upload_result = self.mechanical.run_python_script(upload_script)
logger.info(f"Upload import result: {upload_result}")
self.current_geometry = {
"file_path": file_path,
"uploaded_file": uploaded_file,
"imported_at": datetime.now(),
"import_method": "upload_script",
"import_result": upload_result if upload_result else "Upload import executed"
}
logger.info("✓ Real geometry import completed via upload method")
return True
except Exception as e:
logger.error(f"PyMechanical geometry import error: {str(e)}")
logger.error("This could be due to:")
logger.error("- Invalid or corrupted STEP file")
logger.error("- Unsupported geometry format")
logger.error("- ANSYS Mechanical session issues")
logger.error("- PyMechanical API version mismatch")
return False
except Exception as e:
logger.error(f"Geometry import error: {str(e)}")
return False
def validate_geometry(self) -> Dict[str, Any]:
"""
Validate imported geometry
Returns:
Dictionary with validation results
"""
try:
if not self.current_geometry:
return {
"valid": False,
"error": "No geometry imported"
}
logger.info("Validating imported geometry...")
# Real geometry validation using PyMechanical
try:
# Get geometry information using PyMechanical script
validation_script = '''
# Get geometry validation information
body_count = len(Model.Geometry.GetChildren(Ansys.Mechanical.DataModel.Enums.DataModelObjectCategory.Body, True))
surface_count = body_count # Simplified - assume each body has surfaces
volume_count = body_count # Simplified - assume each body has volume
print("Bodies: " + str(body_count) + ", Surfaces: " + str(surface_count) + ", Volumes: " + str(volume_count))
'''
validation_result_str = self.mechanical.run_python_script(validation_script)
logger.info(f"Validation script result: {validation_result_str}")
# Parse the result or use default values
body_count = 1 # Default value
surface_count = 1
volume_count = 1
# Try to extract numbers from the result string
if validation_result_str:
import re
numbers = re.findall(r'\d+', str(validation_result_str))
if len(numbers) >= 3:
body_count = int(numbers[0])
surface_count = int(numbers[1])
volume_count = int(numbers[2])
validation_result = {
"valid": True,
"file_path": self.current_geometry["file_path"],
"body_count": body_count,
"surface_count": surface_count,
"volume_count": volume_count,
"imported_at": self.current_geometry["imported_at"].isoformat(),
"import_method": self.current_geometry.get("import_method", "unknown")
}
logger.info("✓ Real geometry validation completed")
return validation_result
except Exception as e:
logger.error(f"Geometry validation error: {str(e)}")
return {
"valid": False,
"error": str(e)
}
except Exception as e:
logger.error(f"Validation error: {str(e)}")
return {
"valid": False,
"error": str(e)
}
def create_named_selections(self) -> Dict[str, Any]:
"""
Create named selections for blade geometry
Returns:
Dictionary with creation results
"""
try:
if not self.is_session_active:
logger.error("No active ANSYS session")
return {
"success": False,
"error": "No active ANSYS session"
}
if not self.current_geometry:
logger.error("No geometry imported")
return {
"success": False,
"error": "No geometry imported"
}
logger.info("Creating named selections for blade geometry...")
# Real named selection creation using PyMechanical
if not self.named_selection_manager:
logger.error("Named selection manager not initialized")
return {
"success": False,
"error": "Named selection manager not initialized"
}
result = self.named_selection_manager.create_blade_named_selections()
if result["success"]:
logger.info(f"✓ Named selections created successfully: {result['total_selections']} selections")
else:
logger.error(f"Named selection creation failed: {result.get('error', 'Unknown error')}")
return result
except Exception as e:
logger.error(f"Named selection creation error: {str(e)}")
return {
"success": False,
"error": str(e),
"selections_created": [],
"selections_failed": [],
"total_selections": 0,
"created_at": datetime.now()
}
def get_named_selections_info(self) -> Dict[str, Any]:
"""
Get information about created named selections
Returns:
Dictionary with named selection information
"""
try:
if not self.is_session_active:
return {
"success": False,
"error": "No active ANSYS session"
}
if not self.named_selection_manager:
return {
"success": False,
"error": "Named selection manager not initialized"
}
return self.named_selection_manager.get_named_selections_info()
except Exception as e:
logger.error(f"Error getting named selections info: {str(e)}")
return {
"success": False,
"error": str(e)
}
def validate_named_selections(self) -> Dict[str, Any]:
"""
Validate created named selections
Returns:
Dictionary with validation results
"""
try:
if not self.is_session_active:
return {
"success": False,
"error": "No active ANSYS session"
}
if not self.named_selection_manager:
return {
"success": False,
"error": "Named selection manager not initialized"
}
return self.named_selection_manager.validate_named_selections()
except Exception as e:
logger.error(f"Named selection validation error: {str(e)}")
return {
"success": False,
"error": str(e)
}
def apply_mesh_controls(self, named_selections: List[str] = None) -> Dict[str, Any]:
"""
Apply mesh controls for blade geometry
Args:
named_selections: List of named selections to use for controls
Returns:
Dictionary with mesh control application results
"""
try:
if not self.is_session_active:
logger.error("No active ANSYS session")
return {
"success": False,
"error": "No active ANSYS session"
}
if not self.current_geometry:
logger.error("No geometry imported")
return {
"success": False,
"error": "No geometry imported"
}
logger.info("Applying mesh controls for blade geometry...")
# Real mesh control application using PyMechanical
if not self.mesh_controller:
logger.error("Mesh controller not initialized")
return {
"success": False,
"error": "Mesh controller not initialized"
}
# Use provided named selections or get from named selection manager
if named_selections is None:
if self.named_selection_manager:
# Get created named selections
info_result = self.named_selection_manager.get_named_selections_info()
if info_result.get('success'):
named_selections = list(info_result.get('created_selections', {}).keys())
else:
named_selections = ['leading_edge', 'trailing_edge', 'blade_root', 'blade_surfaces']
else:
named_selections = ['leading_edge', 'trailing_edge', 'blade_root', 'blade_surfaces']
result = self.mesh_controller.apply_all_mesh_controls(named_selections)
if result["success"]:
logger.info(f"✓ Mesh controls applied successfully: {len(result['controls_applied'])} controls")
else:
logger.warning(f"⚠ Mesh controls partially applied: {result.get('controls_applied', [])} successful, {result.get('controls_failed', [])} failed")
return result
except Exception as e:
logger.error(f"Mesh control application error: {str(e)}")
return {
"success": False,
"error": str(e),
"controls_applied": [],
"controls_failed": ["all"],
"applied_at": datetime.now()
}
def get_mesh_control_summary(self) -> Dict[str, Any]:
"""
Get summary of applied mesh controls
Returns:
Dictionary with mesh control summary
"""
try:
if not self.is_session_active:
return {
"success": False,
"error": "No active ANSYS session"
}
if not self.mesh_controller:
return {
"success": False,
"error": "Mesh controller not initialized"
}
return self.mesh_controller.get_mesh_control_summary()
except Exception as e:
logger.error(f"Failed to get mesh control summary: {str(e)}")
return {
"success": False,
"error": str(e)
}
def generate_mesh(self, progress_callback: Optional[Callable[[float, str], None]] = None) -> Dict[str, Any]:
"""
Generate mesh for the imported geometry
Args:
progress_callback: Optional callback for progress updates
Returns:
Dictionary with mesh generation results
"""
try:
if not self.is_session_active:
logger.error("No active ANSYS session")
return {
"success": False,
"error": "No active ANSYS session"
}
if not self.current_geometry:
logger.error("No geometry imported")
return {
"success": False,
"error": "No geometry imported"
}
logger.info("Starting mesh generation...")
# Real mesh generation using PyMechanical
if not self.mesh_generator:
logger.error("Mesh generator not initialized")
return {
"success": False,
"error": "Mesh generator not initialized"
}
# Set progress callback if provided
if progress_callback:
self.mesh_generator.set_progress_callback(progress_callback)
# Generate mesh
result = self.mesh_generator.generate_mesh()
# Convert result to dictionary
result_dict = {
"success": result.success,
"status": result.status.value,
"element_count": result.element_count,
"node_count": result.node_count,
"generation_time": result.generation_time,
"progress_percentage": result.progress_percentage,
"started_at": result.started_at.isoformat() if result.started_at else None,
"completed_at": result.completed_at.isoformat() if result.completed_at else None,
"error_message": result.error_message,
"warnings": result.warnings
}
if result.success:
logger.info(f"✓ Mesh generation completed: {result.element_count} elements, {result.node_count} nodes")
else:
logger.error(f"✗ Mesh generation failed: {result.error_message}")
return result_dict
except Exception as e:
logger.error(f"Mesh generation error: {str(e)}")
return {
"success": False,
"error": str(e),
"element_count": 0,
"node_count": 0,
"generation_time": 0.0,
"started_at": datetime.now().isoformat(),
"completed_at": datetime.now().isoformat()
}
def get_mesh_statistics(self) -> Dict[str, Any]:
"""
Get detailed mesh statistics
Returns:
Dictionary with mesh statistics
"""
try:
if not self.is_session_active:
return {
"success": False,
"error": "No active ANSYS session"
}
if not self.mesh_generator:
return {
"success": False,
"error": "Mesh generator not initialized"
}
return self.mesh_generator.get_mesh_statistics()
except Exception as e:
logger.error(f"Failed to get mesh statistics: {str(e)}")
return {
'error': str(e),
'element_count': 0,
'node_count': 0,
'has_mesh': False,
'retrieved_at': datetime.now()
}
def check_mesh_quality(self) -> Dict[str, Any]:
"""
Check mesh quality using ANSYS quality metrics
Returns:
Dictionary with quality check results
"""
try:
if not self.is_session_active:
logger.error("No active ANSYS session")
return {
"success": False,
"error": "No active ANSYS session"
}
logger.info("Starting mesh quality check...")
# Real quality check using PyMechanical
if not self.mesh_quality_checker:
logger.error("Mesh quality checker not initialized")
return {
"success": False,
"error": "Mesh quality checker not initialized"
}
result = self.mesh_quality_checker.check_mesh_quality()
if result.get("success", False):
logger.info(f"✓ Mesh quality check completed: {result.get('overall_status', 'UNKNOWN')}")
else:
logger.warning(f"⚠ Mesh quality check had issues: {result.get('error', 'Unknown error')}")
return result
except Exception as e:
logger.error(f"Mesh quality check error: {str(e)}")
return {
"success": False,
"error": str(e),
"quality_score": 0.0,
"overall_status": "ERROR",
"checked_at": datetime.now()
}
def validate_mesh_quality(self, quality_thresholds: Dict[str, float] = None) -> Dict[str, Any]:
"""
Validate mesh quality against specified thresholds
Args:
quality_thresholds: Dictionary of quality thresholds
Returns:
Dictionary with validation results
"""
try:
if not self.is_session_active:
return {
"success": False,
"error": "No active ANSYS session"
}
if not self.mesh_quality_checker:
return {
"success": False,
"error": "Mesh quality checker not initialized"
}
return self.mesh_quality_checker.validate_mesh_quality(quality_thresholds)
except Exception as e:
logger.error(f"Mesh quality validation error: {str(e)}")
return {
"success": False,
"error": str(e)
}
def close_session(self) -> bool:
"""
Close ANSYS Mechanical session and clean up resources
Returns:
True if successful, False otherwise
"""
try:
logger.info("Closing ANSYS session...")
# Clean up temp files first
self.cleanup_temp_files()
# Real session cleanup
if self.mechanical:
try:
self.mechanical.exit()
logger.info("✓ ANSYS Mechanical session closed")
except Exception as e:
logger.warning(f"Error closing ANSYS session: {str(e)}")
# Reset session state
self.session = None
self.mechanical = None
self.is_session_active = False
self.current_geometry = None
self.named_selection_manager = None
self.mesh_controller = None
self.mesh_generator = None
self.mesh_quality_checker = None
logger.info("✓ Session cleanup completed")
return True
except Exception as e:
logger.error(f"Session cleanup error: {str(e)}")
return False
def cleanup_temp_files(self):
"""Clean up temporary files created during processing"""
try:
# Use resource manager for cleanup
resource_manager.cleanup_temp_files()
# Clean up session-specific temp files
for temp_file in self.temp_files:
try:
if os.path.exists(temp_file):
os.remove(temp_file)
logger.debug(f"Removed temp file: {temp_file}")
except Exception as e:
logger.warning(f"Could not remove temp file {temp_file}: {str(e)}")
self.temp_files.clear()
except Exception as e:
logger.warning(f"Temp file cleanup error: {str(e)}")
def get_session_info(self) -> Dict[str, Any]:
"""
Get current session information
Returns:
Dictionary with session information
"""
session_time = 0.0
if self.session_start_time:
session_time = (datetime.now() - self.session_start_time).total_seconds()
return {
"is_active": self.is_session_active,
"session_time": session_time,
"has_geometry": self.current_geometry is not None,
"geometry_info": self.current_geometry,
"temp_files_count": len(self.temp_files),
"components_initialized": {
"named_selection_manager": self.named_selection_manager is not None,
"mesh_controller": self.mesh_controller is not None,
"mesh_generator": self.mesh_generator is not None,
"mesh_quality_checker": self.mesh_quality_checker is not None
}
}
def __enter__(self):
"""Context manager entry"""
if self.start_session():
return self
else:
raise Exception("Failed to start ANSYS session")
def __exit__(self, exc_type, exc_val, exc_tb):
"""Context manager exit"""
self.close_session()