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

"""Tkinter GUI for timestamped Motor-CAD parameter scans."""

from __future__ import annotations

import csv
import json
import math
import os
import queue
import subprocess
import sys
import threading
import time
import traceback
from datetime import datetime
from pathlib import Path
import tkinter as tk
from tkinter import filedialog, messagebox, ttk

from scan_central_mag_arc import parse_export, pick, values_inclusive

try:
    from build_info import BUILD_COMMIT
except ImportError:
    BUILD_COMMIT = "source-tree"


APP_TITLE = "Motor-CAD Parameter Scan"
DEFAULT_MODEL_NAME = "MARS-9S8P_SSSR_Halbach_PCB-V1.0.mot"
PROFILE_FILE = "scan_parameters.json"
METRICS = [
    ("ripple_pct", "Torque Ripple [%]", "Torque Ripple (VW) [%]"),
    ("ripple_nm", "Torque Ripple [Nm]", "Torque Ripple (VW)"),
    ("tavg_nm", "Tavg VW [Nm]", "Average torque (virtual work)"),
    ("efficiency_pct", "Efficiency [%]", "System Efficiency"),
    ("back_emf_v", "Back EMF LL rms [V]", "Back EMF Line-Line Voltage (rms)"),
    ("back_emf_thd_pct", "Back EMF THD [%]", "Harmonic Distortion Back EMF Line-Line Voltage"),
    ("total_losses_w", "Total losses [W]", "Total Losses (on load)"),
    ("copper_loss_w", "DC copper loss [W]", "Armature DC Copper Loss (on load)"),
    ("magnet_loss_w", "Magnet loss [W]", "Magnet Loss (on load)"),
    ("iron_loss_w", "Stator iron loss [W]", "Stator iron Loss [total] (on load)"),
    ("input_power_w", "Input power [W]", "Input Power"),
    ("output_power_w", "Output power [W]", "Output Power"),
    ("em_power_w", "EM power [W]", "Electromagnetic Power"),
    ("shaft_speed_rpm", "Shaft speed [rpm]", "Shaft Speed"),
    ("no_load_speed_rpm", "No-load speed [rpm]", "No load speed"),
]
CSV_FIELDS = ["run_index", "parameter_1_label", "parameter_1_variable", "parameter_1_value",
              "parameter_2_label", "parameter_2_variable", "parameter_2_value",
              "linked_variable", "linked_value", "conductor_count", "clearance_mm"] + [
    item[0] for item in METRICS
] + ["seconds", "status", "error", "timestamp"]


def app_root() -> Path:
    if getattr(sys, "frozen", False):
        return Path(sys.executable).resolve().parent
    return Path(__file__).resolve().parent


def bundled_root() -> Path:
    return Path(getattr(sys, "_MEIPASS", app_root()))


def default_model_path() -> Path:
    for base in (app_root(), *app_root().parents):
        candidate = base / DEFAULT_MODEL_NAME
        if candidate.exists():
            return candidate
    return app_root() / DEFAULT_MODEL_NAME


def load_profiles() -> dict:
    external = app_root() / PROFILE_FILE
    bundled = bundled_root() / PROFILE_FILE
    path = external if external.exists() else bundled
    return json.loads(path.read_text(encoding="ascii"))


def find_repo(start: Path) -> Path | None:
    for candidate in (start, *start.parents):
        if (candidate / ".git").exists():
            return candidate
    return None


def git_preflight() -> tuple[bool, str]:
    repo = find_repo(app_root())
    if repo is None:
        if BUILD_COMMIT not in ("", "source-tree", "unknown"):
            return True, BUILD_COMMIT
        return False, "No Git repository or embedded build commit was 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


def metric_value(results: dict, key: str):
    value = pick(results, key)
    if value != "":
        return value
    wanted = key.lower().replace(" ", "")
    for section in results.values():
        for name, candidate in section.items():
            normalized = name.lower().replace(" ", "")
            if normalized.startswith(wanted):
                return candidate
    return ""


