pcb-afm-simulation-system/src/afmcore/adapters/motorcad.py

"""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 = "",
        thermal_mode: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Run one point through the full robust protocol and map the result
        to the uniform adapter schema.

        thermal_mode: per-task override ("off"/"steady"/"coupled"). None uses
        the solver default (self.thermal_mode or the legacy enable_thermal).
        """
        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, thermal_mode=thermal_mode
        )
        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)