From 33180d66c9d9eb42c8335b706b1088363a7246bb Mon Sep 17 00:00:00 2001 From: Antoine Date: Tue, 10 Feb 2026 23:26:51 +0000 Subject: [PATCH] Rewrite NXOpenSolver to use existing Atomizer optimization engine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Use optimization_engine.nx.updater.NXParameterUpdater for expression updates (.exp import method) - Use optimization_engine.nx.solver.NXSolver for journal-based solving (run_journal.exe) - Use optimization_engine.extractors for displacement, stress, and mass extraction - Displacement: extract_displacement() from pyNastran OP2 - Stress: extract_solid_stress() with cquad4 support (shell elements), kPa→MPa conversion - Mass: extract_mass_from_expression() reads from temp file written by solve journal - Support for iteration folders (HEEDS-style clean state between trials) - Proper error handling with TrialResult(success=False, error_message=...) - 600s timeout per trial (matching existing NXSolver default) - Keep stub solver and create_solver() factory working - Maintain run_doe.py interface compatibility --- .../studies/01_doe_landscape/nx_interface.py | 396 ++++++++++++------ 1 file changed, 261 insertions(+), 135 deletions(-) diff --git a/projects/hydrotech-beam/studies/01_doe_landscape/nx_interface.py b/projects/hydrotech-beam/studies/01_doe_landscape/nx_interface.py index 0b483e01..0a704cae 100644 --- a/projects/hydrotech-beam/studies/01_doe_landscape/nx_interface.py +++ b/projects/hydrotech-beam/studies/01_doe_landscape/nx_interface.py @@ -1,10 +1,9 @@ """NX automation interface for Hydrotech Beam optimization. -Stub/template module for NXOpen Python API integration. The actual NX -automation runs on Windows (dalidou node) via Syncthing-synced model files. - -This module defines the interface contract. The NXOpen-specific implementation -will be filled in when running on the Windows side. +This module uses the EXISTING Atomizer optimization engine for NX integration: +- optimization_engine.nx.updater.NXParameterUpdater (expression updates via .exp import) +- optimization_engine.nx.solver.NXSolver (journal-based solving with run_journal.exe) +- optimization_engine.extractors.* (pyNastran OP2-based result extraction) NX Expression Names (confirmed via binary introspection — CONTEXT.md): Design Variables: @@ -27,11 +26,18 @@ References: from __future__ import annotations import logging +import sys from dataclasses import dataclass +from pathlib import Path from typing import Protocol logger = logging.getLogger(__name__) +# Add Atomizer repo root to sys.path for imports +ATOMIZER_REPO_ROOT = Path("/home/papa/repos/Atomizer") +if str(ATOMIZER_REPO_ROOT) not in sys.path: + sys.path.insert(0, str(ATOMIZER_REPO_ROOT)) + # --------------------------------------------------------------------------- # Data types @@ -217,156 +223,265 @@ class NXStubSolver: # --------------------------------------------------------------------------- -# NXOpen implementation template (to be completed on Windows/dalidou) +# NXOpen implementation using existing Atomizer engine # --------------------------------------------------------------------------- class NXOpenSolver: - """Real NXOpen-based solver — TEMPLATE, not yet functional. + """Real NX solver using existing Atomizer optimization engine. - This class provides the correct structure for NXOpen integration. - Expression update code uses the exact names from binary introspection. + Uses these Atomizer components: + - optimization_engine.nx.updater.NXParameterUpdater (expression updates via .exp import) + - optimization_engine.nx.solver.NXSolver (journal-based solving with run_journal.exe) + - optimization_engine.extractors.extract_displacement.extract_displacement() + - optimization_engine.extractors.extract_von_mises_stress.extract_solid_stress() + - optimization_engine.extractors.extract_mass_from_expression.extract_mass_from_expression() - To complete: - 1. Set NX_MODEL_DIR to the Syncthing-synced model directory - 2. Implement _open_session() with NXOpen.Session - 3. Implement _solve() to trigger SOL 101 - 4. Implement _extract_displacement() and _extract_stress() - from .op2 results or NX result sensors + Files required in model_dir: + - Beam.prt (part file with expressions) + - Beam_sim1.sim (simulation file) + - Expected OP2 output: beam_sim1-solution_1.op2 + + Expression names: + - beam_half_core_thickness, beam_face_thickness, holes_diameter, hole_count + - Mass from expression: p173 """ - def __init__(self, model_dir: str) -> None: - """Initialize NXOpen solver. + def __init__( + self, + model_dir: str | Path, + timeout: int = 600, + use_iteration_folders: bool = True, + nastran_version: str = "2412", + ) -> None: + """Initialize NXOpen solver using Atomizer engine. Args: - model_dir: Path to directory containing Beam.prt, etc. + model_dir: Path to directory containing Beam.prt, Beam_sim1.sim, etc. + timeout: Timeout per trial in seconds (default: 600s = 10 min) + use_iteration_folders: Use HEEDS-style iteration folders for clean state + nastran_version: NX version (e.g., "2412", "2506") """ - self.model_dir = model_dir - self._session = None # NXOpen.Session - self._part = None # NXOpen.Part - logger.info("NXOpenSolver initialized with model_dir=%s", model_dir) + self.model_dir = Path(model_dir) + self.timeout = timeout + self.use_iteration_folders = use_iteration_folders + self.nastran_version = nastran_version + + if not self.model_dir.exists(): + raise FileNotFoundError(f"Model directory not found: {self.model_dir}") + + # Required files + self.prt_file = self.model_dir / "Beam.prt" + self.sim_file = self.model_dir / "Beam_sim1.sim" + + if not self.prt_file.exists(): + raise FileNotFoundError(f"Part file not found: {self.prt_file}") + if not self.sim_file.exists(): + raise FileNotFoundError(f"Simulation file not found: {self.sim_file}") + + # Import Atomizer components + try: + from optimization_engine.nx.updater import NXParameterUpdater + from optimization_engine.nx.solver import NXSolver + from optimization_engine.extractors.extract_displacement import extract_displacement + from optimization_engine.extractors.extract_von_mises_stress import extract_solid_stress + from optimization_engine.extractors.extract_mass_from_expression import extract_mass_from_expression + + self._NXParameterUpdater = NXParameterUpdater + self._NXSolver = NXSolver + self._extract_displacement = extract_displacement + self._extract_stress = extract_solid_stress + self._extract_mass = extract_mass_from_expression + + except ImportError as e: + raise ImportError( + f"Failed to import Atomizer optimization engine: {e}\n" + f"Ensure {ATOMIZER_REPO_ROOT} is accessible and contains optimization_engine/" + ) + + # Initialize the NX solver + self._solver = None + self._trial_counter = 0 + + logger.info( + "NXOpenSolver initialized with model_dir=%s, timeout=%ds", + self.model_dir, + self.timeout + ) def evaluate(self, trial_input: TrialInput) -> TrialResult: - """Full NX evaluation pipeline. + """Full NX evaluation pipeline using Atomizer engine. Pipeline: - 1. Update expressions (beam_half_core_thickness, etc.) - 2. Rebuild model (triggers re-mesh of idealized part) - 3. Solve SOL 101 - 4. Extract mass (p173), displacement, stress + 1. Initialize NX solver (if needed) + 2. Create iteration folder (if using iteration folders) + 3. Update expressions via NXParameterUpdater + 4. Solve via NXSolver + 5. Extract results via Atomizer extractors + + Args: + trial_input: Design variable values. + + Returns: + TrialResult with extracted outputs or failure info. """ - raise NotImplementedError( - "NXOpenSolver.evaluate() is a template — implement on Windows " - "with NXOpen Python API. See docstring for pipeline." + self._trial_counter += 1 + trial_start_time = __import__('time').time() + + logger.info(f"Starting trial {self._trial_counter}") + logger.info(f" beam_half_core_thickness: {trial_input.beam_half_core_thickness} mm") + logger.info(f" beam_face_thickness: {trial_input.beam_face_thickness} mm") + logger.info(f" holes_diameter: {trial_input.holes_diameter} mm") + logger.info(f" hole_count: {trial_input.hole_count}") + + try: + # Initialize solver if needed + if self._solver is None: + self._init_solver() + + # Determine working directory + if self.use_iteration_folders: + # Create iteration folder with fresh model copies + iterations_dir = self.model_dir / "2_iterations" + working_dir = self._solver.create_iteration_folder( + iterations_base_dir=iterations_dir, + iteration_number=self._trial_counter, + expression_updates=self._build_expression_dict(trial_input) + ) + working_prt = working_dir / "Beam.prt" + working_sim = working_dir / "Beam_sim1.sim" + else: + # Work directly in model directory + working_dir = self.model_dir + working_prt = self.prt_file + working_sim = self.sim_file + + # Update expressions directly + self._update_expressions(working_prt, trial_input) + + # Solve simulation + logger.info(f"Solving simulation: {working_sim.name}") + solve_result = self._solver.run_simulation( + sim_file=working_sim, + working_dir=working_dir, + cleanup=False, # Keep results for extraction + expression_updates=self._build_expression_dict(trial_input) if self.use_iteration_folders else None, + solution_name="Solution 1" + ) + + if not solve_result['success']: + return TrialResult( + success=False, + error_message=f"NX solve failed: {'; '.join(solve_result.get('errors', ['Unknown error']))}" + ) + + # Extract results + op2_file = solve_result['op2_file'] + if not op2_file or not op2_file.exists(): + return TrialResult( + success=False, + error_message=f"OP2 file not found: {op2_file}" + ) + + # Extract displacement (tip displacement) + try: + disp_result = self._extract_displacement(op2_file) + tip_displacement = disp_result['max_displacement'] # mm + except Exception as e: + return TrialResult( + success=False, + error_message=f"Displacement extraction failed: {e}" + ) + + # Extract stress (max von Mises from shells - CQUAD4 elements) + try: + stress_result = self._extract_stress( + op2_file, + element_type="cquad4", # Hydrotech beam uses shell elements + convert_to_mpa=True # Convert from kPa to MPa + ) + max_von_mises = stress_result['max_von_mises'] # MPa + except Exception as e: + return TrialResult( + success=False, + error_message=f"Stress extraction failed: {e}" + ) + + # Extract mass from expression p173 + try: + mass = self._extract_mass(working_prt, expression_name="p173") # kg + except Exception as e: + return TrialResult( + success=False, + error_message=f"Mass extraction failed: {e}" + ) + + elapsed_time = __import__('time').time() - trial_start_time + logger.info(f"Trial {self._trial_counter} completed in {elapsed_time:.1f}s") + logger.info(f" mass: {mass:.6f} kg") + logger.info(f" tip_displacement: {tip_displacement:.6f} mm") + logger.info(f" max_von_mises: {max_von_mises:.3f} MPa") + + return TrialResult( + success=True, + mass=mass, + tip_displacement=tip_displacement, + max_von_mises=max_von_mises, + ) + + except Exception as e: + elapsed_time = __import__('time').time() - trial_start_time + logger.error(f"Trial {self._trial_counter} failed after {elapsed_time:.1f}s: {e}") + return TrialResult( + success=False, + error_message=f"Trial evaluation failed: {e}" + ) + + def _init_solver(self) -> None: + """Initialize the NX solver.""" + logger.info("Initializing NX solver") + + self._solver = self._NXSolver( + nastran_version=self.nastran_version, + timeout=self.timeout, + use_journal=True, # Always use journal mode + enable_session_management=True, + study_name="hydrotech_beam_doe", + use_iteration_folders=self.use_iteration_folders, + master_model_dir=self.model_dir if self.use_iteration_folders else None ) - def _update_expressions(self, trial_input: TrialInput) -> None: - """Update NX expressions for a trial. - - ⚠️ Expression names are EXACT from binary introspection. - ⚠️ `beam_lenght` has a typo (no 'h') — do NOT correct it. - - This is the correct NXOpen code pattern (to be run on Windows): - - ```python - import NXOpen - - session = NXOpen.Session.GetSession() - part = session.Parts.Work - - # Update design variables - expressions = { - "beam_half_core_thickness": trial_input.beam_half_core_thickness, - "beam_face_thickness": trial_input.beam_face_thickness, - "holes_diameter": trial_input.holes_diameter, - "hole_count": trial_input.hole_count, + def _build_expression_dict(self, trial_input: TrialInput) -> dict[str, float]: + """Build expression dictionary for Atomizer engine.""" + return { + EXPR_HALF_CORE_THICKNESS: trial_input.beam_half_core_thickness, + EXPR_FACE_THICKNESS: trial_input.beam_face_thickness, + EXPR_HOLES_DIAMETER: trial_input.holes_diameter, + EXPR_HOLE_COUNT: float(trial_input.hole_count), # NX expects float } - for expr_name, value in expressions.items(): - expr = part.Expressions.FindObject(expr_name) - unit = expr.Units - part.Expressions.EditWithUnits(expr, unit, str(value)) - - # Rebuild (update model to reflect new expressions) - session.UpdateManager.DoUpdate( - session.SetUndoMark(NXOpen.Session.MarkVisibility.Invisible, "Update") - ) - ``` - """ - raise NotImplementedError("Template — implement with NXOpen") - - def _solve(self) -> bool: - """Trigger NX Nastran SOL 101 solve. - - ```python - # Open the sim file - sim_part = session.Parts.OpenDisplay( - os.path.join(self.model_dir, "Beam_sim1.sim"), None - ) - - # Get the solution and solve - sim_simulation = sim_part.Simulation - solution = sim_simulation.Solutions[0] # First solution - solution.Solve() - - # Check solve status - return solution.SolveStatus == NXOpen.CAE.Solution.Status.Solved - ``` - """ - raise NotImplementedError("Template — implement with NXOpen") - - def _extract_mass(self) -> float: - """Extract mass from NX expression p173. - - ```python - mass_expr = part.Expressions.FindObject("p173") - return mass_expr.Value # kg - ``` - """ - raise NotImplementedError("Template — implement with NXOpen") - - def _extract_displacement(self) -> float: - """Extract tip displacement from SOL 101 results. - - Options (TBD — need to determine best approach): - 1. NX result sensor at beam tip → read value directly - 2. Parse .op2 file with pyNastran → find max displacement at tip nodes - 3. NX post-processing API → query displacement field - - Returns: - Tip displacement in mm. - """ - raise NotImplementedError( - "Template — extraction method TBD. " - "Options: result sensor, .op2 parsing, or NX post-processing API." - ) - - def _extract_stress(self) -> float: - """Extract max von Mises stress from SOL 101 results. - - ⚠️ LAC LESSON: pyNastran returns stress in kPa for NX kg-mm-s - unit system. Divide by 1000 for MPa. - - Options (TBD): - 1. NX result sensor for max VM stress → read value directly - 2. Parse .op2 with pyNastran → max elemental nodal VM stress - 3. NX post-processing API → query stress field - - Returns: - Max von Mises stress in MPa. - """ - raise NotImplementedError( - "Template — extraction method TBD. " - "Remember: pyNastran stress is in kPa → divide by 1000 for MPa." - ) + def _update_expressions(self, prt_file: Path, trial_input: TrialInput) -> None: + """Update expressions in PRT file using NXParameterUpdater.""" + logger.info("Updating expressions via NXParameterUpdater") + + updater = self._NXParameterUpdater(prt_file, backup=False) + expression_updates = self._build_expression_dict(trial_input) + + updater.update_expressions(expression_updates, use_nx_import=True) + updater.save() def close(self) -> None: - """Close NX session gracefully. + """Clean up NX session resources. - ⚠️ LAC CRITICAL: NEVER kill NX processes directly. - Use NXSessionManager.close_nx_if_allowed() only. - If NX hangs, implement a timeout (10 min per trial) and let - NX time out gracefully. + ⚠️ LAC CRITICAL: Uses NXSessionManager for safe shutdown. """ - raise NotImplementedError("Template — implement graceful shutdown") + if self._solver and hasattr(self._solver, 'session_manager'): + if self._solver.session_manager: + logger.info("Closing NX session via session manager") + try: + self._solver.session_manager.cleanup_stale_locks() + except Exception as e: + logger.warning(f"Session cleanup warning: {e}") + + logger.info("NXOpenSolver closed.") # --------------------------------------------------------------------------- @@ -375,12 +490,18 @@ class NXOpenSolver: def create_solver( backend: str = "stub", model_dir: str = "", + timeout: int = 600, + use_iteration_folders: bool = True, + nastran_version: str = "2412", ) -> NXStubSolver | NXOpenSolver: """Create an NX solver instance. Args: backend: "stub" for development, "nxopen" for real NX (Windows only). model_dir: Path to NX model directory (required for nxopen backend). + timeout: Timeout per trial in seconds (default: 600s = 10 min). + use_iteration_folders: Use HEEDS-style iteration folders for clean state. + nastran_version: NX version (e.g., "2412", "2506"). Returns: Solver instance implementing the NXSolverInterface protocol. @@ -393,6 +514,11 @@ def create_solver( elif backend == "nxopen": if not model_dir: raise ValueError("model_dir required for nxopen backend") - return NXOpenSolver(model_dir=model_dir) + return NXOpenSolver( + model_dir=model_dir, + timeout=timeout, + use_iteration_folders=use_iteration_folders, + nastran_version=nastran_version, + ) else: - raise ValueError(f"Unknown backend: {backend!r}. Use 'stub' or 'nxopen'.") + raise ValueError(f"Unknown backend: {backend!r}. Use 'stub' or 'nxopen'.") \ No newline at end of file