class ScanWorker(threading.Thread):
    def __init__(self, config: dict, events: queue.Queue, cancel: threading.Event):
        super().__init__(daemon=True)
        self.config = config
        self.events = events
        self.cancel = cancel

    def emit(self, kind: str, payload=None) -> None:
        self.events.put((kind, payload))

    def run(self) -> None:
        mc = None
        log_file = None
        try:
            import ansys.motorcad.core as pymotorcad

            run_dir = Path(self.config["run_dir"])
            raw_dir = run_dir / "raw"
            raw_dir.mkdir(parents=True, exist_ok=True)
            csv_path = run_dir / f"scan_results_{self.config['timestamp']}.csv"
            log_path = run_dir / f"program_log_{self.config['timestamp']}.log"
            manifest_path = run_dir / f"run_manifest_{self.config['timestamp']}.json"
            manifest_path.write_text(json.dumps(self.config, indent=2), encoding="ascii")
            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.emit("log", line)

            points = self.config["points"]
            log(f"Run directory: {run_dir}")
            log(f"Git commit: {self.config['git_commit']}")
            log("Opening a separate Motor-CAD instance")
            mc = pymotorcad.MotorCAD(open_new_instance=True, keep_instance_open=False)
            mc.set_visible(True)
            mc.set_variable("MessageDisplayState", 2)
            mc.display_screen("Scripting")

            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 index, point in enumerate(points, 1):
                    if self.cancel.is_set():
                        log("Cancel requested; stopping before next point")
                        break
                    started = time.time()
                    row = {field: "" for field in CSV_FIELDS}
                    row.update(
                        run_index=index,
                        parameter_1_label=self.config["parameter_1"]["label"],
                        parameter_1_variable=self.config["parameter_1"]["variable"],
                        parameter_1_value=point["parameter_1_value"],
                        parameter_2_label=self.config["parameter_2"]["label"],
                        parameter_2_variable=self.config["parameter_2"]["variable"],
                        parameter_2_value=point["parameter_2_value"],
                        linked_variable=self.config["linked_value"]["variable"],
                        linked_value=point["linked_value"],
                        conductor_count=self.config["conductor_count"],
                        clearance_mm=self.config["clearance"],
                        status="FAILED",
                        timestamp=datetime.now().isoformat(timespec="milliseconds"),
                    )
                    log(f"[{index}/{len(points)}] Starting "
                        f"{self.config['parameter_1']['variable']}={point['parameter_1_value']:g}, "
                        f"{self.config['parameter_2']['variable']}={point['parameter_2_value']:g}, "
                        f"{self.config['linked_value']['variable']}={point['linked_value']:g}")
                    try:
                        mc.load_from_file(self.config["model"])
                        mc.set_visible(True)
                        mc.display_screen("Scripting")
                        if self.config["torque_points"] > 0:
                            mc.set_variable("TorquePointsPerCycle", self.config["torque_points"])
                        if self.config["airgap_mesh"] > 0:
                            mc.set_variable("AirgapMeshPoints_mesh", self.config["airgap_mesh"])
                            mc.set_variable("AirgapMeshPoints_layers", self.config["airgap_mesh"])
                        writes = (
                            (self.config["parameter_2"]["variable"], point["parameter_2_value"]),
                            (self.config["linked_value"]["variable"], point["linked_value"]),
                            (self.config["parameter_1"]["variable"], point["parameter_1_value"]),
                        )
                        for variable, value in writes:
                            mc.set_variable(variable, value)
                        for variable, value in writes:
                            applied = float(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}: wrote {value}, read {applied}"
                                )
                        mc.do_magnetic_calculation()
                        raw_path = raw_dir / (
                            f"result_{index:04d}_p1_{point['parameter_1_value']:g}_"
                            f"p2_{point['parameter_2_value']:g}_{self.config['timestamp']}.csv"
                        )
                        mc.export_results("EMagnetic", str(raw_path))
                        results = parse_export(raw_path)
                        for field, _, result_name in METRICS:
                            row[field] = metric_value(results, result_name)
                        if row["ripple_pct"] == "":
                            raise RuntimeError("Torque Ripple (VW) [%] was not found")
                        row["status"] = "OK"
                    except Exception as exc:
                        row["error"] = f"{type(exc).__name__}: {exc}"
                        log(row["error"])
                        log(traceback.format_exc())
                    row["seconds"] = round(time.time() - started, 1)
                    writer.writerow(row)
                    csv_file.flush()
                    self.emit("row", row)
                    self.emit("progress", (index, len(points)))
                    log(f"[{index}/{len(points)}] {row['status']} ripple={row['ripple_pct']} "
                        f"tavg={row['tavg_nm']} seconds={row['seconds']}")

            log(f"Scan ended. Results: {csv_path}")
            self.emit("done", {"run_dir": str(run_dir), "csv": str(csv_path), "log": str(log_path)})
        except Exception as exc:
            self.emit("fatal", f"{type(exc).__name__}: {exc}\n{traceback.format_exc()}")
        finally:
            if mc is not None:
                try:
                    mc.load_from_file(self.config["model"])
                except Exception:
                    pass
            if log_file is not None:
                log_file.close()


