pcb-afm-simulation-system/scripts/robust_motorcad.py

"""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,
        }