pcb-afm-simulation-system/torqrippswap-master/torqrippswap/solver.py

"""Motor-CAD parameter scan solver core.

Pure computation module, no GUI dependencies.
Handles Motor-CAD connection, model reload, parameter write-back
verification, magnetic calculation, result export parsing, and
metric extraction with bilingual (English/Chinese) field matching.

All source is ASCII; Chinese field names use \\uXXXX escapes.
"""

from __future__ import annotations

import csv
import json
import math
import os
import subprocess
import time
import traceback
from datetime import datetime
from pathlib import Path


# ---------------------------------------------------------------------------
# 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": "back_emf_thd_pct",
        "label": "Back EMF THD [%]",
        "aliases": [
            "Harmonic Distortion Back EMF Line-Line Voltage",
            "\u7ebf\u53cd\u5411\u7535\u52a8\u52bf\u8c10\u6ce2",
            "\u7ebf\u7535\u538b\u8c10\u6ce2",
        ],
    },
    {
        "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)",
            "\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)",
            "\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)",
            "\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",
            "\u8f93\u51fa\u529f\u7387_\u7535\u538b\u9650\u5236\u9644\u8fd1\u5de5\u4f5c\u70b9",
        ],
    },
    {
        "key": "em_power_w",
        "label": "EM power [W]",
        "aliases": [
            "Electromagnetic Power",
            "\u7535\u78c1\u529f\u7387_\u7535\u538b\u9650\u5236\u9644\u8fd1\u5de5\u4f5c\u70b9",
        ],
    },
    {
        "key": "shaft_speed_rpm",
        "label": "Shaft speed [rpm]",
        "aliases": [
            "Shaft Speed",
            "\u8f6c\u901f[RPM]",
            "\u8f6c\u901f [RPM]",
        ],
    },
    {
        "key": "no_load_speed_rpm",
        "label": "No-load speed [rpm]",
        "aliases": [
            "No load speed",
            "\u7a7a\u8f7d\u8f6c\u901f",
        ],
    },
    {
        "key": "shaft_torque_nm",
        "label": "Shaft torque [Nm]",
        "aliases": [
            "Shaft Torque",
            "\u8f74\u8f6c\u77e9",
        ],
    },
]

METRIC_KEYS = [m["key"] for m in METRIC_DEFINITIONS]
METRIC_LABELS = {m["key"]: m["label"] for m in METRIC_DEFINITIONS}

# Section name aliases for priority ordering.
SECTION_PRIORITY = [
    "E-Magnetics",
    "\u7535\u78c1",
    "Drive",
    "\u9a71\u52a8",
    "Losses",
    "\u635f\u8017",
    "Materials",
    "\u6750\u6599",
    "Miscellaneous",
    "\u6742\u9879",
]


# ---------------------------------------------------------------------------
# Utility functions
# ---------------------------------------------------------------------------

def values_inclusive(start: float, stop: float, step: float) -> list[float]:
    """Generate values from start to stop inclusive, appending stop if
    it is not exactly reachable by integer steps."""
    if step <= 0 or stop < start:
        raise ValueError("step must be positive and stop must be >= start")
    count = int(math.floor((stop - start) / step + 1e-9))
    values = [round(start + i * step, 10) for i in range(count + 1)]
    if not math.isclose(values[-1], stop, abs_tol=1e-9):
        values.append(float(stop))
    return values


def _normalize_name(name: str) -> str:
    """Normalize a field name for matching: unify brackets, remove
    whitespace, lowercase."""
    s = name
    # Full-width brackets to half-width.
    s = s.replace("\uff08", "(").replace("\uff09", ")")
    # Remove all whitespace.
    s = "".join(s.split())
    return s.lower()


# Pre-normalize aliases for fast matching.
_METRIC_ALIAS_MAP: dict[str, str] = {}
for _m in METRIC_DEFINITIONS:
    for _alias in _m["aliases"]:
        _METRIC_ALIAS_MAP[_normalize_name(_alias)] = _m["key"]


