feat: Add MLP surrogate with Turbo Mode for 100x faster optimization

Neural Acceleration (MLP Surrogate):
- Add run_nn_optimization.py with hybrid FEA/NN workflow
- MLP architecture: 4-layer (64->128->128->64) with BatchNorm/Dropout
- Three workflow modes:
  - --all: Sequential export->train->optimize->validate
  - --hybrid-loop: Iterative Train->NN->Validate->Retrain cycle
  - --turbo: Aggressive single-best validation (RECOMMENDED)
- Turbo mode: 5000 NN trials + 50 FEA validations in ~12 minutes
- Separate nn_study.db to avoid overloading dashboard

Performance Results (bracket_pareto_3obj study):
- NN prediction errors: mass 1-5%, stress 1-4%, stiffness 5-15%
- Found minimum mass designs at boundary (angle~30deg, thick~30mm)
- 100x speedup vs pure FEA exploration

Protocol Operating System:
- Add .claude/skills/ with Bootstrap, Cheatsheet, Context Loader
- Add docs/protocols/ with operations (OP_01-06) and system (SYS_10-14)
- Update SYS_14_NEURAL_ACCELERATION.md with MLP Turbo Mode docs

NX Automation:
- Add optimization_engine/hooks/ for NX CAD/CAE automation
- Add study_wizard.py for guided study creation
- Fix FEM mesh update: load idealized part before UpdateFemodel()

New Study:
- bracket_pareto_3obj: 3-objective Pareto (mass, stress, stiffness)
- 167 FEA trials + 5000 NN trials completed
- Demonstrates full hybrid workflow

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Antoine
2025-12-06 20:01:59 -05:00
parent 0cb2808c44
commit 602560c46a
70 changed files with 31018 additions and 289 deletions

View File

@@ -0,0 +1,83 @@
"""
NX CAD Hooks
============
Direct manipulation of NX CAD parts via NX Open Python API.
This submodule contains hooks for CAD-level operations on NX parts:
geometry, expressions, features, and part management.
Modules
-------
part_manager
Open, close, save, and query NX part files.
Functions:
- open_part(path) -> Open an NX part file
- close_part(path) -> Close an open part
- save_part(path) -> Save a part
- save_part_as(path, new_path) -> Save with new name
- get_part_info(path) -> Get part metadata
expression_manager
Get and set NX expressions (design parameters).
Functions:
- get_expressions(path) -> Get all expressions
- get_expression(path, name) -> Get single expression
- set_expression(path, name, value) -> Set single expression
- set_expressions(path, dict) -> Set multiple expressions
geometry_query
Query geometric properties (mass, volume, area, bodies).
Functions:
- get_mass_properties(path) -> Get mass, volume, area, centroid
- get_bodies(path) -> Get body count and types
- get_volume(path) -> Get total volume
- get_surface_area(path) -> Get total surface area
- get_material(path) -> Get material name
feature_manager
Suppress and unsuppress features for design exploration.
Functions:
- get_features(path) -> List all features
- get_feature_status(path, name) -> Check if suppressed
- suppress_feature(path, name) -> Suppress a feature
- unsuppress_feature(path, name) -> Unsuppress a feature
- suppress_features(path, names) -> Suppress multiple
- unsuppress_features(path, names) -> Unsuppress multiple
Example
-------
>>> from optimization_engine.hooks.nx_cad import geometry_query
>>> result = geometry_query.get_mass_properties("C:/model.prt")
>>> if result["success"]:
... print(f"Mass: {result['data']['mass']:.4f} kg")
... print(f"Material: {result['data']['material']}")
NX Open APIs Used
-----------------
- Session.Parts.OpenActiveDisplay() - Open parts
- Part.Close(), Part.Save(), Part.SaveAs() - Part operations
- Part.Expressions, ExpressionCollection.Edit() - Expressions
- MeasureManager.NewMassProperties() - Mass properties
- Part.Bodies - Body collection
- Feature.Suppress(), Feature.Unsuppress() - Feature control
- Session.UpdateManager.DoUpdate() - Model update
"""
from . import part_manager
from . import expression_manager
from . import geometry_query
from . import feature_manager
from . import model_introspection
__all__ = [
'part_manager',
'expression_manager',
'geometry_query',
'feature_manager',
'model_introspection',
]

View File

