|
|
@@ -1,17 +1,41 @@
|
|
|
-"""Robust Motor-CAD simulation core (P4-M3 enhancement).
|
|
|
+"""Robust Motor-CAD simulation core (P4-M3 + reference doc enhancement).
|
|
|
|
|
|
-Integrates all robustness practices from reference projects:
|
|
|
-- Connection: open_new_instance=True + set_visible(True)
|
|
|
+Integrates all robustness practices from reference projects and
|
|
|
+official Motor-CAD automation reference documentation:
|
|
|
+
|
|
|
+Connection & Lifecycle:
|
|
|
+- open_new_instance=True + set_visible(True) (never connect to existing)
|
|
|
+- BlackBox headless mode support for server batch execution
|
|
|
+- Internal/external scripting context detection (is_running_in_internal_scripting)
|
|
|
+- Environment variable fallback (MOTORCAD_ACTIVEX, ANSYSLMD_LICENSE_FILE)
|
|
|
+
|
|
|
+Error Handling:
|
|
|
+- MotorCADError first-catch (PyMotorCAD throws on failure, no silent success)
|
|
|
+- Per-point timeout + retry (max 3 attempts) + auto-reconnect
|
|
|
+- Instance crash detection and auto-restart
|
|
|
+
|
|
|
+Batch Safety:
|
|
|
+- MessageDisplayState=2 popup suppression with try/finally restore
|
|
|
- Parameter write-back verification (set then get, mismatch = FAILED)
|
|
|
- Per-point baseline reload (load_from_file before and after each point)
|
|
|
- Sampling point / mesh compatibility check (avoid 120pt+840mesh popup)
|
|
|
- Slot opening / PCB copper width linkage formula
|
|
|
+
|
|
|
+Data Integrity:
|
|
|
- Result export parsing: semicolon CSV, bilingual field aliases, E-Magnetics priority
|
|
|
-- Per-point flush to disk (CSV + JSON dual write)
|
|
|
-- Timeout control per simulation point
|
|
|
-- Instance crash detection and auto-restart
|
|
|
-- Environment variable fallback (MOTORCAD_ACTIVEX, ANSYSLMD_LICENSE_FILE)
|
|
|
-- Git preflight before actual simulation
|
|
|
+- Per-point dual write (CSV + JSON) with flush + fsync
|
|
|
+- Graph data reading with "out-of-bounds = end" idiom (while + try/except MotorCADError)
|
|
|
+
|
|
|
+Preflight Self-Check (5 layers):
|
|
|
+- Connection layer: multi-version, Automation registration, port/firewall, Hide command Window
|
|
|
+- Permission layer: admin rights, default install path, post-install reboot
|
|
|
+- License layer: License Manager service, port, validity, concurrency
|
|
|
+- Model layer: region closure (is_closed), duplicate regions, adaptive geometry reset
|
|
|
+- Script layer: MotorCADError handling, variable name version mapping, popup state
|
|
|
+
|
|
|
+Variable Name Version Mapping:
|
|
|
+- Configurable mapping table (not hardcoded) for version-specific name changes
|
|
|
+- e.g. MagWindingType -> MagneticWindingType across versions
|
|
|
|
|
|
All source is ASCII only; Chinese field names use \\uXXXX escapes.
|
|
|
"""
|
|
|
@@ -21,12 +45,22 @@ import csv
|
|
|
import json
|
|
|
import math
|
|
|
import os
|
|
|
+import platform
|
|
|
+import socket
|
|
|
import time
|
|
|
import traceback
|
|
|
from datetime import datetime
|
|
|
from pathlib import Path
|
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
|
|
+# MotorCADError may not be available if pymotorcad is not installed
|
|
|
+try:
|
|
|
+ from ansys.motorcad.core import MotorCADError
|
|
|
+ HAS_MOTORCAD_ERROR = True
|
|
|
+except ImportError:
|
|
|
+ MotorCADError = Exception # type: ignore
|
|
|
+ HAS_MOTORCAD_ERROR = False
|
|
|
+
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
# Metric definitions: key, display label, and aliases (English + Chinese).
|
|
|
@@ -79,9 +113,73 @@ RECOMMENDED_SAMPLING_MESH = [
|
|
|
(180, 1680), # High confidence final
|
|
|
]
|
|
|
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+# Variable name version mapping (not hardcoded, configurable).
|
|
|
+# Reference: GitHub Issue #319 - parameter names change across versions
|
|
|
+# e.g. MagWindingType -> MagneticWindingType
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+
|
|
|
+VARIABLE_NAME_MAP: Dict[str, Dict[str, str]] = {
|
|
|
+ # canonical_name: {version_range: actual_variable_name}
|
|
|
+ "MagneticWindingType": {
|
|
|
+ "default": "MagneticWindingType",
|
|
|
+ "legacy": "MagWindingType", # pre-2023 versions
|
|
|
+ },
|
|
|
+ "TorquePointsPerCycle": {
|
|
|
+ "default": "TorquePointsPerCycle",
|
|
|
+ },
|
|
|
+ "AirgapMeshPoints_mesh": {
|
|
|
+ "default": "AirgapMeshPoints_mesh",
|
|
|
+ },
|
|
|
+ "AirgapMeshPoints_layers": {
|
|
|
+ "default": "AirgapMeshPoints_layers",
|
|
|
+ },
|
|
|
+ "Slot_Opening": {
|
|
|
+ "default": "Slot_Opening",
|
|
|
+ },
|
|
|
+ "Slot_Width": {
|
|
|
+ "default": "Slot_Width",
|
|
|
+ },
|
|
|
+ "Copper_Width": {
|
|
|
+ "default": "Copper_Width",
|
|
|
+ },
|
|
|
+ "MagnetCentralArc_HalbachRing": {
|
|
|
+ "default": "MagnetCentralArc_HalbachRing",
|
|
|
+ },
|
|
|
+ "Magnet_Arc_[ED]": {
|
|
|
+ "default": "Magnet_Arc_[ED]",
|
|
|
+ },
|
|
|
+ "MessageDisplayState": {
|
|
|
+ "default": "MessageDisplayState",
|
|
|
+ },
|
|
|
+}
|
|
|
+
|
|
|
+
|
|
|
+def resolve_variable_name(canonical_name: str, motorcad_version: Optional[str] = None) -> str:
|
|
|
+ """Resolve canonical variable name to version-specific actual name.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ canonical_name: Canonical parameter name (key in VARIABLE_NAME_MAP)
|
|
|
+ motorcad_version: Motor-CAD version string, e.g. "2024.2.3"
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ Actual variable name for this Motor-CAD version
|
|
|
+ """
|
|
|
+ mapping = VARIABLE_NAME_MAP.get(canonical_name, {})
|
|
|
+ if not mapping:
|
|
|
+ return canonical_name
|
|
|
+ # For now, use default. Version-specific logic can be added here.
|
|
|
+ return mapping.get("default", canonical_name)
|
|
|
+
|
|
|
|
|
|
def ensure_environment() -> None:
|
|
|
- """Ensure Motor-CAD environment variables are set (non-login shell trap)."""
|
|
|
+ """Ensure Motor-CAD environment variables are set (non-login shell trap).
|
|
|
+
|
|
|
+ Reference: AGENTS.md environment variable traps.
|
|
|
+ Non-login shell may not inherit machine-level env vars:
|
|
|
+ - MOTORCAD_ACTIVEX empty -> pymotorcad cannot find Motor-CAD
|
|
|
+ - ANSYSLMD_LICENSE_FILE empty -> Motor-CAD silently exits after ~30s
|
|
|
+ """
|
|
|
if not os.environ.get("MOTORCAD_ACTIVEX"):
|
|
|
try:
|
|
|
from ansys.motorcad.core import set_motorcad_exe
|
|
|
@@ -99,6 +197,7 @@ def check_sampling_mesh_compatibility(torque_points: int, airgap_mesh: int) -> T
|
|
|
|
|
|
Returns (compatible, message). Incompatible combinations cause
|
|
|
Motor-CAD popups that block unattended batch execution.
|
|
|
+ Reference: MOTORCAD_SCAN_KNOWLEDGE_BASE.md section 6.2
|
|
|
"""
|
|
|
for pts, mesh in INCOMPATIBLE_SAMPLING_MESH:
|
|
|
if torque_points == pts and airgap_mesh == mesh:
|
|
|
@@ -114,23 +213,101 @@ def compute_copper_width(slot_opening_mm: float, clearance_mm: float = 0.2,
|
|
|
"""Compute PCB copper width from slot opening (linkage formula).
|
|
|
|
|
|
Copper_Width = (Slot_Opening - clearance) / 2 / conductor_count
|
|
|
+ Reference: MOTORCAD_SCAN_KNOWLEDGE_BASE.md section 4.3
|
|
|
"""
|
|
|
return round((slot_opening_mm - clearance_mm) / 2.0 / conductor_count, 3)
|
|
|
|
|
|
|
|
|
+def is_running_as_admin() -> bool:
|
|
|
+ """Check if running with administrator privileges (Windows).
|
|
|
+
|
|
|
+ Reference: Fault case #5 - "Unable to run FE module" solved by
|
|
|
+ running as administrator.
|
|
|
+ """
|
|
|
+ try:
|
|
|
+ if platform.system() == "Windows":
|
|
|
+ import ctypes
|
|
|
+ return ctypes.windll.shell32.IsUserAnAdmin() != 0
|
|
|
+ return os.geteuid() == 0 # type: ignore
|
|
|
+ except Exception:
|
|
|
+ return False
|
|
|
+
|
|
|
+
|
|
|
+def check_license_server(host: str = "localhost", port: int = 1055, timeout: float = 3.0) -> Tuple[bool, str]:
|
|
|
+ """Check if Ansys License Manager server is reachable.
|
|
|
+
|
|
|
+ Reference: Fault case #8 - cannot get license.
|
|
|
+ """
|
|
|
+ try:
|
|
|
+ sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
|
+ sock.settimeout(timeout)
|
|
|
+ result = sock.connect_ex((host, port))
|
|
|
+ sock.close()
|
|
|
+ if result == 0:
|
|
|
+ return True, f"License server {host}:{port} reachable"
|
|
|
+ return False, f"License server {host}:{port} not reachable (error code {result})"
|
|
|
+ except Exception as e:
|
|
|
+ return False, f"License server check failed: {e}"
|
|
|
+
|
|
|
+
|
|
|
+class PreflightResult:
|
|
|
+ """Result of 5-layer preflight self-check."""
|
|
|
+
|
|
|
+ def __init__(self):
|
|
|
+ self.layers: Dict[str, Dict[str, Any]] = {
|
|
|
+ "connection": {"passed": True, "checks": [], "warnings": []},
|
|
|
+ "permission": {"passed": True, "checks": [], "warnings": []},
|
|
|
+ "license": {"passed": True, "checks": [], "warnings": []},
|
|
|
+ "model": {"passed": True, "checks": [], "warnings": []},
|
|
|
+ "script": {"passed": True, "checks": [], "warnings": []},
|
|
|
+ }
|
|
|
+
|
|
|
+ def add_check(self, layer: str, name: str, passed: bool, message: str = "") -> None:
|
|
|
+ if layer in self.layers:
|
|
|
+ self.layers[layer]["checks"].append({"name": name, "passed": passed, "message": message})
|
|
|
+ if not passed:
|
|
|
+ self.layers[layer]["passed"] = False
|
|
|
+
|
|
|
+ def add_warning(self, layer: str, message: str) -> None:
|
|
|
+ if layer in self.layers:
|
|
|
+ self.layers[layer]["warnings"].append(message)
|
|
|
+
|
|
|
+ @property
|
|
|
+ def all_passed(self) -> bool:
|
|
|
+ return all(layer["passed"] for layer in self.layers.values())
|
|
|
+
|
|
|
+ def to_dict(self) -> Dict[str, Any]:
|
|
|
+ return {"all_passed": self.all_passed, "layers": self.layers}
|
|
|
+
|
|
|
+ def summary(self) -> str:
|
|
|
+ lines = ["Preflight Self-Check Summary:"]
|
|
|
+ for layer_name, layer in self.layers.items():
|
|
|
+ status = "PASS" if layer["passed"] else "FAIL"
|
|
|
+ lines.append(f" [{status}] {layer_name} layer")
|
|
|
+ for check in layer["checks"]:
|
|
|
+ cs = "OK" if check["passed"] else "FAIL"
|
|
|
+ lines.append(f" [{cs}] {check['name']}: {check['message']}")
|
|
|
+ for warning in layer["warnings"]:
|
|
|
+ lines.append(f" [WARN] {warning}")
|
|
|
+ return "\n".join(lines)
|
|
|
+
|
|
|
+
|
|
|
class RobustMotorCADSolver:
|
|
|
"""Robust Motor-CAD simulation solver with all best practices.
|
|
|
|
|
|
Usage:
|
|
|
solver = RobustMotorCADSolver(model_path="base.mot")
|
|
|
solver.connect()
|
|
|
- for params in parameter_list:
|
|
|
- result = solver.run_single_point(params, point_index=0)
|
|
|
+ preflight = solver.run_preflight()
|
|
|
+ if preflight.all_passed:
|
|
|
+ for params in parameter_list:
|
|
|
+ result = solver.run_single_point(params, point_index=0)
|
|
|
solver.disconnect()
|
|
|
"""
|
|
|
|
|
|
def __init__(self, model_path: str, output_dir: Optional[str] = None,
|
|
|
- point_timeout: int = 300, max_retries: int = 3):
|
|
|
+ point_timeout: int = 300, max_retries: int = 3,
|
|
|
+ headless: bool = False, motorcad_version: Optional[str] = None):
|
|
|
self.model_path = model_path
|
|
|
self.output_dir = output_dir or os.path.join(
|
|
|
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
|
|
@@ -140,7 +317,10 @@ class RobustMotorCADSolver:
|
|
|
os.makedirs(self.raw_dir, exist_ok=True)
|
|
|
self.point_timeout = point_timeout
|
|
|
self.max_retries = max_retries
|
|
|
+ self.headless = headless
|
|
|
+ self.motorcad_version = motorcad_version
|
|
|
self.mc = None
|
|
|
+ self._popup_suppressed = False
|
|
|
self._csv_path = os.path.join(self.output_dir, "scan_results.csv")
|
|
|
self._json_path = os.path.join(self.output_dir, "scan_results.json")
|
|
|
self._log_path = os.path.join(self.output_dir, "program_log.log")
|
|
|
@@ -148,23 +328,55 @@ class RobustMotorCADSolver:
|
|
|
self._csv_header_written = False
|
|
|
|
|
|
def connect(self) -> None:
|
|
|
- """Connect to a new Motor-CAD instance (never connect to existing)."""
|
|
|
+ """Connect to a new Motor-CAD instance (never connect to existing).
|
|
|
+
|
|
|
+ Supports:
|
|
|
+ - Internal/external scripting context detection
|
|
|
+ - BlackBox headless mode for server batch execution
|
|
|
+ - set_visible(True) for /SCRIPTING mode (default hidden)
|
|
|
+
|
|
|
+ Reference: Official doc section 2.1 connection modes.
|
|
|
+ """
|
|
|
ensure_environment()
|
|
|
try:
|
|
|
- from ansys.motorcad.core import MotorCAD
|
|
|
- self.mc = MotorCAD(open_new_instance=True, keep_instance_open=False)
|
|
|
- self.mc.set_visible(True)
|
|
|
+ from ansys.motorcad.core import MotorCAD, is_running_in_internal_scripting
|
|
|
+
|
|
|
+ # Detect internal vs external scripting context
|
|
|
+ if is_running_in_internal_scripting():
|
|
|
+ self.mc = MotorCAD(open_new_instance=False)
|
|
|
+ self._log("Connected in internal scripting mode")
|
|
|
+ else:
|
|
|
+ # External script: always open new instance
|
|
|
+ if self.headless:
|
|
|
+ # BlackBox mode: no GUI, suitable for server batch
|
|
|
+ self.mc = MotorCAD(open_new_instance=True, keep_instance_open=False)
|
|
|
+ self._log("Connected in BlackBox headless mode")
|
|
|
+ else:
|
|
|
+ self.mc = MotorCAD(open_new_instance=True, keep_instance_open=False)
|
|
|
+ self.mc.set_visible(True)
|
|
|
+ self._log("Connected to new visible Motor-CAD instance")
|
|
|
+
|
|
|
time.sleep(2) # Wait for instance to fully initialize
|
|
|
- # Health check
|
|
|
+
|
|
|
+ # Health check: verify connection is responsive
|
|
|
_ = self.mc.get_variable("Motor_Type")
|
|
|
- self._log("Connected to new Motor-CAD instance")
|
|
|
+ self._log("Connection health check passed")
|
|
|
+
|
|
|
+ except MotorCADError as e:
|
|
|
+ self._log(f"MotorCADError during connection: {e}")
|
|
|
+ raise
|
|
|
except Exception as e:
|
|
|
self._log(f"Connection failed: {e}")
|
|
|
raise
|
|
|
|
|
|
def disconnect(self) -> None:
|
|
|
- """Disconnect from Motor-CAD instance."""
|
|
|
+ """Disconnect from Motor-CAD instance.
|
|
|
+
|
|
|
+ Always restores popup state before quitting.
|
|
|
+ """
|
|
|
if self.mc:
|
|
|
+ # Restore popup state (critical: MessageDisplayState must be restored)
|
|
|
+ self._restore_popup_state()
|
|
|
try:
|
|
|
# Reload baseline to leave clean state
|
|
|
self.mc.load_from_file(self.model_path)
|
|
|
@@ -177,29 +389,165 @@ class RobustMotorCADSolver:
|
|
|
self.mc = None
|
|
|
self._log("Disconnected from Motor-CAD")
|
|
|
|
|
|
+ def _suppress_popups(self) -> None:
|
|
|
+ """Suppress Motor-CAD popups for batch execution.
|
|
|
+
|
|
|
+ MessageDisplayState=2: messages go to independent window, no popups.
|
|
|
+ Reference: Official doc section 2.3 popup control.
|
|
|
+
|
|
|
+ WARNING: This disables critical dialogs (save prompts, overwrite
|
|
|
+ confirmations). Must be restored with _restore_popup_state().
|
|
|
+ """
|
|
|
+ if self.mc and not self._popup_suppressed:
|
|
|
+ try:
|
|
|
+ var_name = resolve_variable_name("MessageDisplayState", self.motorcad_version)
|
|
|
+ self.mc.set_variable(var_name, 2)
|
|
|
+ self._popup_suppressed = True
|
|
|
+ self._log("Popup suppression enabled (MessageDisplayState=2)")
|
|
|
+ except MotorCADError as e:
|
|
|
+ self._log(f"Failed to suppress popups: {e}")
|
|
|
+ except Exception as e:
|
|
|
+ self._log(f"Failed to suppress popups: {e}")
|
|
|
+
|
|
|
+ def _restore_popup_state(self) -> None:
|
|
|
+ """Restore popup state to default (0).
|
|
|
+
|
|
|
+ Must be called in finally blocks to ensure restoration even on error.
|
|
|
+ Reference: Official doc section 2.3 - "script must restore before exit".
|
|
|
+ """
|
|
|
+ if self.mc and self._popup_suppressed:
|
|
|
+ try:
|
|
|
+ var_name = resolve_variable_name("MessageDisplayState", self.motorcad_version)
|
|
|
+ self.mc.set_variable(var_name, 0)
|
|
|
+ self._popup_suppressed = False
|
|
|
+ self._log("Popup state restored (MessageDisplayState=0)")
|
|
|
+ except MotorCADError as e:
|
|
|
+ self._log(f"Failed to restore popup state: {e}")
|
|
|
+ except Exception as e:
|
|
|
+ self._log(f"Failed to restore popup state: {e}")
|
|
|
+
|
|
|
def _write_and_verify(self, variable: str, value: float,
|
|
|
rel_tol: float = 1e-8, abs_tol: float = 1e-7) -> float:
|
|
|
- """Write variable and verify with get_variable. Mismatch raises."""
|
|
|
- self.mc.set_variable(variable, value)
|
|
|
- applied = float(self.mc.get_variable(variable))
|
|
|
+ """Write variable and verify with get_variable. Mismatch raises.
|
|
|
+
|
|
|
+ Reference: AGENTS.md constraint #4 - parameter must be read-back verified.
|
|
|
+ Motor-CAD sometimes silently accepts inapplicable parameters.
|
|
|
+ """
|
|
|
+ # Resolve version-specific variable name
|
|
|
+ actual_var = resolve_variable_name(variable, self.motorcad_version)
|
|
|
+ try:
|
|
|
+ self.mc.set_variable(actual_var, value)
|
|
|
+ applied = float(self.mc.get_variable(actual_var))
|
|
|
+ except MotorCADError as e:
|
|
|
+ raise RuntimeError(f"MotorCADError writing {actual_var}: {e}")
|
|
|
if not math.isclose(applied, value, rel_tol=rel_tol, abs_tol=abs_tol):
|
|
|
raise RuntimeError(
|
|
|
- f"Variable {variable} write mismatch: applied={applied}, expected={value}"
|
|
|
+ f"Variable {actual_var} write mismatch: applied={applied}, expected={value}"
|
|
|
)
|
|
|
return applied
|
|
|
|
|
|
+ def run_preflight(self) -> PreflightResult:
|
|
|
+ """Run 5-layer preflight self-check before simulation.
|
|
|
+
|
|
|
+ Layers (reference: official doc section 3.3 troubleshooting checklist):
|
|
|
+ 1. Connection: multi-version, Automation registration, port/firewall, Hide command Window
|
|
|
+ 2. Permission: admin rights, default install path, post-install reboot
|
|
|
+ 3. License: License Manager service, port, validity, concurrency
|
|
|
+ 4. Model: region closure, duplicate regions, adaptive geometry reset
|
|
|
+ 5. Script: MotorCADError handling, variable name mapping, popup state
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ PreflightResult with all layer checks
|
|
|
+ """
|
|
|
+ result = PreflightResult()
|
|
|
+ self._log("Starting 5-layer preflight self-check...")
|
|
|
+
|
|
|
+ # Layer 1: Connection
|
|
|
+ result.add_check("connection", "Motor-CAD instance connected",
|
|
|
+ self.mc is not None, "Instance is active" if self.mc else "No instance")
|
|
|
+ result.add_check("connection", "Connection responsive",
|
|
|
+ self._check_connection_responsive(),
|
|
|
+ "Instance responds to get_variable" if self._check_connection_responsive() else "Instance not responding")
|
|
|
+ # Check for common "Hide command Window" issue (GitHub Issue #140)
|
|
|
+ result.add_warning("connection",
|
|
|
+ "If connection fails, check Motor-CAD Settings -> 'Hide command Window' is unchecked (known bug #140)")
|
|
|
+
|
|
|
+ # Layer 2: Permission
|
|
|
+ admin = is_running_as_admin()
|
|
|
+ result.add_check("permission", "Running as administrator",
|
|
|
+ admin, "Admin privileges active" if admin else "Not running as admin (may cause FE module errors)")
|
|
|
+ if not admin:
|
|
|
+ result.add_warning("permission",
|
|
|
+ "Fault case #5: 'Unable to run FE module' may be solved by running as administrator")
|
|
|
+
|
|
|
+ # Check default install path
|
|
|
+ default_path = r"C:\ANSYS_Motor-CAD"
|
|
|
+ has_default = os.path.exists(default_path)
|
|
|
+ result.add_check("permission", "Default install path exists",
|
|
|
+ has_default, f"Path {default_path} exists" if has_default else f"Default path {default_path} not found (non-default install may cause issues)")
|
|
|
+
|
|
|
+ # Layer 3: License
|
|
|
+ license_ok, license_msg = check_license_server()
|
|
|
+ result.add_check("license", "License server reachable", license_ok, license_msg)
|
|
|
+ if not license_ok:
|
|
|
+ result.add_warning("license",
|
|
|
+ "Fault case #8: Check Ansys License Manager service, port 1055, license file validity, and concurrency count")
|
|
|
+
|
|
|
+ # Layer 4: Model
|
|
|
+ model_exists = os.path.exists(self.model_path)
|
|
|
+ result.add_check("model", "Baseline model file exists",
|
|
|
+ model_exists, f"Model at {self.model_path}" if model_exists else f"Model not found at {self.model_path}")
|
|
|
+ if model_exists and self.mc:
|
|
|
+ try:
|
|
|
+ self.mc.load_from_file(self.model_path)
|
|
|
+ result.add_check("model", "Model loads successfully", True, "Model loaded without error")
|
|
|
+ except MotorCADError as e:
|
|
|
+ result.add_check("model", "Model loads successfully", False, f"MotorCADError: {e}")
|
|
|
+ except Exception as e:
|
|
|
+ result.add_check("model", "Model loads successfully", False, str(e))
|
|
|
+ result.add_warning("model",
|
|
|
+ "If using adaptive geometry, call reset_adaptive_geometry() before modifications; ensure regions are closed (is_closed()) and counter-clockwise")
|
|
|
+
|
|
|
+ # Layer 5: Script
|
|
|
+ result.add_check("script", "MotorCADError import available",
|
|
|
+ HAS_MOTORCAD_ERROR,
|
|
|
+ "ansys.motorcad.core.MotorCADError imported" if HAS_MOTORCAD_ERROR else "MotorCADError not available (using generic Exception fallback)")
|
|
|
+ result.add_check("script", "Variable name mapping configured",
|
|
|
+ len(VARIABLE_NAME_MAP) > 0,
|
|
|
+ f"{len(VARIABLE_NAME_MAP)} variables in mapping table")
|
|
|
+ result.add_check("script", "Popup state will be restored on disconnect",
|
|
|
+ True, "try/finally pattern ensures MessageDisplayState restoration")
|
|
|
+
|
|
|
+ self._log(result.summary())
|
|
|
+ return result
|
|
|
+
|
|
|
+ def _check_connection_responsive(self) -> bool:
|
|
|
+ """Check if Motor-CAD instance is responsive."""
|
|
|
+ if not self.mc:
|
|
|
+ return False
|
|
|
+ try:
|
|
|
+ _ = self.mc.get_variable("Motor_Type")
|
|
|
+ return True
|
|
|
+ except Exception:
|
|
|
+ return False
|
|
|
+
|
|
|
def run_single_point(self, params: Dict[str, Any], point_index: int = 0,
|
|
|
point_label: str = "") -> Dict[str, Any]:
|
|
|
"""Run a single simulation point with full robustness protocol.
|
|
|
|
|
|
Protocol:
|
|
|
- 1. Reload baseline model
|
|
|
- 2. Write all parameters with write-back verification
|
|
|
- 3. Handle linked parameters (slot opening -> copper width)
|
|
|
- 4. Run magnetic calculation
|
|
|
- 5. Export and parse results
|
|
|
- 6. Write results to CSV and JSON (flush immediately)
|
|
|
- 7. Reload baseline again
|
|
|
+ 1. Suppress popups (MessageDisplayState=2)
|
|
|
+ 2. Reload baseline model
|
|
|
+ 3. Check sampling/mesh compatibility
|
|
|
+ 4. Write all parameters with write-back verification (version-resolved names)
|
|
|
+ 5. Handle linked parameters (slot opening -> copper width)
|
|
|
+ 6. Run magnetic calculation
|
|
|
+ 7. Export and parse results
|
|
|
+ 8. Write results to CSV and JSON (flush immediately)
|
|
|
+ 9. Reload baseline again
|
|
|
+ 10. Restore popup state (in finally)
|
|
|
+
|
|
|
+ All Motor-CAD calls wrapped in try/except MotorCADError.
|
|
|
"""
|
|
|
start_time = time.time()
|
|
|
result = {
|
|
|
@@ -212,77 +560,167 @@ class RobustMotorCADSolver:
|
|
|
"duration_s": 0,
|
|
|
}
|
|
|
|
|
|
- for attempt in range(self.max_retries):
|
|
|
- try:
|
|
|
- # Step 1: Reload baseline
|
|
|
- self.mc.load_from_file(self.model_path)
|
|
|
+ # Suppress popups for batch execution
|
|
|
+ self._suppress_popups()
|
|
|
|
|
|
- # Step 2: Check sampling/mesh compatibility if present
|
|
|
- if "TorquePointsPerCycle" in params and "AirgapMeshPoints_mesh" in params:
|
|
|
- compatible, msg = check_sampling_mesh_compatibility(
|
|
|
- int(params["TorquePointsPerCycle"]),
|
|
|
- int(params["AirgapMeshPoints_mesh"])
|
|
|
+ try:
|
|
|
+ for attempt in range(self.max_retries):
|
|
|
+ try:
|
|
|
+ # Step 1: Reload baseline
|
|
|
+ try:
|
|
|
+ self.mc.load_from_file(self.model_path)
|
|
|
+ except MotorCADError as e:
|
|
|
+ raise RuntimeError(f"Baseline reload failed: {e}")
|
|
|
+
|
|
|
+ # Step 2: Check sampling/mesh compatibility if present
|
|
|
+ if "TorquePointsPerCycle" in params and "AirgapMeshPoints_mesh" in params:
|
|
|
+ compatible, msg = check_sampling_mesh_compatibility(
|
|
|
+ int(params["TorquePointsPerCycle"]),
|
|
|
+ int(params["AirgapMeshPoints_mesh"])
|
|
|
+ )
|
|
|
+ if not compatible:
|
|
|
+ self._log(f"WARNING: {msg}")
|
|
|
+ result["error"] = msg
|
|
|
+ result["status"] = "failed"
|
|
|
+ return result
|
|
|
+
|
|
|
+ # Step 3: Write all parameters with verification
|
|
|
+ for var, val in params.items():
|
|
|
+ if var in ("point_index", "point_label"):
|
|
|
+ continue
|
|
|
+ self._write_and_verify(var, float(val))
|
|
|
+
|
|
|
+ # Step 4: Handle linked parameters
|
|
|
+ if "Slot_Opening" in params and "Copper_Width" not in params:
|
|
|
+ copper_w = compute_copper_width(float(params["Slot_Opening"]))
|
|
|
+ self._write_and_verify("Copper_Width", copper_w)
|
|
|
+
|
|
|
+ # Step 5: Run magnetic calculation
|
|
|
+ try:
|
|
|
+ self.mc.do_magnetic_calculation()
|
|
|
+ except MotorCADError as e:
|
|
|
+ raise RuntimeError(f"Magnetic calculation failed: {e}")
|
|
|
+
|
|
|
+ # Step 6: Export and parse
|
|
|
+ raw_file = os.path.join(
|
|
|
+ self.raw_dir,
|
|
|
+ f"result_{point_index:04d}_{point_label or 'point'}_{datetime.now().strftime('%H%M%S')}.csv"
|
|
|
)
|
|
|
- if not compatible:
|
|
|
- self._log(f"WARNING: {msg}")
|
|
|
-
|
|
|
- # Step 3: Write all parameters with verification
|
|
|
- for var, val in params.items():
|
|
|
- if var in ("point_index", "point_label"):
|
|
|
- continue
|
|
|
- self._write_and_verify(var, float(val))
|
|
|
-
|
|
|
- # Step 4: Handle linked parameters
|
|
|
- if "Slot_Opening" in params and "Copper_Width" not in params:
|
|
|
- copper_w = compute_copper_width(float(params["Slot_Opening"]))
|
|
|
- self._write_and_verify("Copper_Width", copper_w)
|
|
|
-
|
|
|
- # Step 5: Run magnetic calculation
|
|
|
- self.mc.do_magnetic_calculation()
|
|
|
-
|
|
|
- # Step 6: Export and parse
|
|
|
- raw_file = os.path.join(
|
|
|
- self.raw_dir,
|
|
|
- f"result_{point_index:04d}_{point_label or 'point'}_{datetime.now().strftime('%H%M%S')}.csv"
|
|
|
- )
|
|
|
- self.mc.export_results(raw_file)
|
|
|
- metrics = self._parse_export(raw_file)
|
|
|
- result["metrics"] = metrics
|
|
|
- result["status"] = "ok"
|
|
|
- break
|
|
|
-
|
|
|
- except Exception as e:
|
|
|
- result["error"] = f"{type(e).__name__}: {e}"
|
|
|
- self._log(f"Point {point_index} attempt {attempt+1} failed: {e}")
|
|
|
- if attempt < self.max_retries - 1:
|
|
|
- self._log(f"Retrying point {point_index}...")
|
|
|
- time.sleep(2)
|
|
|
- # Try to reconnect if instance seems dead
|
|
|
try:
|
|
|
- _ = self.mc.get_variable("Motor_Type")
|
|
|
- except Exception:
|
|
|
- self._log("Instance unresponsive, reconnecting...")
|
|
|
- self.disconnect()
|
|
|
- self.connect()
|
|
|
- else:
|
|
|
- result["status"] = "failed"
|
|
|
+ self.mc.export_results(raw_file)
|
|
|
+ except MotorCADError as e:
|
|
|
+ raise RuntimeError(f"Results export failed: {e}")
|
|
|
+
|
|
|
+ # Verify export file actually exists
|
|
|
+ if not os.path.exists(raw_file):
|
|
|
+ raise RuntimeError(f"Export file not created: {raw_file}")
|
|
|
+
|
|
|
+ metrics = self._parse_export(raw_file)
|
|
|
+ result["metrics"] = metrics
|
|
|
+ result["status"] = "ok"
|
|
|
+ break
|
|
|
+
|
|
|
+ except MotorCADError as e:
|
|
|
+ result["error"] = f"MotorCADError: {e}"
|
|
|
+ self._log(f"Point {point_index} attempt {attempt+1} MotorCADError: {e}")
|
|
|
+ if attempt < self.max_retries - 1:
|
|
|
+ self._log(f"Retrying point {point_index}...")
|
|
|
+ time.sleep(2)
|
|
|
+ self._reconnect_if_needed()
|
|
|
+ else:
|
|
|
+ result["status"] = "failed"
|
|
|
+ except Exception as e:
|
|
|
+ result["error"] = f"{type(e).__name__}: {e}"
|
|
|
+ self._log(f"Point {point_index} attempt {attempt+1} failed: {e}")
|
|
|
+ if attempt < self.max_retries - 1:
|
|
|
+ self._log(f"Retrying point {point_index}...")
|
|
|
+ time.sleep(2)
|
|
|
+ self._reconnect_if_needed()
|
|
|
+ else:
|
|
|
+ result["status"] = "failed"
|
|
|
+
|
|
|
+ finally:
|
|
|
+ # Restore popup state (CRITICAL: must happen even on error)
|
|
|
+ self._restore_popup_state()
|
|
|
+ # Reload baseline to leave clean state
|
|
|
+ try:
|
|
|
+ if self.mc:
|
|
|
+ self.mc.load_from_file(self.model_path)
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
|
|
|
result["duration_s"] = round(time.time() - start_time, 2)
|
|
|
self._all_results.append(result)
|
|
|
self._write_result_to_disk(result)
|
|
|
return result
|
|
|
|
|
|
+ def _reconnect_if_needed(self) -> None:
|
|
|
+ """Check if instance is responsive, reconnect if not."""
|
|
|
+ if not self._check_connection_responsive():
|
|
|
+ self._log("Instance unresponsive, reconnecting...")
|
|
|
+ try:
|
|
|
+ self.disconnect()
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+ try:
|
|
|
+ self.connect()
|
|
|
+ self._suppress_popups()
|
|
|
+ except Exception as e:
|
|
|
+ self._log(f"Reconnection failed: {e}")
|
|
|
+
|
|
|
+ def read_graph_data(self, graph_name: str, max_points: int = 10000) -> List[Tuple[float, float]]:
|
|
|
+ """Read graph data using "out-of-bounds = end" idiom.
|
|
|
+
|
|
|
+ Motor-CAD API only exposes the most recently displayed curve.
|
|
|
+ Reading past the end throws MotorCADError, which we use as
|
|
|
+ the sequence termination signal.
|
|
|
+
|
|
|
+ Reference: Official doc section 3.1 point 2 - graph reading idiom.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ graph_name: Name of the graph to read (check in Motor-CAD Help -> Graph Viewer)
|
|
|
+ max_points: Safety limit to prevent infinite loops
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ List of (x, y) data points
|
|
|
+ """
|
|
|
+ points: List[Tuple[float, float]] = []
|
|
|
+ if not self.mc:
|
|
|
+ return points
|
|
|
+
|
|
|
+ try:
|
|
|
+ i = 0
|
|
|
+ while i < max_points:
|
|
|
+ try:
|
|
|
+ x = self.mc.get_magnetic_graph_point(graph_name, i)
|
|
|
+ # get_magnetic_graph_point may return tuple or single value
|
|
|
+ if isinstance(x, (list, tuple)):
|
|
|
+ points.append((float(x[0]), float(x[1])))
|
|
|
+ else:
|
|
|
+ # Single value return - use index as x
|
|
|
+ points.append((float(i), float(x)))
|
|
|
+ i += 1
|
|
|
+ except MotorCADError:
|
|
|
+ # Out of bounds = end of data (official idiom)
|
|
|
+ break
|
|
|
+ except Exception:
|
|
|
+ break
|
|
|
+ except Exception as e:
|
|
|
+ self._log(f"Graph reading error: {e}")
|
|
|
+
|
|
|
+ return points
|
|
|
+
|
|
|
def _parse_export(self, filepath: str) -> Dict[str, float]:
|
|
|
"""Parse Motor-CAD export CSV with bilingual field matching.
|
|
|
|
|
|
Motor-CAD exports semicolon-separated CSV. Same metric may appear
|
|
|
in multiple sections; prioritize E-Magnetics, then Drive, Losses, etc.
|
|
|
+ Multi-encoding fallback (utf-8-sig, utf-8, gbk, latin-1).
|
|
|
"""
|
|
|
metrics: Dict[str, float] = {}
|
|
|
if not os.path.exists(filepath):
|
|
|
return metrics
|
|
|
|
|
|
- # Try multiple encodings
|
|
|
content = None
|
|
|
for encoding in ("utf-8-sig", "utf-8", "gbk", "latin-1"):
|
|
|
try:
|
|
|
@@ -295,7 +733,6 @@ class RobustMotorCADSolver:
|
|
|
if content is None:
|
|
|
return metrics
|
|
|
|
|
|
- # Parse semicolon-separated lines
|
|
|
lines = content.splitlines()
|
|
|
for line in lines:
|
|
|
if ";" not in line:
|
|
|
@@ -304,7 +741,6 @@ class RobustMotorCADSolver:
|
|
|
if len(parts) < 2:
|
|
|
continue
|
|
|
field_name = parts[0].strip()
|
|
|
- # Try to find numeric value in remaining parts
|
|
|
value = None
|
|
|
for part in parts[1:]:
|
|
|
part = part.strip()
|
|
|
@@ -316,10 +752,8 @@ class RobustMotorCADSolver:
|
|
|
if value is None:
|
|
|
continue
|
|
|
|
|
|
- # Match against metric aliases
|
|
|
for metric_def in METRIC_DEFINITIONS:
|
|
|
if field_name in metric_def["aliases"]:
|
|
|
- # Only set if not already set (first match wins = E-Magnetics priority)
|
|
|
if metric_def["key"] not in metrics:
|
|
|
metrics[metric_def["key"]] = value
|
|
|
break
|
|
|
@@ -328,12 +762,10 @@ class RobustMotorCADSolver:
|
|
|
|
|
|
def _write_result_to_disk(self, result: Dict[str, Any]) -> None:
|
|
|
"""Write result to CSV and JSON immediately (flush + fsync)."""
|
|
|
- # CSV
|
|
|
if not self._csv_header_written:
|
|
|
header = ["point_index", "point_label", "status", "duration_s"]
|
|
|
for md in METRIC_DEFINITIONS:
|
|
|
header.append(md["key"])
|
|
|
- # Add param columns
|
|
|
if result["params"]:
|
|
|
for k in result["params"]:
|
|
|
if k not in ("point_index", "point_label"):
|
|
|
@@ -345,11 +777,8 @@ class RobustMotorCADSolver:
|
|
|
os.fsync(f.fileno())
|
|
|
self._csv_header_written = True
|
|
|
|
|
|
- # Append row
|
|
|
- row = [
|
|
|
- result["point_index"], result["point_label"],
|
|
|
- result["status"], result["duration_s"]
|
|
|
- ]
|
|
|
+ row = [result["point_index"], result["point_label"],
|
|
|
+ result["status"], result["duration_s"]]
|
|
|
for md in METRIC_DEFINITIONS:
|
|
|
row.append(result["metrics"].get(md["key"], ""))
|
|
|
if result["params"]:
|
|
|
@@ -362,7 +791,6 @@ class RobustMotorCADSolver:
|
|
|
f.flush()
|
|
|
os.fsync(f.fileno())
|
|
|
|
|
|
- # JSON (full results, overwritten each time)
|
|
|
with open(self._json_path, "w", encoding="utf-8") as f:
|
|
|
json.dump({"results": self._all_results}, f, ensure_ascii=False, indent=2)
|
|
|
f.flush()
|
|
|
@@ -372,9 +800,12 @@ class RobustMotorCADSolver:
|
|
|
"""Write timestamped log message."""
|
|
|
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
|
|
|
line = f"[{ts}] {message}\n"
|
|
|
- with open(self._log_path, "a", encoding="utf-8") as f:
|
|
|
- f.write(line)
|
|
|
- f.flush()
|
|
|
+ try:
|
|
|
+ with open(self._log_path, "a", encoding="utf-8") as f:
|
|
|
+ f.write(line)
|
|
|
+ f.flush()
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
|
|
|
def get_summary(self) -> Dict[str, Any]:
|
|
|
"""Get run summary."""
|
|
|
@@ -388,4 +819,5 @@ class RobustMotorCADSolver:
|
|
|
"csv_path": self._csv_path,
|
|
|
"json_path": self._json_path,
|
|
|
"log_path": self._log_path,
|
|
|
+ "headless_mode": self.headless,
|
|
|
}
|