def parse_export(path: Path) -> dict[str, dict[str, float]]:
    """Parse a Motor-CAD semicolon-delimited export file.

    Returns a dict of section_name -> {field_name: value}.
    Handles UTF-8, cp1252, gbk, and latin-1 encodings.
    """
    text = None
    for encoding in ("utf-8-sig", "gbk", "cp1252", "latin-1"):
        try:
            text = path.read_text(encoding=encoding)
            break
        except UnicodeDecodeError:
            continue
    if text is None:
        return {}
    result: dict[str, dict[str, float]] = {}
    section = "(root)"
    for raw in text.splitlines():
        line = raw.strip()
        if not line:
            continue
        if ";" not in line:
            section = line
            result.setdefault(section, {})
            continue
        parts = line.split(";")
        try:
            field = parts[0].strip().strip('"')
            value = float(parts[1])
            result.setdefault(section, {})[field] = value
        except (IndexError, ValueError):
            continue
    return result


def pick_metric(results: dict[str, dict[str, float]], metric_key: str):
    """Extract a metric value from parsed export results.

    Tries exact alias match first (section priority order), then
    falls back to prefix-based fuzzy match across all sections.
    Returns the value (float) or "" if not found.
    """
    # Build the set of normalized aliases for this metric.
    wanted_aliases = set()
    for m in METRIC_DEFINITIONS:
        if m["key"] == metric_key:
            for alias in m["aliases"]:
                wanted_aliases.add(_normalize_name(alias))
            break
    if not wanted_aliases:
        return ""

    # Phase 1: exact match in priority section order.
    for section_name in SECTION_PRIORITY:
        section = results.get(section_name)
        if section is None:
            continue
        for field, value in section.items():
            if _normalize_name(field) in wanted_aliases:
                return value

    # Phase 2: exact match across all sections.
    for section in results.values():
        for field, value in section.items():
            if _normalize_name(field) in wanted_aliases:
                return value

    # Phase 3: prefix fuzzy match.
    for alias_norm in wanted_aliases:
        for section in results.values():
            for field, value in section.items():
                field_norm = _normalize_name(field)
                if field_norm.startswith(alias_norm) or alias_norm.startswith(field_norm):
                    if len(field_norm) > 3:  # avoid trivial matches
                        return value
    return ""


def extract_all_metrics(results: dict[str, dict[str, float]]) -> dict[str, float]:
    """Extract all defined metrics from parsed results."""
    out: dict[str, float] = {}
    for m in METRIC_DEFINITIONS:
        val = pick_metric(results, m["key"])
        if val != "":
            out[m["key"]] = val
    return out


# ---------------------------------------------------------------------------
# Git preflight
# ---------------------------------------------------------------------------

def find_repo(start: Path) -> Path | None:
    """Walk up from start to find a directory containing .git."""
    for candidate in (start, *start.parents):
        if (candidate / ".git").exists():
            return candidate
    return None


def git_preflight(script_dir: Path) -> tuple[bool, str]:
    """Check that a git repo exists, HEAD is valid, and tracked files
    are clean. Returns (ok, commit_or_error)."""
    repo = find_repo(script_dir)
    if repo is None:
        return False, "No Git repository found."
    safe = f"safe.directory={repo.as_posix()}"
    base = ["git", "-c", safe]
    try:
        commit = subprocess.check_output(
            base + ["rev-parse", "--short", "HEAD"],
            cwd=repo, text=True, stderr=subprocess.STDOUT,
        ).strip()
        dirty = subprocess.check_output(
            base + ["status", "--porcelain", "--untracked-files=no"],
            cwd=repo, text=True, stderr=subprocess.STDOUT,
        ).strip()
    except (OSError, subprocess.CalledProcessError) as exc:
        return False, f"Git preflight failed: {exc}"
    if dirty:
        return False, "Tracked files have uncommitted changes. Commit before starting a scan."
    return True, commit


# ---------------------------------------------------------------------------
# Motor-CAD solver
# ---------------------------------------------------------------------------

