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