class MotorCADScanApp(tk.Tk):
    def __init__(self):
        super().__init__()
        self.title(APP_TITLE)
        self.geometry("1500x850")
        self.minsize(1050, 650)
        self.profiles = load_profiles()
        self.events: queue.Queue = queue.Queue()
        self.cancel_event = threading.Event()
        self.worker: ScanWorker | None = None
        self.vars = {
            "model": tk.StringVar(value=str(default_model_path())),
            "profile": tk.StringVar(value=next(iter(self.profiles))),
            "p1_start": tk.StringVar(),
            "p1_stop": tk.StringVar(),
            "p1_step": tk.StringVar(),
            "p2_start": tk.StringVar(),
            "p2_stop": tk.StringVar(),
            "p2_step": tk.StringVar(),
            "conductor_count": tk.StringVar(),
            "clearance": tk.StringVar(),
            "total_runs": tk.StringVar(value="Total combinations: 0"),
            "torque_points": tk.StringVar(value="30"),
            "airgap_mesh": tk.StringVar(value="840"),
            "status": tk.StringVar(value="Ready"),
            "commit": tk.StringVar(value=BUILD_COMMIT),
        }
        self.build_ui()
        self.profile_changed()
        self.after(150, self.poll_events)

    def build_ui(self) -> None:
        controls = ttk.LabelFrame(self, text="Scan setup", padding=8)
        controls.pack(fill="x", padx=8, pady=6)
        ttk.Label(controls, text="Model").grid(row=0, column=0, sticky="w")
        ttk.Entry(controls, textvariable=self.vars["model"], width=100).grid(row=0, column=1, sticky="ew", padx=4)
        ttk.Button(controls, text="Browse", command=self.browse_model).grid(row=0, column=2, padx=4)

        ttk.Label(controls, text="Scan profile").grid(row=1, column=0, sticky="w")
        combo = ttk.Combobox(controls, textvariable=self.vars["profile"], values=list(self.profiles), state="readonly", width=32)
        combo.grid(row=1, column=1, sticky="w", padx=4)
        combo.bind("<<ComboboxSelected>>", lambda _: self.profile_changed())

        p1_frame = ttk.LabelFrame(controls, text="Parameter 1: Central Mag Arc [ED]", padding=4)
        p1_frame.grid(row=2, column=1, columnspan=2, sticky="w", padx=4, pady=(6, 2))
        for col, (label, key) in enumerate((("Start", "p1_start"), ("Stop", "p1_stop"), ("Step", "p1_step"))):
            group = ttk.Frame(p1_frame)
            group.grid(row=0, column=col, padx=(0, 18), sticky="w")
            ttk.Label(group, text=label, width=6).pack(side="left")
            entry = ttk.Entry(group, textvariable=self.vars[key], width=12)
            entry.pack(side="left")
            entry.bind("<KeyRelease>", lambda _: self.update_total())

        p2_frame = ttk.LabelFrame(controls, text="Parameter 2: Slot Opening [mm]", padding=4)
        p2_frame.grid(row=3, column=1, columnspan=2, sticky="w", padx=4, pady=2)
        for col, (label, key) in enumerate((("Start", "p2_start"), ("Stop", "p2_stop"), ("Step", "p2_step"))):
            group = ttk.Frame(p2_frame)
            group.grid(row=0, column=col, padx=(0, 18), sticky="w")
            ttk.Label(group, text=label, width=6).pack(side="left")
            entry = ttk.Entry(group, textvariable=self.vars[key], width=12)
            entry.pack(side="left")
            entry.bind("<KeyRelease>", lambda _: self.update_total())

        link_frame = ttk.LabelFrame(controls, text="Linked copper width", padding=4)
        link_frame.grid(row=4, column=1, columnspan=2, sticky="w", padx=4, pady=2)
        for col, (label, key) in enumerate((("Conductors per side", "conductor_count"),
                                            ("Clearance [mm]", "clearance"))):
            group = ttk.Frame(link_frame)
            group.grid(row=0, column=col, padx=(0, 24), sticky="w")
            ttk.Label(group, text=label, width=20).pack(side="left")
            ttk.Entry(group, textvariable=self.vars[key], width=12).pack(side="left")
        ttk.Label(link_frame, text="Copper Width = (Slot Opening - Clearance) / 2 / Conductors").grid(
            row=1, column=0, columnspan=2, sticky="w", pady=(5, 0)
        )

        solver_frame = ttk.Frame(controls)
        solver_frame.grid(row=5, column=1, columnspan=2, sticky="w", padx=4, pady=(6, 2))
        for col, (label, key) in enumerate((("Torque points", "torque_points"),
                                            ("Airgap mesh", "airgap_mesh"))):
            group = ttk.Frame(solver_frame)
            group.grid(row=0, column=col, padx=(0, 28), sticky="w")
            ttk.Label(group, text=label, width=15).pack(side="left")
            ttk.Entry(group, textvariable=self.vars[key], width=12).pack(side="left")
        ttk.Label(solver_frame, textvariable=self.vars["total_runs"]).grid(row=1, column=0, columnspan=2, sticky="w", pady=(6, 0))
        controls.columnconfigure(1, weight=1)

        actions = ttk.Frame(self, padding=(8, 2))
        actions.pack(fill="x")
        self.start_button = ttk.Button(actions, text="Start scan", command=self.start_scan)
        self.start_button.pack(side="left")
        self.cancel_button = ttk.Button(actions, text="Stop after current point", command=self.cancel_scan, state="disabled")
        self.cancel_button.pack(side="left", padx=6)
        ttk.Label(actions, textvariable=self.vars["status"]).pack(side="left", padx=12)
        git_frame = ttk.Frame(actions)
        git_frame.pack(side="right")
        ttk.Label(git_frame, text="Git:").pack(side="left")
        ttk.Label(git_frame, textvariable=self.vars["commit"]).pack(side="left", padx=4)

        self.progress = ttk.Progressbar(self, mode="determinate")
        self.progress.pack(fill="x", padx=8, pady=4)

        table_frame = ttk.Frame(self)
        table_frame.pack(fill="both", expand=True, padx=8, pady=4)
        columns = ["run_index", "parameter_1_value", "parameter_2_value", "linked_value"] + [item[0] for item in METRICS] + ["seconds", "status"]
        self.tree = ttk.Treeview(table_frame, columns=columns, show="headings", height=16)
        headings = {"run_index": "#", "parameter_1_value": "Central Arc [ED]",
                    "parameter_2_value": "Slot Opening [mm]", "linked_value": "Copper Width [mm]",
                    "seconds": "Seconds", "status": "Status"}
        headings.update({field: label for field, label, _ in METRICS})
        for column in columns:
            self.tree.heading(column, text=headings[column])
            self.tree.column(column, width=105, anchor="center", stretch=False)
        self.tree.column("run_index", width=45)
        self.tree.column("status", width=70)
        ybar = ttk.Scrollbar(table_frame, orient="vertical", command=self.tree.yview)
        xbar = ttk.Scrollbar(table_frame, orient="horizontal", command=self.tree.xview)
        self.tree.configure(yscrollcommand=ybar.set, xscrollcommand=xbar.set)
        self.tree.grid(row=0, column=0, sticky="nsew")
        ybar.grid(row=0, column=1, sticky="ns")
        xbar.grid(row=1, column=0, sticky="ew")
        table_frame.rowconfigure(0, weight=1)
        table_frame.columnconfigure(0, weight=1)

        log_frame = ttk.LabelFrame(self, text="Program log", padding=4)
        log_frame.pack(fill="both", expand=False, padx=8, pady=(2, 8))
        self.log_text = tk.Text(log_frame, height=9, wrap="none")
        self.log_text.pack(fill="both", expand=True)

    def browse_model(self) -> None:
        path = filedialog.askopenfilename(title="Select Motor-CAD model", filetypes=[("Motor-CAD model", "*.mot"), ("All files", "*.*")])
        if path:
            self.vars["model"].set(path)

    def profile_changed(self) -> None:
        profile = self.profiles[self.vars["profile"].get()]
        p1 = profile["parameter_1"]
        p2 = profile["parameter_2"]
        linked = profile["linked_value"]
        for prefix, parameter in (("p1", p1), ("p2", p2)):
            self.vars[f"{prefix}_start"].set(str(parameter["default_start"]))
            self.vars[f"{prefix}_stop"].set(str(parameter["default_stop"]))
            self.vars[f"{prefix}_step"].set(str(parameter["default_step"]))
        self.vars["conductor_count"].set(str(linked["default_conductor_count"]))
        self.vars["clearance"].set(str(linked["default_clearance"]))
        self.update_total()

    def update_total(self) -> None:
        try:
            p1_values = values_inclusive(float(self.vars["p1_start"].get()),
                                         float(self.vars["p1_stop"].get()),
                                         float(self.vars["p1_step"].get()))
            p2_values = values_inclusive(float(self.vars["p2_start"].get()),
                                         float(self.vars["p2_stop"].get()),
                                         float(self.vars["p2_step"].get()))
            self.vars["total_runs"].set(
                f"Total combinations: {len(p1_values)} x {len(p2_values)} = {len(p1_values) * len(p2_values)}"
            )
        except ValueError:
            self.vars["total_runs"].set("Total combinations: invalid range")

    def append_log(self, text: str) -> None:
        self.log_text.insert("end", text + "\n")
        self.log_text.see("end")

    def start_scan(self) -> None:
        ok, commit = git_preflight()
        if not ok:
            messagebox.showerror("Git preflight failed", commit)
            return
        try:
            model = Path(self.vars["model"].get()).resolve()
            if not model.exists():
                raise ValueError(f"Model does not exist: {model}")
            p1_start = float(self.vars["p1_start"].get())
            p1_stop = float(self.vars["p1_stop"].get())
            p1_step = float(self.vars["p1_step"].get())
            p2_start = float(self.vars["p2_start"].get())
            p2_stop = float(self.vars["p2_stop"].get())
            p2_step = float(self.vars["p2_step"].get())
            conductor_count = int(self.vars["conductor_count"].get())
            clearance = float(self.vars["clearance"].get())
            torque_points = int(self.vars["torque_points"].get())
            airgap_mesh = int(self.vars["airgap_mesh"].get())
            p1_values = values_inclusive(p1_start, p1_stop, p1_step)
            p2_values = values_inclusive(p2_start, p2_stop, p2_step)
            profile = self.profiles[self.vars["profile"].get()]
            p1 = profile["parameter_1"]
            p2 = profile["parameter_2"]
            linked = profile["linked_value"]
            if p1_values[0] < p1["minimum"] or p1_values[-1] > p1["maximum"]:
                raise ValueError(f"Parameter 1 must stay within {p1['minimum']}..{p1['maximum']} {p1['unit']}")
            if p2_values[0] < p2["minimum"] or p2_values[-1] > p2["maximum"]:
                raise ValueError(f"Parameter 2 must stay within {p2['minimum']}..{p2['maximum']} {p2['unit']}")
            if conductor_count < 1:
                raise ValueError("Conductor count must be at least 1")
            if clearance < 0 or clearance >= p2_values[0]:
                raise ValueError("Clearance must be non-negative and smaller than every slot opening")
            points = []
            for slot_opening in p2_values:
                copper_width = round((slot_opening - clearance) / 2.0 / conductor_count, 10)
                for magnet_arc in p1_values:
                    points.append({
                        "parameter_1_value": magnet_arc,
                        "parameter_2_value": slot_opening,
                        "linked_value": copper_width,
                    })
        except ValueError as exc:
            messagebox.showerror("Invalid settings", str(exc))
            return

        estimated_hours = len(points) * 180.0 / 3600.0
        proceed = messagebox.askyesno(
            "Confirm two-parameter scan",
            f"Parameter 1 levels: {len(p1_values)}\n"
            f"Parameter 2 levels: {len(p2_values)}\n"
            f"Total simulations: {len(points)}\n"
            f"Rough duration at 3 minutes per point: {estimated_hours:.1f} hours\n\n"
            "Start the scan?",
        )
        if not proceed:
            return

        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:-3]
        run_dir = app_root() / "runs" / f"{timestamp}_two_parameter_scan"
        config = {
            "timestamp": timestamp,
            "run_dir": str(run_dir),
            "model": str(model),
            "profile": self.vars["profile"].get(),
            "parameter_1": p1,
            "parameter_2": p2,
            "linked_value": linked,
            "points": points,
            "conductor_count": conductor_count,
            "clearance": clearance,
            "torque_points": torque_points,
            "airgap_mesh": airgap_mesh,
            "git_commit": commit,
        }
        for item in self.tree.get_children():
            self.tree.delete(item)
        self.log_text.delete("1.0", "end")
        self.progress.configure(maximum=len(points), value=0)
        self.vars["commit"].set(commit)
        self.vars["status"].set(f"Running 0/{len(points)}")
        self.cancel_event.clear()
        self.start_button.configure(state="disabled")
        self.cancel_button.configure(state="normal")
        self.worker = ScanWorker(config, self.events, self.cancel_event)
        self.worker.start()

    def cancel_scan(self) -> None:
        self.cancel_event.set()
        self.cancel_button.configure(state="disabled")
        self.vars["status"].set("Stop requested; waiting for current point")

    def poll_events(self) -> None:
        try:
            while True:
                kind, payload = self.events.get_nowait()
                if kind == "log":
                    self.append_log(payload)
                elif kind == "row":
                    columns = ["run_index", "parameter_1_value", "parameter_2_value", "linked_value"] + [
                        item[0] for item in METRICS
                    ] + ["seconds", "status"]
                    self.tree.insert("", "end", values=[payload.get(column, "") for column in columns])
                    self.tree.yview_moveto(1.0)
                elif kind == "progress":
                    current, total = payload
                    self.progress.configure(value=current)
                    self.vars["status"].set(f"Running {current}/{total}")
                elif kind == "done":
                    self.finish_run(f"Complete: {payload['run_dir']}")
                    messagebox.showinfo("Scan complete", f"Results saved in:\n{payload['run_dir']}")
                elif kind == "fatal":
                    self.finish_run("Failed")
                    self.append_log(payload)
                    messagebox.showerror("Scan failed", payload)
        except queue.Empty:
            pass
        self.after(150, self.poll_events)

    def finish_run(self, status: str) -> None:
        self.vars["status"].set(status)
        self.start_button.configure(state="normal")
        self.cancel_button.configure(state="disabled")


def main() -> int:
    app = MotorCADScanApp()
    app.mainloop()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())