class MotorCADSolver:
    """Manages a dedicated, foreground-visible Motor-CAD instance and
    runs parameter scans with per-point baseline reload, write-back
    verification, and immediate CSV/log persistence.

    No GUI dependencies. Callbacks (log_cb, progress_cb, row_cb) allow
    the caller to receive events.
    """

    def __init__(
        self,
        model_path: Path,
        log_cb=None,
        progress_cb=None,
        row_cb=None,
    ):
        self.model_path = Path(model_path).resolve()
        self.log_cb = log_cb
        self.progress_cb = progress_cb
        self.row_cb = row_cb
        self.mc = None
        self._cancel = False

    def _log(self, text: str) -> None:
        if self.log_cb:
            self.log_cb(text)

    def cancel(self) -> None:
        """Request cancellation after the current point finishes."""
        self._cancel = True

    def connect(self) -> None:
        """Open a dedicated, foreground-visible Motor-CAD instance.

        Falls back to set_motorcad_exe if MOTORCAD_ACTIVEX is not set.
        """
        import ansys.motorcad.core as pymotorcad

        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)
                    self._log(f"MOTORCAD_ACTIVEX not set; using {candidate}")
            except (ImportError, Exception):
                pass

        self._log("Opening a separate, visible Motor-CAD instance")
        self.mc = pymotorcad.MotorCAD(open_new_instance=True, keep_instance_open=False)
        self.mc.set_visible(True)
        self.mc.set_variable("MessageDisplayState", 2)
        self.mc.display_screen("Scripting")

    def disconnect(self) -> None:
        """Reload the baseline model and close the instance."""
        if self.mc is not None:
            try:
                self.mc.load_from_file(str(self.model_path))
            except Exception:
                pass
            try:
                self.mc.quit()
            except Exception:
                pass
            self.mc = None

    def _write_and_verify(self, variable: str, value: float) -> None:
        """Write a variable and read it back. Raises RuntimeError on
        mismatch."""
        self.mc.set_variable(variable, value)
        applied = float(self.mc.get_variable(variable))
        if not math.isclose(applied, value, rel_tol=1e-8, abs_tol=1e-7):
            raise RuntimeError(
                f"Write verification failed for {variable}: "
                f"wrote {value}, read {applied}"
            )

    def run_single_point(
        self,
        index: int,
        total: int,
        writes: list[tuple[str, float]],
        torque_points: int = 0,
        airgap_mesh: int = 0,
        raw_dir: Path | None = None,
    ) -> dict:
        """Run a single simulation point.

        Args:
            index: 1-based point index.
            total: total number of points.
            writes: list of (variable_name, value) to set, in order.
            torque_points: TorquePointsPerCycle (0 = keep model default).
            airgap_mesh: Airgap mesh/layers value (0 = keep model default).
            raw_dir: directory to save raw export CSV.

        Returns:
            dict with metrics, status, error, seconds.
        """
        started = time.time()
        result: dict = {
            "index": index,
            "status": "FAILED",
            "error": "",
            "seconds": 0,
            "metrics": {},
        }
        try:
            # Reload baseline model for every point.
            self.mc.load_from_file(str(self.model_path))
            self.mc.set_visible(True)
            self.mc.display_screen("Scripting")

            # Solver discretization (if specified).
            if torque_points > 0:
                self.mc.set_variable("TorquePointsPerCycle", torque_points)
            if airgap_mesh > 0:
                self.mc.set_variable("AirgapMeshPoints_mesh", airgap_mesh)
                self.mc.set_variable("AirgapMeshPoints_layers", airgap_mesh)

            # Write design variables and verify each.
            for variable, value in writes:
                self._write_and_verify(variable, value)

            self._log(f"[{index}/{total}] Starting magnetic calculation")
            self.mc.do_magnetic_calculation()

            # Export raw results.
            if raw_dir is not None:
                raw_dir.mkdir(parents=True, exist_ok=True)
                ts = datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:-3]
                val_tag = "_".join(f"{v}_{val:g}" for v, val in writes)
                raw_path = raw_dir / f"result_{index:04d}_{val_tag}_{ts}.csv"
                self.mc.export_results("EMagnetic", str(raw_path))
                parsed = parse_export(raw_path)
                result["metrics"] = extract_all_metrics(parsed)
                result["raw_path"] = str(raw_path)

            result["status"] = "OK"
        except Exception as exc:
            result["status"] = "FAILED"
            result["error"] = f"{type(exc).__name__}: {exc}"
            self._log(result["error"])
            self._log(traceback.format_exc())

        result["seconds"] = round(time.time() - started, 1)
        return result

    def run_scan(
        self,
        points: list[dict],
        output_dir: Path,
        torque_points: int = 0,
        airgap_mesh: int = 0,
        scan_name: str = "scan",
        extra_csv_fields: list[str] | None = None,
    ) -> dict:
        """Run a full parameter scan.

        Args:
            points: list of dicts, each with 'writes' (list of (var, val))
                    and optional extra fields for CSV.
            output_dir: base directory for run output.
            torque_points: solver setting.
            airgap_mesh: solver setting.
            scan_name: name for the run subdirectory.
            extra_csv_fields: additional CSV column names beyond metrics.

        Returns:
            dict with run_dir, csv_path, log_path, manifest_path, summary.
        """
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:-3]
        run_dir = Path(output_dir) / f"{timestamp}_{scan_name}"
        raw_dir = run_dir / "raw"
        run_dir.mkdir(parents=True, exist_ok=True)

        csv_path = run_dir / f"scan_results_{timestamp}.csv"
        log_path = run_dir / f"program_log_{timestamp}.log"
        manifest_path = run_dir / f"run_manifest_{timestamp}.json"

        extra_fields = extra_csv_fields or []
        csv_fields = ["run_index", "status", "seconds", "error"] + extra_fields + METRIC_KEYS

        log_file = log_path.open("a", encoding="ascii", errors="backslashreplace")

        def log(text: str) -> None:
            stamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
            line = f"{stamp} {text}"
            log_file.write(line + "\n")
            log_file.flush()
            self._log(line)

        # Write manifest.
        manifest = {
            "timestamp": timestamp,
            "model": str(self.model_path),
            "scan_name": scan_name,
            "total_points": len(points),
            "torque_points": torque_points,
            "airgap_mesh": airgap_mesh,
            "points": [
                {
                    "writes": [[v, val] for v, val in p.get("writes", [])],
                    **{k: p[k] for k in p if k != "writes"},
                }
                for p in points
            ],
        }
        manifest_path.write_text(json.dumps(manifest, indent=2), encoding="ascii")
        log(f"Run directory: {run_dir}")
        log(f"Model: {self.model_path}")
        log(f"Total points: {len(points)}")

        summary = {"ok": 0, "failed": 0, "results": []}

        try:
            with csv_path.open("w", newline="", encoding="utf-8-sig") as csv_file:
                writer = csv.DictWriter(csv_file, fieldnames=csv_fields)
                writer.writeheader()
                csv_file.flush()

                for idx, point in enumerate(points, 1):
                    if self._cancel:
                        log("Cancel requested; stopping before next point")
                        break

                    writes = point.get("writes", [])
                    log(
                        f"[{idx}/{len(points)}] "
                        + ", ".join(f"{v}={val:g}" for v, val in writes)
                    )

                    point_result = self.run_single_point(
                        index=idx,
                        total=len(points),
                        writes=writes,
                        torque_points=torque_points,
                        airgap_mesh=airgap_mesh,
                        raw_dir=raw_dir,
                    )

                    row = {field: "" for field in csv_fields}
                    row["run_index"] = idx
                    row["status"] = point_result["status"]
                    row["seconds"] = point_result["seconds"]
                    row["error"] = point_result["error"]
                    for key in extra_fields:
                        if key in point:
                            row[key] = point[key]
                    for key, value in point_result["metrics"].items():
                        row[key] = value

                    writer.writerow(row)
                    csv_file.flush()

                    if point_result["status"] == "OK":
                        summary["ok"] += 1
                    else:
                        summary["failed"] += 1
                    summary["results"].append(row)

                    if self.row_cb:
                        self.row_cb(row)
                    if self.progress_cb:
                        self.progress_cb(idx, len(points))

                    ripple = row.get("ripple_pct", "")
                    tavg = row.get("tavg_nm", "")
                    log(
                        f"[{idx}/{len(points)}] {row['status']} "
                        f"ripple={ripple} tavg={tavg} seconds={row['seconds']}"
                    )

            log(f"Scan ended. OK={summary['ok']} FAILED={summary['failed']}")
            log(f"Results: {csv_path}")
        finally:
            log_file.close()

        return {
            "run_dir": str(run_dir),
            "csv_path": str(csv_path),
            "log_path": str(log_path),
            "manifest_path": str(manifest_path),
            "summary": summary,
        }