@@ -229,93 +229,131 @@ class NXOpenSolver:
""" Real NX solver using existing Atomizer optimization engine.
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)
→ handles iteration folders, expression import via .exp, and NX solve
- 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()
Pipeline per trial (HEEDS-style iteration folder pattern):
1. NXSolver.create_iteration_folder() — copies model files + writes params.exp
2. NXSolver.run_simulation() — runs solve_simulation.py journal via run_journal.exe
→ The journal imports params.exp, rebuilds geometry, updates FEM, solves, extracts mass
3. extract_displacement(op2) — max displacement from SOL 101
4. extract_solid_stress(op2) — max von Mises (auto-detect element type)
5. extract_mass_from_expression(prt) — reads _temp_mass.txt written by journal
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
Expression names (confirmed from binary introspection) :
- DVs: beam_half_core_thickness, beam_face_thickness, holes_diameter, hole_count
- Mass output: p173 (body_property147.mass, kg)
References:
- M1_Mirror/SAT3_Trajectory_V7/run_optimization.py — FEARunner pattern
- optimization_engine/nx/solver.py — NXSolver API
- optimization_engine/nx/solve_simulation.py — Journal internals
"""
# SIM filename and solution name for this model
SIM_FILENAME = " Beam_sim1.sim "
PRT_FILENAME = " Beam.prt "
SOLUTION_NAME = " Solution 1 "
# Expected OP2: <sim_stem>-<solution_name_lower_underscored>.op2
# = beam_sim1-solution_1.op2
def __init__ (
self ,
model_dir : str | Path ,
nx_install_dir : str | Path | None = None ,
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, 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 " )
This is the " master model " directory — files are copied per iteration.
nx_install_dir: Path to NX installation (auto-detected if None).
timeout: Timeout per trial in seconds (default: 600s = 10 min).
nastran_version: NX version string (e.g., " 2412 " , " 2506 " , " 2512 " ).
"""
import time as _time # avoid repeated __import__
self . _time = _time
self . model_dir = Path ( model_dir )
self . timeout = timeout
self . use_iteration_folders = use_iteration_folders
self . nastran_version = nastran_version
self . nx_install_dir = str ( nx_install_dir ) if nx_install_dir else None
if not self . model_dir . exists ( ) :
raise FileNotFoundError ( f " Model directory not found: { self . model_dir } " )
# R equired files
self . prt_file = self . model_dir / " Beam.prt "
self . sim_file = self . model_dir / " Beam_sim1.sim "
# Validate r equired files
self . prt_file = self . model_dir / self . PRT_FILENAME
self . sim_file = self . model_dir / self . SIM_FILENAME
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 } " )
for f in ( self . prt_file , self . sim_file ) :
if not f . exists ( ) :
raise FileNotFoundError ( f " Required file not found: { f } " )
# Import Atomizer components
# Iterations output directory (sibling to model_dir per study convention)
# Layout: studies/01_doe_landscape/
# 1_setup/model/ ← model_dir (master)
# 2_iterations/ ← iteration folders
# 3_results/ ← final outputs
self . iterations_dir = self . model_dir . parent . parent / " 2_iterations "
self . iterations_dir . mkdir ( parents = True , exist_ok = True )
# Import Atomizer components at init time (fail-fast on missing engine)
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
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
self . _extract_displacement = staticmethod ( extract_displacement )
self . _extract_stress = staticmethod ( extract_solid_stress )
self . _extract_mass = staticmethod ( 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/ "
)
) from e
# Initialize the NX solver
self . _solver = None
self . _trial_counter = 0
# Lazy-init solver on first evaluate() call
self . _solver : object | None = None
self . _trial_counter : int = 0
logger . info (
" NXOpenSolver initialized with model_dir= %s , timeout= %d s " ,
self . model_dir ,
self . timeout
" NXOpenSolver initialized — model_dir= %s , timeout= %d s, nastran= %s " ,
self . model_dir ,
self . timeout ,
self . nastran_version ,
)
def evaluate ( self , trial_input : TrialInput ) - > TrialResult :
""" Full NX evaluation pipeline using Atomizer engine.
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
Pipeline:
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 extractor s
def evaluate ( self , trial_input : TrialInput ) - > TrialResult :
""" Full NX evaluation pipeline for one trial.
Pipeline (mirrors M1_Mirror/SAT3_Trajectory_V7 FEARunner.run_fea):
1 . create_iteration_folder → copies model + writes params.exp
2 . run_simulation → journal updates expressions, rebuilds, solve s
3. extract displacement, stress, mass from results
Args:
trial_input: Design variable values.
@@ -324,101 +362,141 @@ class NXOpenSolver:
TrialResult with extracted outputs or failure info.
"""
self . _trial_counter + = 1
trial_start_time = __import__ ( ' time ' ) . time ( )
trial_num = self . _trial_counter
t_start = self . _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 } " )
logger . info (
" Trial %d — DVs: core= %.3f mm, face= %.3f mm, hole_d= %.3f mm, n_holes= %d " ,
trial_num ,
trial_input . beam_half_core_thickness ,
trial_input . beam_face_thickness ,
trial_input . holes_diameter ,
trial_input . hole_count ,
)
try :
# Initialize solver if needed
# 0. Lazy-init solver
if self . _solver is None :
self . _init_solver ( )
# Determine working directory
if self . use_iteration_folders :
# Create iteration folder with fresh model copies
iterations_di r = self . model_dir / " 2_iterations "
working _dir = self . _solver . create_ iteration_folde r(
iterations_base_dir = iterations_dir ,
iteration_number = self . _trial_counter ,
expression_updates = self . _build_expression_dict ( trial_input )
expressions = self . _build_expression_dict ( trial_input )
# 1. Create iteration folder with fresh model copies + params.exp
iter_folde r = self . _solver . create_iteration_folder (
iterations_base _dir= self . iterations_di r ,
iteration_number = trial_num ,
expression_updates = expressions ,
)
logger . info ( " Iteration folder: %s " , iter_folder )
working_sim = iter_folder / self . SIM_FILENAME
working_prt = iter_folder / self . PRT_FILENAME
if not working_sim . exists ( ) :
return TrialResult (
success = False ,
error_message = f " SIM file missing in iteration folder: { working_sim } " ,
)
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 } " )
# 2. Solve — journal handles expression import + geometry rebuild + FEM update + solve
# expression_updates are passed as argv to the journal (key=value pairs )
logger . info ( " Solving: %s " , working_sim . name )
solve_result = self . _solver . run_simulation (
sim_file = working_sim ,
working_dir = working_di r,
cleanup = False , # K eep results for extraction
expression_updates = self . _build_expression_dict ( trial_input ) if self . use_iteration_folders else None ,
solution_name = " Solution 1 "
working_dir = iter_folde r,
cleanup = False , # k eep OP2/F06 for extraction
expression_updates = expressions ,
solution_name = self . SOLUTION_NAME ,
)
if not solve_result [ ' ' ] :
return TrialResult (
success = False ,
error_message = f " NX solve failed: { ' ; ' . join ( solve_result . get ( ' errors ' , [ ' Unknown error ' ] ) ) } "
)
if not solve_result [ " " ] :
errors = solve_result . get ( " errors " , [ " Unknown error " ] )
rc = solve_result . get ( " return_code " , " ? " )
msg = f " NX solve failed (rc= { rc } ) : { ' ; ' . join ( errors ) } "
logger . error ( " %s " , msg )
return TrialResult ( success = False , error_message = msg )
# Extract results
op2_file = solve_result [ ' ' ]
if not op2_file or not op2_file . exists ( ) :
return TrialResult (
success = False ,
error_message = f " OP2 file not found: { op2_file } "
)
# 3. Locate OP2
op2_file = solve_result . get ( " " )
if op2_file is None or not Path ( ) . exists ( ) :
# Fallback: try the expected naming convention
op2_file = iter_folder / " beam_sim1-solution_1.op2 "
if not op2_file . exists ( ) :
return TrialResult (
success = False ,
error_message = f " OP2 not found. Expected: { op2_file } " ,
)
else :
op2_file = Path ( op2_file )
# Extract displacement (tip displacement )
logger . info ( " OP2: %s " , op2_file . name
# 4a. Extract displacement
try :
disp_result = self . _extract_displacement ( op2_file )
tip_displacement = disp_result [ ' ' ] # mm
tip_displacement = disp_result [ " " ] # mm
except Exception as e :
logger . error ( " Displacement extraction failed: %s " , e )
return TrialResult (
success = False ,
error_message = f " Displacement extraction failed: { e } "
error_message = f " Displacement extraction failed: { e } " ,
)
# Extract stress (max von Mises from shells - CQUAD4 elements )
# 4b. Extract stress — auto-detect element type (solid or shell )
# Pass element_type=None so it checks CTETRA, CHEXA, CPENTA, CPYRAM
try :
stress_result = self . _extract_stress (
op2_file ,
element_type = " cquad4 " , # Hydrotech beam uses shell elem ents
convert_to_mpa = True # Convert from kPa to MPa
op2_file ,
element_type = None , # auto-detect from OP2 cont ents
convert_to_mpa = True , # NX kg-mm-s → kPa, convert to MPa
)
max_von_mises = stress_result [ ' ' ] # MPa
max_von_mises = stress_result [ " " ] # MPa
except Exception as e :
return TrialResult (
success = False ,
error_message = f " Stress extraction failed: { e } "
)
# Fallback: try shell elements if solid extraction failed
logger . warning ( " Solid stress extraction failed, trying shell: %s " , e )
try :
stress_result = self . _extract_stress (
op2_file ,
element_type = " cquad4 " ,
convert_to_mpa = True ,
)
max_von_mises = stress_result [ " max_von_mises " ]
except Exception as e2 :
logger . error ( " Stress extraction failed (all types): %s " , e2 )
return TrialResult (
success = False ,
error_message = f " Stress extraction failed: { e } ; shell fallback: { e2 } " ,
)
# Extract mass from expression p173
# 4c. Extract mass — reads _temp_mass.txt written by solve_simulation.py journal
try :
mass = self . _extract_mass ( working_prt , expression_name = " p173 " ) # kg
mass = self . _extract_mass ( working_prt , expression_name = EXPR_MASS ) # kg
except FileNotFoundError :
# _temp_mass.txt not found — journal may not have written it for single-part models
# Fallback: try reading from _temp_part_properties.json
logger . warning ( " _temp_mass.txt not found, trying _temp_part_properties.json " )
mass = self . _extract_mass_fallback ( iter_folder )
if mass is None :
return TrialResult (
success = False ,
error_message = " Mass extraction failed: _temp_mass.txt not found " ,
)
except Exception as e :
logger . error ( " Mass extraction failed: %s " , e )
return TrialResult (
success = False ,
error_message = f " Mass extraction failed: { e } "
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 " )
elapsed = self . _time . time ( ) - t_start
logger . info (
" Trial %d OK ( %.1f s) — mass= %.4f kg, disp= %.4f mm, σ %.2f MPa " ,
trial_num ,
elapsed ,
mass ,
tip_displacement ,
max_von_mises ,
)
return TrialResult (
success = True ,
@@ -428,61 +506,88 @@ class NXOpenSolver:
)
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 } " )
elapsed = self . _time . time ( ) - t_start
logger . error ( " Trial %d FAILED ( %.1f s): %s " , trial_num , elapsed , e )
return TrialResult (
success = False ,
error_message = f " Trial evaluation failed : { e } "
error_message = f " Unexpected error in trial { trial_num } { 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 _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
}
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 :
""" Clean up NX session resources.
⚠️ LAC CRITICAL: Uses NXSessionManager for safe shutdown .
⚠️ LAC CRITICAL: NEVER kill NX processes directly .
Uses NXSessionManager for safe lock cleanup only.
"""
if self . _solver and hasattr ( self . _solver , ' session_manager ' ) :
if self . _solver . :
logger . info ( " Closing NX session via session manager " )
if self . _solver is not None :
sm = getattr ( self . _solver , " " , None )
if sm is not None :
logger . info ( " Cleaning up NX session locks via session manager " )
try :
self . _solver . session_manager . cleanup_stale_locks ( )
sm . cleanup_stale_locks ( )
except Exception as e :
logger . warning ( f " Session cleanup warning: { e } " )
logger . warning ( " Session lock cleanup warning: %s " , e )
self . _solver = None
logger . info ( " NXOpenSolver closed. " )
# ------------------------------------------------------------------
# Private helpers
# ------------------------------------------------------------------
def _init_solver ( self ) - > None :
""" Lazy-initialize NXSolver (matches SAT3_V7 FEARunner.setup pattern). """
logger . info ( " Initializing NXSolver (nastran= %s , timeout= %d s) " , self . nastran_version , self . timeout )
kwargs : dict = {
" nastran_version " : self . nastran_version ,
" timeout " : self . timeout ,
" use_journal " : True ,
" enable_session_management " : True ,
" study_name " : " hydrotech_beam_doe " ,
" use_iteration_folders " : True ,
" master_model_dir " : str ( self . model_dir ) ,
}
if self . nx_install_dir :
kwargs [ " nx_install_dir " ] = self . nx_install_dir
self . _solver = self . _NXSolver ( * * kwargs )
logger . info ( " NXSolver ready " )
def _build_expression_dict ( self , trial_input : TrialInput ) - > dict [ str , float ] :
""" Build NX expression name→value dict for the solver.
These are passed to:
- create_iteration_folder() → writes params.exp (unit defaulting to mm)
- run_simulation(expression_updates=...) → passed as argv to solve journal
"""
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 expressions are float
}
@staticmethod
def _extract_mass_fallback ( iter_folder : Path ) - > float | None :
""" Try to read mass from _temp_part_properties.json (backup path). """
import json as _json
props_file = iter_folder / " _temp_part_properties.json "
if not props_file . exists ( ) :
return None
try :
with open ( props_file ) as f :
props = _json . load ( f )
mass = props . get ( " mass_kg " , 0.0 )
if mass > 0 :
logger . info ( " Mass from _temp_part_properties.json: %.6f kg " , mass )
return mass
return None
except Exception as e :
logger . warning ( " Failed to read %s : %s " , props_file , e )
return None
# ---------------------------------------------------------------------------
# Factory
@@ -490,24 +595,24 @@ class NXOpenSolver:
def create_solver (
backend : str = " stub " ,
model_dir : str = " " ,
nx_install_dir : str | None = None ,
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).
backend: " stub " for development, " nxopen " for real NX (Windows/dalidou only).
model_dir: Path to NX model directory (required for nxopen backend).
nx_install_dir: Path to NX installation (auto-detected if None).
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 " ).
nastran_version: NX version (e.g., " 2412 " , " 2506 " , " 2512 " ) .
Returns:
Solver instance implementing the NXSolverInterface protocol.
Raises:
ValueError: If backend is unknown.
ValueError: If backend is unknown or model_dir missing for nxopen .
"""
if backend == " stub " :
return NXStubSolver ( )
@@ -516,8 +621,8 @@ def create_solver(
raise ValueError ( " model_dir required for nxopen backend " )
return NXOpenSolver (
model_dir = model_dir ,
nx_install_dir = nx_install_dir ,
timeout = timeout ,
use_iteration_folders = use_iteration_folders ,
nastran_version = nastran_version ,
)
else :
