"""Motor-CAD adapter - concrete SimulationAdapter implementation. Wraps RobustMotorCADSolver (scripts/robust_motorcad.py) behind the uniform SimulationAdapter protocol so the executor / GUI / future strategies never depend on Motor-CAD specifics. Registering this module (importing it) makes the "motorcad" tool available through afmcore.adapters.get_adapter("motorcad", ...). To add a new tool (e.g. Maxwell): create a sibling module implementing the same protocol and register it under its own tool name. Nothing in the executor changes. All source is ASCII only. """ from __future__ import annotations from typing import Any, Dict, Optional from . import SimulationAdapter, register_adapter class MotorCADAdapter(SimulationAdapter): """Adapter over RobustMotorCADSolver (reuses its full robustness protocol: write-back verification, baseline reload, popup suppression, per-point retry/reconnect, per-point disk write).""" tool_name = "motorcad" tool_label = "Motor-CAD (ANSYS)" capability_domains = ("electromagnetic",) def __init__( self, model_path: Optional[str] = None, output_dir: Optional[str] = None, point_timeout: int = 300, max_retries: int = 3, headless: bool = False, enable_thermal: bool = False, ambient_temperature: Optional[float] = None, log_cb=None, progress_cb=None, **kwargs: Any, ): super().__init__(log_cb=log_cb, progress_cb=progress_cb, **kwargs) self.model_path = model_path self.output_dir = output_dir self.point_timeout = point_timeout self.max_retries = max_retries self.headless = headless # P5-M6: after each EM solve, also run a steady-state thermal solve # and merge thermal metrics (OFF by default, preserves EM-only flow). self.enable_thermal = bool(enable_thermal) # P5-M6 thermal boundary: Ambient_Temperature override (degC); None = # leave model value. MARS ships 125 C, so pass 25-40 for valid results. self.ambient_temperature = ambient_temperature self._solver = None # lazy RobustMotorCADSolver # -- internal ---------------------------------------------------------- def _ensure_solver(self): """Lazily create the wrapped RobustMotorCADSolver instance.""" if self._solver is None: import os import sys # scripts/robust_motorcad.py is imported via the `scripts` package, # which requires the repo root on sys.path. Executor runs may only # have scripts/ and src/ on sys.path, so ensure the repo root is # present or `from scripts.robust_motorcad import ...` fails with # "No module named 'scripts'". _root = os.path.dirname( os.path.dirname( os.path.dirname(os.path.dirname(os.path.abspath(__file__))) ) ) if _root not in sys.path: sys.path.insert(0, _root) from scripts.robust_motorcad import RobustMotorCADSolver # type: ignore self._solver = RobustMotorCADSolver( model_path=self.model_path or "", output_dir=self.output_dir, point_timeout=self.point_timeout, max_retries=self.max_retries, headless=self.headless, enable_thermal=self.enable_thermal, ambient_temperature=self.ambient_temperature, ) return self._solver # -- lifecycle --------------------------------------------------------- def connect(self) -> None: self._ensure_solver().connect() self._log("Motor-CAD connected (MotorCADAdapter)") def disconnect(self) -> None: if self._solver is not None: try: self._solver.disconnect() finally: self._solver = None # -- model & parameters ------------------------------------------------ def load_model(self, model_path: str) -> None: # RobustMotorCADSolver reloads the baseline before every point, so we # only record the path here. self.model_path = model_path self._log("Model path set: %s" % model_path) def set_parameter(self, name: str, value: float) -> None: # Delegates to the robust write-back verification. self._ensure_solver()._write_and_verify(name, float(value)) def run_simulation(self, mode: str = "electromagnetic") -> None: solver = self._ensure_solver() if solver.mc is None: solver.connect() if mode == "electromagnetic": solver.mc.do_magnetic_calculation() else: raise ValueError( "MotorCADAdapter currently supports mode='electromagnetic' only" ) # -- results ----------------------------------------------------------- def extract_metrics(self, output_dir: str, tag: str = "") -> Dict[str, Any]: """Extract metrics from the most recent raw export written by the wrapped solver (it manages its own raw/ directory).""" solver = self._ensure_solver() metrics: Dict[str, float] = {} raw_path = "" error = "" status = "OK" try: import os raw_dir = os.path.join(solver.output_dir, "raw") if os.path.isdir(raw_dir): files = sorted( os.path.join(raw_dir, f) for f in os.listdir(raw_dir) ) if files: raw_path = files[-1] metrics = solver._parse_export(raw_path) if not metrics: status = "UNCERTAIN" error = "No metrics extracted from latest raw export" except Exception as exc: # noqa: BLE001 status = "FAILED" error = f"{type(exc).__name__}: {exc}" return {"metrics": metrics, "raw_path": raw_path, "status": status, "error": error} # -- high-level --------------------------------------------------------- def run_point( self, model_path: str, params: Optional[Dict[str, float]] = None, output_dir: str = "output", tag: str = "", ) -> Dict[str, Any]: """Run one point through the full robust protocol and map the result to the uniform adapter schema.""" import time started = time.time() solver = self._ensure_solver() # run_point is called with an explicit model_path; the solver is # created lazily with a possibly-empty default, so sync it here # or load_from_file would target the wrong (empty) path. if model_path: solver.model_path = model_path if solver.mc is None: try: solver.connect() except Exception as exc: # noqa: BLE001 return { "metrics": {}, "status": "FAILED", "error": f"{type(exc).__name__}: {exc}", "raw_path": "", "solve_time_s": round(time.time() - started, 1), "params": params or {}, } result = solver.run_single_point(params or {}, point_index=0) return { "metrics": result.get("metrics", {}), "status": result.get("status", "FAILED"), "error": result.get("error"), "raw_path": "", "solve_time_s": result.get("duration_s", round(time.time() - started, 1)), "params": params or {}, } # Register the adapter so get_adapter("motorcad", ...) works as soon as this # module is imported (the executor imports it at startup). register_adapter(MotorCADAdapter.tool_name, MotorCADAdapter)