@@ -0,0 +1,566 @@
"""
NX Expression Manager Hook
===========================
Provides Python functions to get and set NX expressions (parameters).
API Reference (verified via Siemens MCP docs):
- Part.Expressions() -> ExpressionCollection
- ExpressionCollection.Edit(expression, value)
- Expression.Name, Expression.Value, Expression.RightHandSide
- Expression.Units.Name
Usage:
from optimization_engine.hooks.nx_cad import expression_manager
# Get all expressions
result = expression_manager.get_expressions("C:/path/to/part.prt")
# Get specific expression
result = expression_manager.get_expression("C:/path/to/part.prt", "thickness")
# Set expression value
result = expression_manager.set_expression("C:/path/to/part.prt", "thickness", 5.0)
# Set multiple expressions
result = expression_manager.set_expressions("C:/path/to/part.prt", {
"thickness": 5.0,
"width": 10.0
})
"""
import os
import json
import subprocess
import tempfile
from pathlib import Path
from typing import Optional, Dict, Any, List, Tuple, Union
# NX installation path (configurable)
NX_BIN_PATH = os.environ.get(
"NX_BIN_PATH",
r"C:\Program Files\Siemens\NX2506\NXBIN"
)
# Journal template for expression operations
EXPRESSION_OPERATIONS_JOURNAL = '''
# NX Open Python Journal - Expression Operations
# Auto-generated by Atomizer hooks
import NXOpen
import NXOpen.UF
import json
import sys
import os
def main():
"""Execute expression operation based on command arguments."""
# Get the NX session
session = NXOpen.Session.GetSession()
# Parse arguments: operation, part_path, output_json, [extra_args...]
args = sys.argv[1:] if len(sys.argv) > 1 else []
if len(args) < 3:
raise ValueError("Usage: script.py <operation> <part_path> <output_json> [args...]")
operation = args[0]
part_path = args[1]
output_json = args[2]
extra_args = args[3:] if len(args) > 3 else []
result = {"success": False, "error": None, "data": {}}
try:
# Ensure part is open
part = ensure_part_open(session, part_path)
if part is None:
result["error"] = f"Failed to open part: {part_path}"
elif operation == "get_all":
result = get_all_expressions(part)
elif operation == "get":
expr_name = extra_args[0] if extra_args else None
result = get_expression(part, expr_name)
elif operation == "set":
expr_name = extra_args[0] if len(extra_args) > 0 else None
expr_value = extra_args[1] if len(extra_args) > 1 else None
result = set_expression(session, part, expr_name, expr_value)
elif operation == "set_multiple":
# Extra args is a JSON string with name:value pairs
expr_dict = json.loads(extra_args[0]) if extra_args else {}
result = set_multiple_expressions(session, part, expr_dict)
else:
result["error"] = f"Unknown operation: {operation}"
except Exception as e:
import traceback
result["error"] = str(e)
result["traceback"] = traceback.format_exc()
# Write result to output JSON
with open(output_json, 'w') as f:
json.dump(result, f, indent=2)
return result
def ensure_part_open(session, part_path):
"""Ensure the part is open and return it."""
# Check if already open
part_path_normalized = os.path.normpath(part_path).lower()
for part in session.Parts:
if os.path.normpath(part.FullPath).lower() == part_path_normalized:
return part
# Need to open it
if not os.path.exists(part_path):
return None
try:
# Set load options for the working directory
working_dir = os.path.dirname(part_path)
session.Parts.LoadOptions.ComponentLoadMethod = NXOpen.LoadOptions.LoadMethod.FromDirectory
session.Parts.LoadOptions.SetSearchDirectories([working_dir], [True])
# Use OpenActiveDisplay instead of OpenBase for better compatibility
part, load_status = session.Parts.OpenActiveDisplay(
part_path,
NXOpen.DisplayPartOption.AllowAdditional
)
load_status.Dispose()
return part
except:
return None
def get_all_expressions(part):
"""Get all expressions from a part.
NX Open API: Part.Expressions() -> ExpressionCollection
"""
result = {"success": False, "error": None, "data": {}}
try:
expressions = {}
for expr in part.Expressions:
try:
expr_data = {
"name": expr.Name,
"value": expr.Value,
"rhs": expr.RightHandSide,
"units": expr.Units.Name if expr.Units else None,
"type": expr.Type.ToString() if hasattr(expr.Type, 'ToString') else str(expr.Type),
}
expressions[expr.Name] = expr_data
except:
# Skip expressions that can't be read
pass
result["success"] = True
result["data"] = {
"count": len(expressions),
"expressions": expressions
}
except Exception as e:
result["error"] = str(e)
return result
def get_expression(part, expr_name):
"""Get a specific expression by name.
NX Open API: ExpressionCollection iteration, Expression properties
"""
result = {"success": False, "error": None, "data": {}}
if not expr_name:
result["error"] = "Expression name is required"
return result
try:
# Find the expression by name
found_expr = None
for expr in part.Expressions:
if expr.Name == expr_name:
found_expr = expr
break
if found_expr is None:
result["error"] = f"Expression not found: {expr_name}"
return result
result["success"] = True
result["data"] = {
"name": found_expr.Name,
"value": found_expr.Value,
"rhs": found_expr.RightHandSide,
"units": found_expr.Units.Name if found_expr.Units else None,
"type": found_expr.Type.ToString() if hasattr(found_expr.Type, 'ToString') else str(found_expr.Type),
}
except Exception as e:
result["error"] = str(e)
return result
def set_expression(session, part, expr_name, expr_value):
"""Set an expression value.
NX Open API: ExpressionCollection.Edit(expression, new_rhs)
"""
result = {"success": False, "error": None, "data": {}}
if not expr_name:
result["error"] = "Expression name is required"
return result
if expr_value is None:
result["error"] = "Expression value is required"
return result
try:
# Find the expression
found_expr = None
for expr in part.Expressions:
if expr.Name == expr_name:
found_expr = expr
break
if found_expr is None:
result["error"] = f"Expression not found: {expr_name}"
return result
# Set undo mark
mark_id = session.SetUndoMark(NXOpen.Session.MarkVisibility.Visible, "Edit Expression")
# Edit the expression
# The value should be a string for the RHS
new_rhs = str(expr_value)
part.Expressions.Edit(found_expr, new_rhs)
# Update the model
session.UpdateManager.DoUpdate(mark_id)
result["success"] = True
result["data"] = {
"name": expr_name,
"old_value": found_expr.Value, # Note: this might be the new value after edit
"new_rhs": new_rhs,
}
except Exception as e:
result["error"] = str(e)
return result
def set_multiple_expressions(session, part, expr_dict):
"""Set multiple expressions at once.
Args:
session: NX session
part: NX part
expr_dict: Dict of expression name -> value
"""
result = {"success": False, "error": None, "data": {}}
if not expr_dict:
result["error"] = "No expressions provided"
return result
try:
# Set undo mark for all changes
mark_id = session.SetUndoMark(NXOpen.Session.MarkVisibility.Visible, "Edit Multiple Expressions")
updated = []
errors = []
for expr_name, expr_value in expr_dict.items():
# Find the expression
found_expr = None
for expr in part.Expressions:
if expr.Name == expr_name:
found_expr = expr
break
if found_expr is None:
errors.append(f"Expression not found: {expr_name}")
continue
try:
# Edit the expression
new_rhs = str(expr_value)
part.Expressions.Edit(found_expr, new_rhs)
updated.append({"name": expr_name, "value": expr_value})
except Exception as e:
errors.append(f"Failed to set {expr_name}: {str(e)}")
# Update the model
session.UpdateManager.DoUpdate(mark_id)
result["success"] = len(errors) == 0
result["data"] = {
"updated": updated,
"errors": errors,
"update_count": len(updated),
"error_count": len(errors),
}
except Exception as e:
result["error"] = str(e)
return result
if __name__ == "__main__":
main()
'''
def _get_run_journal_exe() -> str:
"""Get the path to run_journal.exe."""
return os.path.join(NX_BIN_PATH, "run_journal.exe")
def _run_journal(journal_path: str, args: list) -> Tuple[bool, str]:
"""Run an NX journal with arguments.
Returns:
Tuple of (success, output_or_error)
"""
run_journal = _get_run_journal_exe()
if not os.path.exists(run_journal):
return False, f"run_journal.exe not found at {run_journal}"
cmd = [run_journal, journal_path, "-args"] + args
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=120 # 2 minute timeout
)
if result.returncode != 0:
return False, f"Journal execution failed: {result.stderr}"
return True, result.stdout
except subprocess.TimeoutExpired:
return False, "Journal execution timed out"
except Exception as e:
return False, str(e)
def _execute_expression_operation(
operation: str,
part_path: str,
extra_args: list = None
) -> Dict[str, Any]:
"""Execute an expression operation via NX journal.
Args:
operation: The operation to perform (get_all, get, set, set_multiple)
part_path: Path to the part file
extra_args: Additional arguments for the operation
Returns:
Dict with operation result
"""
# Create temporary journal file
with tempfile.NamedTemporaryFile(
mode='w',
suffix='.py',
delete=False
) as journal_file:
journal_file.write(EXPRESSION_OPERATIONS_JOURNAL)
journal_path = journal_file.name
# Create temporary output file
output_file = tempfile.NamedTemporaryFile(
mode='w',
suffix='.json',
delete=False
).name
try:
# Build arguments
args = [operation, part_path, output_file]
if extra_args:
args.extend(extra_args)
# Run the journal
success, output = _run_journal(journal_path, args)
if not success:
return {"success": False, "error": output, "data": {}}
# Read the result
if os.path.exists(output_file):
with open(output_file, 'r') as f:
return json.load(f)
else:
return {"success": False, "error": "Output file not created", "data": {}}
finally:
# Cleanup temporary files
if os.path.exists(journal_path):
os.unlink(journal_path)
if os.path.exists(output_file):
os.unlink(output_file)
# =============================================================================
# Public API
# =============================================================================
def get_expressions(part_path: str) -> Dict[str, Any]:
"""Get all expressions from an NX part.
Args:
part_path: Full path to the .prt file
Returns:
Dict with keys:
- success: bool
- error: Optional error message
- data: Dict with count and expressions dict
Each expression has: name, value, rhs, units, type
Example:
>>> result = get_expressions("C:/models/bracket.prt")
>>> if result["success"]:
... for name, expr in result["data"]["expressions"].items():
... print(f"{name} = {expr['value']} {expr['units']}")
"""
part_path = os.path.abspath(part_path)
if not os.path.exists(part_path):
return {
"success": False,
"error": f"Part file not found: {part_path}",
"data": {}
}
return _execute_expression_operation("get_all", part_path)
def get_expression(part_path: str, expression_name: str) -> Dict[str, Any]:
"""Get a specific expression from an NX part.
Args:
part_path: Full path to the .prt file
expression_name: Name of the expression
Returns:
Dict with keys:
- success: bool
- error: Optional error message
- data: Dict with name, value, rhs, units, type
Example:
>>> result = get_expression("C:/models/bracket.prt", "thickness")
>>> if result["success"]:
... print(f"thickness = {result['data']['value']}")
"""
part_path = os.path.abspath(part_path)
if not os.path.exists(part_path):
return {
"success": False,
"error": f"Part file not found: {part_path}",
"data": {}
}
return _execute_expression_operation("get", part_path, [expression_name])
def set_expression(
part_path: str,
expression_name: str,
value: Union[float, int, str]
) -> Dict[str, Any]:
"""Set an expression value in an NX part.
Args:
part_path: Full path to the .prt file
expression_name: Name of the expression
value: New value (will be converted to string for RHS)
Returns:
Dict with keys:
- success: bool
- error: Optional error message
- data: Dict with name, old_value, new_rhs
Example:
>>> result = set_expression("C:/models/bracket.prt", "thickness", 5.0)
>>> if result["success"]:
... print("Expression updated!")
"""
part_path = os.path.abspath(part_path)
if not os.path.exists(part_path):
return {
"success": False,
"error": f"Part file not found: {part_path}",
"data": {}
}
return _execute_expression_operation(
"set",
part_path,
[expression_name, str(value)]
)
def set_expressions(
part_path: str,
expressions: Dict[str, Union[float, int, str]]
) -> Dict[str, Any]:
"""Set multiple expressions in an NX part.
Args:
part_path: Full path to the .prt file
expressions: Dict mapping expression names to values
Returns:
Dict with keys:
- success: bool
- error: Optional error message
- data: Dict with updated list, errors list, counts
Example:
>>> result = set_expressions("C:/models/bracket.prt", {
... "thickness": 5.0,
... "width": 10.0,
... "height": 15.0
... })
>>> if result["success"]:
... print(f"Updated {result['data']['update_count']} expressions")
"""
part_path = os.path.abspath(part_path)
if not os.path.exists(part_path):
return {
"success": False,
"error": f"Part file not found: {part_path}",
"data": {}
}
# Convert expressions dict to JSON string
expr_json = json.dumps(expressions)
return _execute_expression_operation(
"set_multiple",
part_path,
[expr_json]
)

