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

"""Robust Motor-CAD simulation core (P4-M3 + reference doc enhancement).

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 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.
"""
from __future__ import annotations

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).
# 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)[%]",
                  "Torque Ripple (VW) %"]},
    {"key": "ripple_nm", "label": "Torque Ripple [Nm]",
     "aliases": ["Torque Ripple (VW)"]},
    {"key": "tavg_nm", "label": "Average Torque VW [Nm]",
     "aliases": ["Average torque (virtual work)",
                  "\u5e73\u5747\u8f6c\u77e9 (virtual work)",
                  "\u5e73\u5747\u8f6c\u77e9(virtual work)",
                  "\u5e73\u5747\u8f6c\u77e9 (DQ)",
                  "\u5e73\u5747\u8f6c\u77e9(DQ)",
                  "\u5e73\u5747\u8f6c\u77e9 (loop torque)",
                  "\u5e73\u5747\u8f6c\u77e9(loop torque)"]},
    {"key": "shaft_torque_nm", "label": "Shaft Torque [Nm]",
     "aliases": ["Shaft Torque", "\u8f74\u8f6c\u77e9"]},
    {"key": "stall_torque_nm", "label": "Stall Torque [Nm]",
     "aliases": ["Stall Torque", "\u5835\u8f6c\u8f6c\u77e9"]},
    {"key": "torque_constant", "label": "Torque Constant Kt [Nm/A]",
     "aliases": ["Torque Constant (Kt)", "\u8f6c\u77e9\u5e38\u6570(Kt)",
                  "\u8f6c\u77e9\u5e38\u6570\uff08Kt\uff09"]},
    {"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",
                  "\u7ebf\u95f4\u53cd\u5411\u7535\u52a8\u52bf\u5e45\u503c"]},
    {"key": "total_losses_w", "label": "Total losses [W]",
     "aliases": ["Total Losses (on load)", "\u603b\u635f\u8017(\u989d\u5b9a)",
                  "\u603b\u635f\u8017 (\u989d\u5b9a)", "\u603b\u635f\u8017(\u7a7a\u8f7d)"]},
    {"key": "copper_loss_w", "label": "DC copper loss [W]",
     "aliases": ["Armature DC Copper Loss (on load)",
                  "\u7535\u67a2\u76f4\u6d41\u94dc\u8017(\u5e26\u8f7d)",
                  "\u7535\u67a2\u76f4\u6d41\u94dc\u8017 \uff08\u5e26\u8f7d\uff09",
                  "\u7535\u67a2\u76f4\u6d41\u94dc\u8017 \uff08\u7a7a\u8f7d\uff09"]},
    {"key": "magnet_loss_w", "label": "Magnet loss [W]",
     "aliases": ["Magnet Loss (on load)",
                  "\u6c38\u78c1\u4f53\u635f\u8017(\u989d\u5b9a)",
                  "\u6c38\u78c1\u4f53\u635f\u8017 (\u989d\u5b9a)",
                  "\u6c38\u78c1\u4f53\u635f\u8017(\u7a7a\u8f7d)"]},
    {"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)",
                  "\u5b9a\u5b50\u94c1\u635f[\u603b\u635f\u8017] (\u989d\u5b9a)",
                  "\u5b9a\u5b50\u94c1\u635f[\u603b\u635f\u8017](\u7a7a\u8f7d)"]},
    {"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", "\u8f93\u51fa\u529f\u7387"]},
    {"key": "em_power_w", "label": "EM power [W]",
     "aliases": ["Electromagnetic Power", "\u7535\u78c1\u529f\u7387"]},
    {"key": "shaft_speed_rpm", "label": "Shaft speed [rpm]",
     "aliases": ["Shaft Speed", "\u8f6c\u901f[RPM]", "\u8f6c\u901f\uff5bRPM\uff5d",
                  "\u7a7a\u8f7d\u8f6c\u901f"]},
    {"key": "phase_current_peak_a", "label": "Phase current peak [A]",
     "aliases": ["Peak Phase Current", "\u76f8\u7535\u6d41\u5cf0\u503c"]},
    {"key": "line_current_rms_a", "label": "Line current rms [A]",
     "aliases": ["Line Current (rms)", "\u7ebf\u7535\u6d41 (\u6709\u6548\u503c)",
                  "\u7ebf\u7535\u6d41(\u6709\u6548\u503c)", "\u76f8\u7535\u6d41\u6709\u6548\u503c"]},
]

# 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
]

# ---------------------------------------------------------------------------
# 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).

    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
            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.
    Reference: MOTORCAD_SCAN_KNOWLEDGE_BASE.md section 6.2
    """
    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
    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()
        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,
                 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__))),
            "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.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")
        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).

        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, 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: verify connection is responsive
            _ = self.mc.get_variable("Motor_Type")
            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.

        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)
            except Exception:
                pass
            try:
                self.mc.quit()
            except Exception:
                pass
            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.

        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 {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. 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 = {
            "point_index": point_index,
            "point_label": point_label,
            "params": params,
            "status": "pending",
            "metrics": {},
            "error": None,
            "duration_s": 0,
        }

        # Suppress popups for batch execution
        self._suppress_popups()

        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"
                            # B5 fix: break instead of return so the point
                            # is appended to results and written to disk
                            break

                    # 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"
                    )
                    try:
                        self.mc.export_results("EMagnetic", 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

        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

        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()
            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

            for metric_def in METRIC_DEFINITIONS:
                if field_name in metric_def["aliases"]:
                    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)."""
        if not self._csv_header_written:
            header = ["point_index", "point_label", "status", "duration_s"]
            for md in METRIC_DEFINITIONS:
                header.append(md["key"])
            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

        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())

        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"
        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."""
        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,
            "headless_mode": self.headless,
        }