"""Robust Motor-CAD simulation core (P4-M3 enhancement). Integrates all robustness practices from reference projects: - Connection: open_new_instance=True + set_visible(True) - 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 - 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 All source is ASCII only; Chinese field names use \\uXXXX escapes. """ from __future__ import annotations import csv import json import math import os import time import traceback from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional, Tuple # --------------------------------------------------------------------------- # Metric definitions: key, display label, and aliases (English + Chinese). # Chinese aliases use Unicode escapes so this file stays pure ASCII. # --------------------------------------------------------------------------- METRIC_DEFINITIONS = [ {"key": "ripple_pct", "label": "Torque Ripple [%]", "aliases": ["Torque Ripple (VW) [%]", "Torque Ripple (VW)[%]"]}, {"key": "ripple_nm", "label": "Torque Ripple [Nm]", "aliases": ["Torque Ripple (VW)"]}, {"key": "tavg_nm", "label": "Tavg VW [Nm]", "aliases": ["Average torque (virtual work)", "\u5e73\u5747\u8f6c\u77e9 (virtual work)", "\u5e73\u5747\u8f6c\u77e9(virtual work)"]}, {"key": "efficiency_pct", "label": "Efficiency [%]", "aliases": ["System Efficiency", "\u7cfb\u7edf\u6548\u7387"]}, {"key": "back_emf_v", "label": "Back EMF LL rms [V]", "aliases": ["Back EMF Line-Line Voltage (rms)", "\u7ebf\u95f4\u53cd\u5411\u7535\u52a8\u52bf\u6709\u6548\u503c"]}, {"key": "total_losses_w", "label": "Total losses [W]", "aliases": ["Total Losses (on load)", "\u603b\u635f\u8017(\u989d\u5b9a)", "\u603b\u635f\u8017 (\u989d\u5b9a)"]}, {"key": "copper_loss_w", "label": "DC copper loss [W]", "aliases": ["Armature DC Copper Loss (on load)", "\u7535\u67a2\u76f4\u6d41\u94dc\u8017(\u5e26\u8f7d)"]}, {"key": "magnet_loss_w", "label": "Magnet loss [W]", "aliases": ["Magnet Loss (on load)", "\u6c38\u78c1\u4f53\u635f\u8017(\u989d\u5b9a)"]}, {"key": "iron_loss_w", "label": "Stator iron loss [W]", "aliases": ["Stator iron Loss [total] (on load)", "\u5b9a\u5b50\u94c1\u635f[\u603b\u635f\u8017](\u989d\u5b9a)"]}, {"key": "input_power_w", "label": "Input power [W]", "aliases": ["Input Power", "\u8f93\u5165\u529f\u7387"]}, {"key": "output_power_w", "label": "Output power [W]", "aliases": ["Output Power"]}, {"key": "shaft_speed_rpm", "label": "Shaft speed [rpm]", "aliases": ["Shaft Speed", "\u8f6c\u901f[RPM]"]}, ] # Known incompatible sampling point / mesh combinations that cause popups INCOMPATIBLE_SAMPLING_MESH = [ (120, 840), # Motor-CAD warns mesh/time step mismatch, blocks batch ] # Recommended compatible combinations RECOMMENDED_SAMPLING_MESH = [ (30, 840), # Fast trend scan (120, 960), # Medium confidence (180, 1680), # High confidence final ] def ensure_environment() -> None: """Ensure Motor-CAD environment variables are set (non-login shell trap).""" if not os.environ.get("MOTORCAD_ACTIVEX"): try: from ansys.motorcad.core import set_motorcad_exe candidate = r"D:\Program Files\ANSYS Inc\v261\motorcad\MotorCAD.exe" if os.path.exists(candidate): set_motorcad_exe(candidate) except Exception: pass if not os.environ.get("ANSYSLMD_LICENSE_FILE"): os.environ["ANSYSLMD_LICENSE_FILE"] = "1055@localhost" def check_sampling_mesh_compatibility(torque_points: int, airgap_mesh: int) -> Tuple[bool, str]: """Check if sampling point / mesh combination is compatible. Returns (compatible, message). Incompatible combinations cause Motor-CAD popups that block unattended batch execution. """ for pts, mesh in INCOMPATIBLE_SAMPLING_MESH: if torque_points == pts and airgap_mesh == mesh: return False, ( f"TorquePoints={torque_points} + AirgapMesh={airgap_mesh} " f"causes Motor-CAD popup. Use {RECOMMENDED_SAMPLING_MESH[1]} instead." ) return True, "OK" def compute_copper_width(slot_opening_mm: float, clearance_mm: float = 0.2, conductor_count: int = 1) -> float: """Compute PCB copper width from slot opening (linkage formula). Copper_Width = (Slot_Opening - clearance) / 2 / conductor_count """ return round((slot_opening_mm - clearance_mm) / 2.0 / conductor_count, 3) 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) solver.disconnect() """ def __init__(self, model_path: str, output_dir: Optional[str] = None, point_timeout: int = 300, max_retries: int = 3): self.model_path = model_path self.output_dir = output_dir or os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "output", f"run_{datetime.now().strftime('%Y%m%d_%H%M%S')}" ) self.raw_dir = os.path.join(self.output_dir, "raw") os.makedirs(self.raw_dir, exist_ok=True) self.point_timeout = point_timeout self.max_retries = max_retries self.mc = None 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") self._all_results: List[Dict[str, Any]] = [] self._csv_header_written = False def connect(self) -> None: """Connect to a new Motor-CAD instance (never connect to existing).""" 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) time.sleep(2) # Wait for instance to fully initialize # Health check _ = self.mc.get_variable("Motor_Type") self._log("Connected to new Motor-CAD instance") except Exception as e: self._log(f"Connection failed: {e}") raise def disconnect(self) -> None: """Disconnect from Motor-CAD instance.""" if self.mc: try: # Reload baseline to leave clean state self.mc.load_from_file(self.model_path) except Exception: pass try: self.mc.quit() except Exception: pass self.mc = None self._log("Disconnected from Motor-CAD") 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)) 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}" ) return applied 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 """ start_time = time.time() result = { "point_index": point_index, "point_label": point_label, "params": params, "status": "pending", "metrics": {}, "error": None, "duration_s": 0, } for attempt in range(self.max_retries): try: # Step 1: Reload baseline self.mc.load_from_file(self.model_path) # 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}") # 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" result["duration_s"] = round(time.time() - start_time, 2) self._all_results.append(result) self._write_result_to_disk(result) return result 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. """ 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: with open(filepath, "r", encoding=encoding) as f: content = f.read() break except (UnicodeDecodeError, Exception): continue if content is None: return metrics # Parse semicolon-separated lines lines = content.splitlines() for line in lines: if ";" not in line: continue parts = line.split(";") 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() try: value = float(part.replace(",", ".")) break except (ValueError, Exception): continue 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 return metrics 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"): header.append(f"param_{k}") with open(self._csv_path, "w", newline="", encoding="utf-8") as f: writer = csv.writer(f, delimiter=";") writer.writerow(header) f.flush() os.fsync(f.fileno()) self._csv_header_written = True # Append row 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"]: for k, v in result["params"].items(): if k not in ("point_index", "point_label"): row.append(v) with open(self._csv_path, "a", newline="", encoding="utf-8") as f: writer = csv.writer(f, delimiter=";") writer.writerow(row) 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() os.fsync(f.fileno()) def _log(self, message: str) -> None: """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() def get_summary(self) -> Dict[str, Any]: """Get run summary.""" ok = [r for r in self._all_results if r["status"] == "ok"] failed = [r for r in self._all_results if r["status"] == "failed"] return { "total": len(self._all_results), "ok": len(ok), "failed": len(failed), "output_dir": self.output_dir, "csv_path": self._csv_path, "json_path": self._json_path, "log_path": self._log_path, }