View File

@@ -0,0 +1,711 @@
"""
NX Feature Manager Hook
=======================
Provides Python functions to manage NX features (suppress, unsuppress, etc.).
API Reference (verified via Siemens MCP docs):
- Part.Features() -> FeatureCollection
- Feature.Suppress() -> Suppresses the feature
- Feature.Unsuppress() -> Unsuppresses the feature
- Feature.Name, Feature.IsSuppressed
- Session.UpdateManager.DoUpdate() -> Update the model
Usage:
from optimization_engine.hooks.nx_cad import feature_manager
# Get all features
result = feature_manager.get_features("C:/path/to/part.prt")
# Suppress a feature
result = feature_manager.suppress_feature("C:/path/to/part.prt", "HOLE(1)")
# Unsuppress a feature
result = feature_manager.unsuppress_feature("C:/path/to/part.prt", "HOLE(1)")
# Get feature status
result = feature_manager.get_feature_status("C:/path/to/part.prt", "HOLE(1)")
"""
import os
import json
import subprocess
import tempfile
from pathlib import Path
from typing import Optional, Dict, Any, List, Tuple
# NX installation path (configurable)
NX_BIN_PATH = os.environ.get(
"NX_BIN_PATH",
r"C:\Program Files\Siemens\NX2506\NXBIN"
)
# Journal template for feature operations
FEATURE_OPERATIONS_JOURNAL = '''
# NX Open Python Journal - Feature Operations
# Auto-generated by Atomizer hooks
#
# Based on Siemens NX Open Python API:
# - Part.Features()
# - Feature.Suppress() / Feature.Unsuppress()
# - Feature.Name, Feature.IsSuppressed
import NXOpen
import NXOpen.Features
import json
import sys
import os
def main():
"""Execute feature operation based on command arguments."""
# Get the NX session
session = NXOpen.Session.GetSession()
# Parse arguments: operation, part_path, output_json, [extra_args...]
args = sys.argv[1:] if len(sys.argv) > 1 else []
if len(args) < 3:
raise ValueError("Usage: script.py <operation> <part_path> <output_json> [args...]")
operation = args[0]
part_path = args[1]
output_json = args[2]
extra_args = args[3:] if len(args) > 3 else []
result = {"success": False, "error": None, "data": {}}
try:
# Ensure part is open
part = ensure_part_open(session, part_path)
if part is None:
result["error"] = f"Failed to open part: {part_path}"
elif operation == "get_all":
result = get_all_features(part)
elif operation == "get_status":
feature_name = extra_args[0] if extra_args else None
result = get_feature_status(part, feature_name)
elif operation == "suppress":
feature_name = extra_args[0] if extra_args else None
result = suppress_feature(session, part, feature_name)
elif operation == "unsuppress":
feature_name = extra_args[0] if extra_args else None
result = unsuppress_feature(session, part, feature_name)
elif operation == "suppress_multiple":
feature_names = json.loads(extra_args[0]) if extra_args else []
result = suppress_multiple_features(session, part, feature_names)
elif operation == "unsuppress_multiple":
feature_names = json.loads(extra_args[0]) if extra_args else []
result = unsuppress_multiple_features(session, part, feature_names)
else:
result["error"] = f"Unknown operation: {operation}"
except Exception as e:
import traceback
result["error"] = str(e)
result["traceback"] = traceback.format_exc()
# Write result to output JSON
with open(output_json, 'w') as f:
json.dump(result, f, indent=2)
return result
def ensure_part_open(session, part_path):
"""Ensure the part is open and return it."""
# Check if already open
part_path_normalized = os.path.normpath(part_path).lower()
for part in session.Parts:
if os.path.normpath(part.FullPath).lower() == part_path_normalized:
return part
# Need to open it
if not os.path.exists(part_path):
return None
try:
# Set load options for the working directory
working_dir = os.path.dirname(part_path)
session.Parts.LoadOptions.ComponentLoadMethod = NXOpen.LoadOptions.LoadMethod.FromDirectory
session.Parts.LoadOptions.SetSearchDirectories([working_dir], [True])
# Use OpenActiveDisplay instead of OpenBase for better compatibility
part, load_status = session.Parts.OpenActiveDisplay(
part_path,
NXOpen.DisplayPartOption.AllowAdditional
)
load_status.Dispose()
return part
except:
return None
def find_feature_by_name(part, feature_name):
"""Find a feature by name."""
for feature in part.Features:
if feature.Name == feature_name:
return feature
return None
def get_all_features(part):
"""Get all features from a part.
NX Open API: Part.Features()
"""
result = {"success": False, "error": None, "data": {}}
try:
features = []
for feature in part.Features:
try:
feature_data = {
"name": feature.Name,
"type": feature.FeatureType,
"is_suppressed": feature.IsSuppressed,
"is_internal": feature.IsInternal,
}
features.append(feature_data)
except:
# Skip features that can't be read
pass
result["success"] = True
result["data"] = {
"count": len(features),
"suppressed_count": sum(1 for f in features if f["is_suppressed"]),
"features": features
}
except Exception as e:
result["error"] = str(e)
return result
def get_feature_status(part, feature_name):
"""Get status of a specific feature.
NX Open API: Feature properties
"""
result = {"success": False, "error": None, "data": {}}
if not feature_name:
result["error"] = "Feature name is required"
return result
try:
feature = find_feature_by_name(part, feature_name)
if feature is None:
result["error"] = f"Feature not found: {feature_name}"
return result
result["success"] = True
result["data"] = {
"name": feature.Name,
"type": feature.FeatureType,
"is_suppressed": feature.IsSuppressed,
"is_internal": feature.IsInternal,
}
except Exception as e:
result["error"] = str(e)
return result
def suppress_feature(session, part, feature_name):
"""Suppress a feature.
NX Open API: Feature.Suppress()
"""
result = {"success": False, "error": None, "data": {}}
if not feature_name:
result["error"] = "Feature name is required"
return result
try:
feature = find_feature_by_name(part, feature_name)
if feature is None:
result["error"] = f"Feature not found: {feature_name}"
return result
if feature.IsSuppressed:
result["success"] = True
result["data"] = {
"name": feature_name,
"action": "already_suppressed",
"is_suppressed": True
}
return result
# Set undo mark
mark_id = session.SetUndoMark(NXOpen.Session.MarkVisibility.Visible, "Suppress Feature")
# Suppress the feature
feature.Suppress()
# Update the model
session.UpdateManager.DoUpdate(mark_id)
result["success"] = True
result["data"] = {
"name": feature_name,
"action": "suppressed",
"is_suppressed": True
}
except Exception as e:
result["error"] = str(e)
return result
def unsuppress_feature(session, part, feature_name):
"""Unsuppress a feature.
NX Open API: Feature.Unsuppress()
"""
result = {"success": False, "error": None, "data": {}}
if not feature_name:
result["error"] = "Feature name is required"
return result
try:
feature = find_feature_by_name(part, feature_name)
if feature is None:
result["error"] = f"Feature not found: {feature_name}"
return result
if not feature.IsSuppressed:
result["success"] = True
result["data"] = {
"name": feature_name,
"action": "already_unsuppressed",
"is_suppressed": False
}
return result
# Set undo mark
mark_id = session.SetUndoMark(NXOpen.Session.MarkVisibility.Visible, "Unsuppress Feature")
# Unsuppress the feature
feature.Unsuppress()
# Update the model
session.UpdateManager.DoUpdate(mark_id)
result["success"] = True
result["data"] = {
"name": feature_name,
"action": "unsuppressed",
"is_suppressed": False
}
except Exception as e:
result["error"] = str(e)
return result
def suppress_multiple_features(session, part, feature_names):
"""Suppress multiple features.
Args:
session: NX session
part: NX part
feature_names: List of feature names to suppress
"""
result = {"success": False, "error": None, "data": {}}
if not feature_names:
result["error"] = "No feature names provided"
return result
try:
# Set undo mark for all changes
mark_id = session.SetUndoMark(NXOpen.Session.MarkVisibility.Visible, "Suppress Multiple Features")
suppressed = []
errors = []
for feature_name in feature_names:
feature = find_feature_by_name(part, feature_name)
if feature is None:
errors.append(f"Feature not found: {feature_name}")
continue
try:
if not feature.IsSuppressed:
feature.Suppress()
suppressed.append(feature_name)
except Exception as e:
errors.append(f"Failed to suppress {feature_name}: {str(e)}")
# Update the model
session.UpdateManager.DoUpdate(mark_id)
result["success"] = len(errors) == 0
result["data"] = {
"suppressed": suppressed,
"errors": errors,
"suppressed_count": len(suppressed),
"error_count": len(errors),
}
except Exception as e:
result["error"] = str(e)
return result
def unsuppress_multiple_features(session, part, feature_names):
"""Unsuppress multiple features.
Args:
session: NX session
part: NX part
feature_names: List of feature names to unsuppress
"""
result = {"success": False, "error": None, "data": {}}
if not feature_names:
result["error"] = "No feature names provided"
return result
try:
# Set undo mark for all changes
mark_id = session.SetUndoMark(NXOpen.Session.MarkVisibility.Visible, "Unsuppress Multiple Features")
unsuppressed = []
errors = []
for feature_name in feature_names:
feature = find_feature_by_name(part, feature_name)
if feature is None:
errors.append(f"Feature not found: {feature_name}")
continue
try:
if feature.IsSuppressed:
feature.Unsuppress()
unsuppressed.append(feature_name)
except Exception as e:
errors.append(f"Failed to unsuppress {feature_name}: {str(e)}")
# Update the model
session.UpdateManager.DoUpdate(mark_id)
result["success"] = len(errors) == 0
result["data"] = {
"unsuppressed": unsuppressed,
"errors": errors,
"unsuppressed_count": len(unsuppressed),
"error_count": len(errors),
}
except Exception as e:
result["error"] = str(e)
return result
if __name__ == "__main__":
main()
'''
def _get_run_journal_exe() -> str:
"""Get the path to run_journal.exe."""
return os.path.join(NX_BIN_PATH, "run_journal.exe")
def _run_journal(journal_path: str, args: list) -> Tuple[bool, str]:
"""Run an NX journal with arguments.
Returns:
Tuple of (success, output_or_error)
"""
run_journal = _get_run_journal_exe()
if not os.path.exists(run_journal):
return False, f"run_journal.exe not found at {run_journal}"
cmd = [run_journal, journal_path, "-args"] + args
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=120 # 2 minute timeout
)
if result.returncode != 0:
return False, f"Journal execution failed: {result.stderr}"
return True, result.stdout
except subprocess.TimeoutExpired:
return False, "Journal execution timed out"
except Exception as e:
return False, str(e)
def _execute_feature_operation(
operation: str,
part_path: str,
extra_args: list = None
) -> Dict[str, Any]:
"""Execute a feature operation via NX journal.
Args:
operation: The operation to perform
part_path: Path to the part file
extra_args: Additional arguments for the operation
Returns:
Dict with operation result
"""
# Create temporary journal file
with tempfile.NamedTemporaryFile(
mode='w',
suffix='.py',
delete=False
) as journal_file:
journal_file.write(FEATURE_OPERATIONS_JOURNAL)
journal_path = journal_file.name
# Create temporary output file
output_file = tempfile.NamedTemporaryFile(
mode='w',
suffix='.json',
delete=False
).name
try:
# Build arguments
args = [operation, part_path, output_file]
if extra_args:
args.extend(extra_args)
# Run the journal
success, output = _run_journal(journal_path, args)
if not success:
return {"success": False, "error": output, "data": {}}
# Read the result
if os.path.exists(output_file):
with open(output_file, 'r') as f:
return json.load(f)
else:
return {"success": False, "error": "Output file not created", "data": {}}
finally:
# Cleanup temporary files
if os.path.exists(journal_path):
os.unlink(journal_path)
if os.path.exists(output_file):
os.unlink(output_file)
# =============================================================================
# Public API
# =============================================================================
def get_features(part_path: str) -> Dict[str, Any]:
"""Get all features from an NX part.
Args:
part_path: Full path to the .prt file
Returns:
Dict with keys:
- success: bool
- error: Optional error message
- data: Dict with count, suppressed_count, features list
Example:
>>> result = get_features("C:/models/bracket.prt")
>>> if result["success"]:
... for f in result["data"]["features"]:
... status = "suppressed" if f["is_suppressed"] else "active"
... print(f"{f['name']} ({f['type']}): {status}")
"""
part_path = os.path.abspath(part_path)
if not os.path.exists(part_path):
return {
"success": False,
"error": f"Part file not found: {part_path}",
"data": {}
}
return _execute_feature_operation("get_all", part_path)
def get_feature_status(part_path: str, feature_name: str) -> Dict[str, Any]:
"""Get status of a specific feature.
Args:
part_path: Full path to the .prt file
feature_name: Name of the feature
Returns:
Dict with keys:
- success: bool
- error: Optional error message
- data: Dict with name, type, is_suppressed, is_internal
Example:
>>> result = get_feature_status("C:/models/bracket.prt", "HOLE(1)")
>>> if result["success"]:
... print(f"Suppressed: {result['data']['is_suppressed']}")
"""
part_path = os.path.abspath(part_path)
if not os.path.exists(part_path):
return {
"success": False,
"error": f"Part file not found: {part_path}",
"data": {}
}
return _execute_feature_operation("get_status", part_path, [feature_name])
def suppress_feature(part_path: str, feature_name: str) -> Dict[str, Any]:
"""Suppress a feature in an NX part.
Args:
part_path: Full path to the .prt file
feature_name: Name of the feature to suppress
Returns:
Dict with keys:
- success: bool
- error: Optional error message
- data: Dict with name, action, is_suppressed
Example:
>>> result = suppress_feature("C:/models/bracket.prt", "HOLE(1)")
>>> if result["success"]:
... print(f"Feature {result['data']['action']}")
"""
part_path = os.path.abspath(part_path)
if not os.path.exists(part_path):
return {
"success": False,
"error": f"Part file not found: {part_path}",
"data": {}
}
return _execute_feature_operation("suppress", part_path, [feature_name])
def unsuppress_feature(part_path: str, feature_name: str) -> Dict[str, Any]:
"""Unsuppress a feature in an NX part.
Args:
part_path: Full path to the .prt file
feature_name: Name of the feature to unsuppress
Returns:
Dict with keys:
- success: bool
- error: Optional error message
- data: Dict with name, action, is_suppressed
Example:
>>> result = unsuppress_feature("C:/models/bracket.prt", "HOLE(1)")
>>> if result["success"]:
... print(f"Feature {result['data']['action']}")
"""
part_path = os.path.abspath(part_path)
if not os.path.exists(part_path):
return {
"success": False,
"error": f"Part file not found: {part_path}",
"data": {}
}
return _execute_feature_operation("unsuppress", part_path, [feature_name])
def suppress_features(part_path: str, feature_names: List[str]) -> Dict[str, Any]:
"""Suppress multiple features in an NX part.
Args:
part_path: Full path to the .prt file
feature_names: List of feature names to suppress
Returns:
Dict with keys:
- success: bool
- error: Optional error message
- data: Dict with suppressed list, errors list, counts
Example:
>>> result = suppress_features("C:/models/bracket.prt", ["HOLE(1)", "HOLE(2)"])
>>> if result["success"]:
... print(f"Suppressed {result['data']['suppressed_count']} features")
"""
part_path = os.path.abspath(part_path)
if not os.path.exists(part_path):
return {
"success": False,
"error": f"Part file not found: {part_path}",
"data": {}
}
# Convert list to JSON string
names_json = json.dumps(feature_names)
return _execute_feature_operation("suppress_multiple", part_path, [names_json])
def unsuppress_features(part_path: str, feature_names: List[str]) -> Dict[str, Any]:
"""Unsuppress multiple features in an NX part.
Args:
part_path: Full path to the .prt file
feature_names: List of feature names to unsuppress
Returns:
Dict with keys:
- success: bool
- error: Optional error message
- data: Dict with unsuppressed list, errors list, counts
Example:
>>> result = unsuppress_features("C:/models/bracket.prt", ["HOLE(1)", "HOLE(2)"])
>>> if result["success"]:
... print(f"Unsuppressed {result['data']['unsuppressed_count']} features")
"""
part_path = os.path.abspath(part_path)
if not os.path.exists(part_path):
return {
"success": False,
"error": f"Part file not found: {part_path}",
"data": {}
}
# Convert list to JSON string
names_json = json.dumps(feature_names)
return _execute_feature_operation("unsuppress_multiple", part_path, [names_json])

View File

@@ -0,0 +1,667 @@
"""
NX Geometry Query Hook
======================
Provides Python functions to query geometry properties from NX parts.
API Reference (verified via Siemens MCP docs):
- Part.MeasureManager() -> Returns measure manager for this part
- MeasureManager.NewMassProperties() -> Create mass properties measurement
- Part.Bodies() -> BodyCollection (solid bodies in the part)
- Body.GetPhysicalMaterial() -> Get material assigned to body
Usage:
from optimization_engine.hooks.nx_cad import geometry_query
# Get mass properties
result = geometry_query.get_mass_properties("C:/path/to/part.prt")
# Get body info
result = geometry_query.get_bodies("C:/path/to/part.prt")
# Get volume
result = geometry_query.get_volume("C:/path/to/part.prt")
"""
import os
import json
import subprocess
import tempfile
from pathlib import Path
from typing import Optional, Dict, Any, List, Tuple
# NX installation path (configurable)
NX_BIN_PATH = os.environ.get(
"NX_BIN_PATH",
r"C:\Program Files\Siemens\NX2506\NXBIN"
)
# Journal template for geometry query operations
GEOMETRY_QUERY_JOURNAL = '''
# NX Open Python Journal - Geometry Query Operations
# Auto-generated by Atomizer hooks
#
# Based on Siemens NX Open Python API:
# - MeasureManager.NewMassProperties()
# - BodyCollection
# - Body.GetPhysicalMaterial()
import NXOpen
import NXOpen.UF
import json
import sys
import os
import math
def main():
"""Execute geometry query operation based on command arguments."""
# Get the NX session
session = NXOpen.Session.GetSession()
# Parse arguments: operation, part_path, output_json, [extra_args...]
args = sys.argv[1:] if len(sys.argv) > 1 else []
if len(args) < 3:
raise ValueError("Usage: script.py <operation> <part_path> <output_json> [args...]")
operation = args[0]
part_path = args[1]
output_json = args[2]
extra_args = args[3:] if len(args) > 3 else []
result = {"success": False, "error": None, "data": {}}
try:
# Ensure part is open
part = ensure_part_open(session, part_path)
if part is None:
result["error"] = f"Failed to open part: {part_path}"
elif operation == "mass_properties":
result = get_mass_properties(part)
elif operation == "bodies":
result = get_bodies(part)
elif operation == "volume":
result = get_volume(part)
elif operation == "surface_area":
result = get_surface_area(part)
elif operation == "material":
result = get_material(part)
else:
result["error"] = f"Unknown operation: {operation}"
except Exception as e:
import traceback
result["error"] = str(e)
result["traceback"] = traceback.format_exc()
# Write result to output JSON
with open(output_json, 'w') as f:
json.dump(result, f, indent=2)
return result
def ensure_part_open(session, part_path):
"""Ensure the part is open and return it."""
# Check if already open
part_path_normalized = os.path.normpath(part_path).lower()
for part in session.Parts:
if os.path.normpath(part.FullPath).lower() == part_path_normalized:
return part
# Need to open it
if not os.path.exists(part_path):
return None
try:
# Set load options for the working directory
working_dir = os.path.dirname(part_path)
session.Parts.LoadOptions.ComponentLoadMethod = NXOpen.LoadOptions.LoadMethod.FromDirectory
session.Parts.LoadOptions.SetSearchDirectories([working_dir], [True])
# Use OpenActiveDisplay instead of OpenBase for better compatibility
part, load_status = session.Parts.OpenActiveDisplay(
part_path,
NXOpen.DisplayPartOption.AllowAdditional
)
load_status.Dispose()
return part
except:
return None
def get_solid_bodies(part):
"""Get all solid bodies from a part."""
solid_bodies = []
for body in part.Bodies:
if body.IsSolidBody:
solid_bodies.append(body)
return solid_bodies
def get_mass_properties(part):
"""Get mass properties from a part.
NX Open API: MeasureManager.NewMassProperties()
Returns mass, volume, surface area, centroid, and inertia properties.
"""
result = {"success": False, "error": None, "data": {}}
try:
# Get solid bodies
solid_bodies = get_solid_bodies(part)
if not solid_bodies:
result["error"] = "No solid bodies found in part"
return result
# Get measure manager
measure_manager = part.MeasureManager
# Get units - use base units array like the working journal
uc = part.UnitCollection
mass_units = [
uc.GetBase("Area"),
uc.GetBase("Volume"),
uc.GetBase("Mass"),
uc.GetBase("Length")
]
# Create mass properties measurement
# Signature: NewMassProperties(mass_units, accuracy, objects)
mass_props = measure_manager.NewMassProperties(mass_units, 0.99, solid_bodies)
# Get properties
mass = mass_props.Mass
volume = mass_props.Volume
area = mass_props.Area
# Get centroid
centroid = mass_props.Centroid
centroid_x = centroid.X
centroid_y = centroid.Y
centroid_z = centroid.Z
# Get principal moments of inertia (may not be available)
ixx = 0.0
iyy = 0.0
izz = 0.0
try:
principal_moments = mass_props.PrincipalMomentsOfInertia
ixx = principal_moments[0]
iyy = principal_moments[1]
izz = principal_moments[2]
except:
pass
# Get material info from first body via attributes
material_name = None
density = None
try:
# Try body attributes (NX stores material as attribute)
attrs = solid_bodies[0].GetUserAttributes()
for attr in attrs:
if 'material' in attr.Title.lower():
material_name = attr.StringValue
break
except:
pass
result["success"] = True
result["data"] = {
"mass": mass,
"mass_unit": "kg",
"volume": volume,
"volume_unit": "mm^3",
"surface_area": area,
"area_unit": "mm^2",
"centroid": {
"x": centroid_x,
"y": centroid_y,
"z": centroid_z,
"unit": "mm"
},
"principal_moments": {
"Ixx": ixx,
"Iyy": iyy,
"Izz": izz,
"unit": "kg*mm^2"
},
"material": material_name,
"density": density,
"body_count": len(solid_bodies)
}
except Exception as e:
import traceback
result["error"] = str(e)
result["traceback"] = traceback.format_exc()
return result
def get_bodies(part):
"""Get information about all bodies in the part.
NX Open API: Part.Bodies()
"""
result = {"success": False, "error": None, "data": {}}
try:
bodies_info = []
for body in part.Bodies:
body_data = {
"name": body.Name if hasattr(body, 'Name') else None,
"is_solid": body.IsSolidBody,
"is_sheet": body.IsSheetBody,
}
# Try to get material
try:
phys_mat = body.GetPhysicalMaterial()
if phys_mat:
body_data["material"] = phys_mat.Name
except:
body_data["material"] = None
bodies_info.append(body_data)
result["success"] = True
result["data"] = {
"count": len(bodies_info),
"solid_count": sum(1 for b in bodies_info if b["is_solid"]),
"sheet_count": sum(1 for b in bodies_info if b["is_sheet"]),
"bodies": bodies_info
}
except Exception as e:
result["error"] = str(e)
return result
def get_volume(part):
"""Get total volume of all solid bodies.
NX Open API: MeasureManager.NewMassProperties()
"""
result = {"success": False, "error": None, "data": {}}
try:
solid_bodies = get_solid_bodies(part)
if not solid_bodies:
result["error"] = "No solid bodies found in part"
return result
measure_manager = part.MeasureManager
units = part.UnitCollection
body_array = [NXOpen.IBody.Wrap(body) for body in solid_bodies]
mass_props = measure_manager.NewMassProperties(
units.FindObject("SquareMilliMeter"),
units.FindObject("CubicMillimeter"),
units.FindObject("Kilogram"),
body_array
)
result["success"] = True
result["data"] = {
"volume": mass_props.Volume,
"unit": "mm^3",
"body_count": len(solid_bodies)
}
except Exception as e:
result["error"] = str(e)
return result
def get_surface_area(part):
"""Get total surface area of all solid bodies.
NX Open API: MeasureManager.NewMassProperties()
"""
result = {"success": False, "error": None, "data": {}}
try:
solid_bodies = get_solid_bodies(part)
if not solid_bodies:
result["error"] = "No solid bodies found in part"
return result
measure_manager = part.MeasureManager
units = part.UnitCollection
body_array = [NXOpen.IBody.Wrap(body) for body in solid_bodies]
mass_props = measure_manager.NewMassProperties(
units.FindObject("SquareMilliMeter"),
units.FindObject("CubicMillimeter"),
units.FindObject("Kilogram"),
body_array
)
result["success"] = True
result["data"] = {
"surface_area": mass_props.Area,
"unit": "mm^2",
"body_count": len(solid_bodies)
}
except Exception as e:
result["error"] = str(e)
return result
def get_material(part):
"""Get material information from bodies in the part.
NX Open API: Body.GetPhysicalMaterial()
"""
result = {"success": False, "error": None, "data": {}}
try:
solid_bodies = get_solid_bodies(part)
if not solid_bodies:
result["error"] = "No solid bodies found in part"
return result
materials = {}
for body in solid_bodies:
try:
phys_mat = body.GetPhysicalMaterial()
if phys_mat:
mat_name = phys_mat.Name
if mat_name not in materials:
mat_data = {"name": mat_name}
# Try to get properties
try:
mat_data["density"] = phys_mat.GetRealPropertyValue("Density")
except:
pass
try:
mat_data["youngs_modulus"] = phys_mat.GetRealPropertyValue("YoungsModulus")
except:
pass
try:
mat_data["poissons_ratio"] = phys_mat.GetRealPropertyValue("PoissonsRatio")
except:
pass
materials[mat_name] = mat_data
except:
pass
result["success"] = True
result["data"] = {
"material_count": len(materials),
"materials": materials,
"body_count": len(solid_bodies)
}
except Exception as e:
result["error"] = str(e)
return result
if __name__ == "__main__":
main()
'''
def _get_run_journal_exe() -> str:
"""Get the path to run_journal.exe."""
return os.path.join(NX_BIN_PATH, "run_journal.exe")
def _run_journal(journal_path: str, args: list) -> Tuple[bool, str]:
"""Run an NX journal with arguments.
Returns:
Tuple of (success, output_or_error)
"""
run_journal = _get_run_journal_exe()
if not os.path.exists(run_journal):
return False, f"run_journal.exe not found at {run_journal}"
cmd = [run_journal, journal_path, "-args"] + args
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=120 # 2 minute timeout
)
if result.returncode != 0:
return False, f"Journal execution failed: {result.stderr}"
return True, result.stdout
except subprocess.TimeoutExpired:
return False, "Journal execution timed out"
except Exception as e:
return False, str(e)
def _execute_geometry_operation(
operation: str,
part_path: str,
extra_args: list = None
) -> Dict[str, Any]:
"""Execute a geometry query operation via NX journal.
Args:
operation: The operation to perform
part_path: Path to the part file
extra_args: Additional arguments for the operation
Returns:
Dict with operation result
"""
# Create temporary journal file
with tempfile.NamedTemporaryFile(
mode='w',
suffix='.py',
delete=False
) as journal_file:
journal_file.write(GEOMETRY_QUERY_JOURNAL)
journal_path = journal_file.name
# Create temporary output file
output_file = tempfile.NamedTemporaryFile(
mode='w',
suffix='.json',
delete=False
).name
try:
# Build arguments
args = [operation, part_path, output_file]
if extra_args:
args.extend(extra_args)
# Run the journal
success, output = _run_journal(journal_path, args)
if not success:
return {"success": False, "error": output, "data": {}}
# Read the result
if os.path.exists(output_file):
with open(output_file, 'r') as f:
return json.load(f)
else:
return {"success": False, "error": "Output file not created", "data": {}}
finally:
# Cleanup temporary files
if os.path.exists(journal_path):
os.unlink(journal_path)
if os.path.exists(output_file):
os.unlink(output_file)
# =============================================================================
# Public API
# =============================================================================
def get_mass_properties(part_path: str) -> Dict[str, Any]:
"""Get mass properties from an NX part.
Args:
part_path: Full path to the .prt file
Returns:
Dict with keys:
- success: bool
- error: Optional error message
- data: Dict with mass, volume, surface_area, centroid,
principal_moments, material, density, body_count
Example:
>>> result = get_mass_properties("C:/models/bracket.prt")
>>> if result["success"]:
... print(f"Mass: {result['data']['mass']} kg")
... print(f"Volume: {result['data']['volume']} mm^3")
"""
part_path = os.path.abspath(part_path)
if not os.path.exists(part_path):
return {
"success": False,
"error": f"Part file not found: {part_path}",
"data": {}
}
return _execute_geometry_operation("mass_properties", part_path)
def get_bodies(part_path: str) -> Dict[str, Any]:
"""Get information about bodies in an NX part.
Args:
part_path: Full path to the .prt file
Returns:
Dict with keys:
- success: bool
- error: Optional error message
- data: Dict with count, solid_count, sheet_count, bodies list
Example:
>>> result = get_bodies("C:/models/bracket.prt")
>>> if result["success"]:
... print(f"Solid bodies: {result['data']['solid_count']}")
"""
part_path = os.path.abspath(part_path)
if not os.path.exists(part_path):
return {
"success": False,
"error": f"Part file not found: {part_path}",
"data": {}
}
return _execute_geometry_operation("bodies", part_path)
def get_volume(part_path: str) -> Dict[str, Any]:
"""Get total volume of solid bodies in an NX part.
Args:
part_path: Full path to the .prt file
Returns:
Dict with keys:
- success: bool
- error: Optional error message
- data: Dict with volume (mm^3), unit, body_count
Example:
>>> result = get_volume("C:/models/bracket.prt")
>>> if result["success"]:
... print(f"Volume: {result['data']['volume']} mm^3")
"""
part_path = os.path.abspath(part_path)
if not os.path.exists(part_path):
return {
"success": False,
"error": f"Part file not found: {part_path}",
"data": {}
}
return _execute_geometry_operation("volume", part_path)
def get_surface_area(part_path: str) -> Dict[str, Any]:
"""Get total surface area of solid bodies in an NX part.
Args:
part_path: Full path to the .prt file
Returns:
Dict with keys:
- success: bool
- error: Optional error message
- data: Dict with surface_area (mm^2), unit, body_count
Example:
>>> result = get_surface_area("C:/models/bracket.prt")
>>> if result["success"]:
... print(f"Area: {result['data']['surface_area']} mm^2")
"""
part_path = os.path.abspath(part_path)
if not os.path.exists(part_path):
return {
"success": False,
"error": f"Part file not found: {part_path}",
"data": {}
}
return _execute_geometry_operation("surface_area", part_path)
def get_material(part_path: str) -> Dict[str, Any]:
"""Get material information from bodies in an NX part.
Args:
part_path: Full path to the .prt file
Returns:
Dict with keys:
- success: bool
- error: Optional error message
- data: Dict with material_count, materials dict, body_count
Example:
>>> result = get_material("C:/models/bracket.prt")
>>> if result["success"]:
... for name, mat in result["data"]["materials"].items():
... print(f"{name}: density={mat.get('density')}")
"""
part_path = os.path.abspath(part_path)
if not os.path.exists(part_path):
return {
"success": False,
"error": f"Part file not found: {part_path}",
"data": {}
}
return _execute_geometry_operation("material", part_path)

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,478 @@
"""
NX Part Manager Hook
====================
Provides Python functions to open, close, and save NX parts.
API Reference (verified via Siemens MCP docs):
- Session.Parts() -> PartCollection
- PartCollection.OpenBase() -> Opens a part file
- Part.Close() -> Closes the part
- Part.Save() -> Saves the part
- Part.SaveAs() -> Saves the part with a new name
Usage:
from optimization_engine.hooks.nx_cad import part_manager
# Open a part
part = part_manager.open_part("C:/path/to/part.prt")
# Save the part
part_manager.save_part(part)
# Close the part
part_manager.close_part(part)
"""
import os
import json
import subprocess
import tempfile
from pathlib import Path
from typing import Optional, Dict, Any, Tuple
# NX installation path (configurable)
NX_BIN_PATH = os.environ.get(
"NX_BIN_PATH",
r"C:\Program Files\Siemens\NX2506\NXBIN"
)
# Journal template for part operations
PART_OPERATIONS_JOURNAL = '''
# NX Open Python Journal - Part Operations
# Auto-generated by Atomizer hooks
import NXOpen
import NXOpen.UF
import json
import sys
import os
def main():
"""Execute part operation based on command arguments."""
# Get the NX session
session = NXOpen.Session.GetSession()
# Parse arguments: operation, part_path, [output_json]
args = sys.argv[1:] if len(sys.argv) > 1 else []
if len(args) < 2:
raise ValueError("Usage: script.py <operation> <part_path> [output_json]")
operation = args[0]
part_path = args[1]
output_json = args[2] if len(args) > 2 else None
result = {"success": False, "error": None, "data": {}}
try:
if operation == "open":
result = open_part(session, part_path)
elif operation == "close":
result = close_part(session, part_path)
elif operation == "save":
result = save_part(session, part_path)
elif operation == "save_as":
new_path = args[3] if len(args) > 3 else None
result = save_part_as(session, part_path, new_path)
elif operation == "info":
result = get_part_info(session, part_path)
else:
result["error"] = f"Unknown operation: {operation}"
except Exception as e:
result["error"] = str(e)
# Write result to output JSON if specified
if output_json:
with open(output_json, 'w') as f:
json.dump(result, f, indent=2)
return result
def open_part(session, part_path):
"""Open a part file.
NX Open API: Session.Parts().OpenActiveDisplay()
"""
result = {"success": False, "error": None, "data": {}}
if not os.path.exists(part_path):
result["error"] = f"Part file not found: {part_path}"
return result
try:
# Set load options for the working directory
working_dir = os.path.dirname(part_path)
session.Parts.LoadOptions.ComponentLoadMethod = NXOpen.LoadOptions.LoadMethod.FromDirectory
session.Parts.LoadOptions.SetSearchDirectories([working_dir], [True])
# Open the part using OpenActiveDisplay (more compatible with batch mode)
part, load_status = session.Parts.OpenActiveDisplay(
part_path,
NXOpen.DisplayPartOption.AllowAdditional
)
load_status.Dispose()
if part is None:
result["error"] = "Failed to open part - returned None"
return result
result["success"] = True
result["data"] = {
"part_name": part.Name,
"full_path": part.FullPath,
"leaf": part.Leaf,
"is_modified": part.IsModified,
"is_fully_loaded": part.IsFullyLoaded,
}
except Exception as e:
result["error"] = str(e)
return result
def close_part(session, part_path):
"""Close a part.
NX Open API: Part.Close()
"""
result = {"success": False, "error": None, "data": {}}
try:
# Find the part in the session
part = find_part_by_path(session, part_path)
if part is None:
result["error"] = f"Part not found in session: {part_path}"
return result
# Close the part
# Parameters: close_whole_tree, close_modified, responses
part.Close(
NXOpen.BasePart.CloseWholeTree.TrueValue,
NXOpen.BasePart.CloseModified.CloseModified,
None
)
result["success"] = True
result["data"] = {"closed": part_path}
except Exception as e:
result["error"] = str(e)
return result
def save_part(session, part_path):
"""Save a part.
NX Open API: Part.Save()
"""
result = {"success": False, "error": None, "data": {}}
try:
# Find the part in the session
part = find_part_by_path(session, part_path)
if part is None:
result["error"] = f"Part not found in session: {part_path}"
return result
# Save the part
# Parameters: save_component_parts, close_after_save
save_status = part.Save(
NXOpen.BasePart.SaveComponents.TrueValue,
NXOpen.BasePart.CloseAfterSave.FalseValue
)
result["success"] = True
result["data"] = {
"saved": part_path,
"is_modified": part.IsModified
}
except Exception as e:
result["error"] = str(e)
return result
def save_part_as(session, part_path, new_path):
"""Save a part with a new name.
NX Open API: Part.SaveAs()
"""
result = {"success": False, "error": None, "data": {}}
if not new_path:
result["error"] = "New path is required for SaveAs operation"
return result
try:
# Find the part in the session
part = find_part_by_path(session, part_path)
if part is None:
result["error"] = f"Part not found in session: {part_path}"
return result
# Save as new file
part.SaveAs(new_path)
result["success"] = True
result["data"] = {
"original": part_path,
"saved_as": new_path
}
except Exception as e:
result["error"] = str(e)
return result
def get_part_info(session, part_path):
"""Get information about a part.
NX Open API: Part properties
"""
result = {"success": False, "error": None, "data": {}}
try:
# Find the part in the session
part = find_part_by_path(session, part_path)
if part is None:
result["error"] = f"Part not found in session: {part_path}"
return result
# Get part info
result["success"] = True
result["data"] = {
"name": part.Name,
"full_path": part.FullPath,
"leaf": part.Leaf,
"is_modified": part.IsModified,
"is_fully_loaded": part.IsFullyLoaded,
"is_read_only": part.IsReadOnly,
"has_write_access": part.HasWriteAccess,
"part_units": str(part.PartUnits),
}
except Exception as e:
result["error"] = str(e)
return result
def find_part_by_path(session, part_path):
"""Find a part in the session by its file path."""
part_path_normalized = os.path.normpath(part_path).lower()
for part in session.Parts:
if os.path.normpath(part.FullPath).lower() == part_path_normalized:
return part
return None
if __name__ == "__main__":
main()
'''
def _get_run_journal_exe() -> str:
"""Get the path to run_journal.exe."""
return os.path.join(NX_BIN_PATH, "run_journal.exe")
def _run_journal(journal_path: str, args: list) -> Tuple[bool, str]:
"""Run an NX journal with arguments.
Returns:
Tuple of (success, output_or_error)
"""
run_journal = _get_run_journal_exe()
if not os.path.exists(run_journal):
return False, f"run_journal.exe not found at {run_journal}"
cmd = [run_journal, journal_path, "-args"] + args
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=120 # 2 minute timeout
)
if result.returncode != 0:
return False, f"Journal execution failed: {result.stderr}"
return True, result.stdout
except subprocess.TimeoutExpired:
return False, "Journal execution timed out"
except Exception as e:
return False, str(e)
def _execute_part_operation(
operation: str,
part_path: str,
extra_args: list = None
) -> Dict[str, Any]:
"""Execute a part operation via NX journal.
Args:
operation: The operation to perform (open, close, save, save_as, info)
part_path: Path to the part file
extra_args: Additional arguments for the operation
Returns:
Dict with operation result
"""
# Create temporary journal file
with tempfile.NamedTemporaryFile(
mode='w',
suffix='.py',
delete=False
) as journal_file:
journal_file.write(PART_OPERATIONS_JOURNAL)
journal_path = journal_file.name
# Create temporary output file
output_file = tempfile.NamedTemporaryFile(
mode='w',
suffix='.json',
delete=False
).name
try:
# Build arguments
args = [operation, part_path, output_file]
if extra_args:
args.extend(extra_args)
# Run the journal
success, output = _run_journal(journal_path, args)
if not success:
return {"success": False, "error": output, "data": {}}
# Read the result
if os.path.exists(output_file):
with open(output_file, 'r') as f:
return json.load(f)
else:
return {"success": False, "error": "Output file not created", "data": {}}
finally:
# Cleanup temporary files
if os.path.exists(journal_path):
os.unlink(journal_path)
if os.path.exists(output_file):
os.unlink(output_file)
# =============================================================================
# Public API
# =============================================================================
def open_part(part_path: str) -> Dict[str, Any]:
"""Open an NX part file.
Args:
part_path: Full path to the .prt file
Returns:
Dict with keys:
- success: bool
- error: Optional error message
- data: Dict with part_name, full_path, leaf, is_modified, is_fully_loaded
Example:
>>> result = open_part("C:/models/bracket.prt")
>>> if result["success"]:
... print(f"Opened: {result['data']['part_name']}")
"""
part_path = os.path.abspath(part_path)
if not os.path.exists(part_path):
return {
"success": False,
"error": f"Part file not found: {part_path}",
"data": {}
}
return _execute_part_operation("open", part_path)
def close_part(part_path: str) -> Dict[str, Any]:
"""Close an NX part.
Args:
part_path: Full path to the .prt file
Returns:
Dict with keys:
- success: bool
- error: Optional error message
- data: Dict with closed path
"""
part_path = os.path.abspath(part_path)
return _execute_part_operation("close", part_path)
def save_part(part_path: str) -> Dict[str, Any]:
"""Save an NX part.
Args:
part_path: Full path to the .prt file
Returns:
Dict with keys:
- success: bool
- error: Optional error message
- data: Dict with saved path and is_modified flag
"""
part_path = os.path.abspath(part_path)
return _execute_part_operation("save", part_path)
def save_part_as(part_path: str, new_path: str) -> Dict[str, Any]:
"""Save an NX part with a new name.
Args:
part_path: Full path to the original .prt file
new_path: Full path for the new file
Returns:
Dict with keys:
- success: bool
- error: Optional error message
- data: Dict with original and saved_as paths
"""
part_path = os.path.abspath(part_path)
new_path = os.path.abspath(new_path)
return _execute_part_operation("save_as", part_path, [new_path])
def get_part_info(part_path: str) -> Dict[str, Any]:
"""Get information about an NX part.
Args:
part_path: Full path to the .prt file
Returns:
Dict with keys:
- success: bool
- error: Optional error message
- data: Dict with name, full_path, leaf, is_modified,
is_fully_loaded, is_read_only, has_write_access, part_units
"""
part_path = os.path.abspath(part_path)
return _execute_part_operation("info", part_path)