"""Task executor for local simulation system (P4-M2). Listens for tasks dispatched from Web backend, executes Motor-CAD simulations via RobustMotorCADSolver, reports progress and results. NOTE: All strings must be ASCII only. Chinese text uses \\uXXXX escapes. """ import gc import json import os import sys import time import uuid import threading from datetime import datetime from pathlib import Path from typing import Dict, List, Optional, Any, Callable try: import requests except ImportError: requests = None # Add scripts directory to path for robust_motorcad import _SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__)) if _SCRIPTS_DIR not in sys.path: sys.path.insert(0, _SCRIPTS_DIR) class TaskExecutor: """Executes simulation tasks dispatched from Web backend.""" def __init__( self, web_base_url: str = "http://127.0.0.1:8000", task_dir: Optional[str] = None, on_progress: Optional[Callable] = None, on_complete: Optional[Callable] = None, on_error: Optional[Callable] = None, enable_mock: bool = False, executor_id: Optional[str] = None, ): self.web_base_url = web_base_url.rstrip("/") self.task_dir = task_dir or os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "output", "tasks" ) os.makedirs(self.task_dir, exist_ok=True) self.on_progress = on_progress self.on_complete = on_complete self.on_error = on_error self.enable_mock = enable_mock self._running = False self._current_task: Optional[Dict[str, Any]] = None self._stop_event = threading.Event() # Set when the web side cancelled the currently-running task # (detected by polling GET /api/tasks/{id} between points). self._task_cancelled = False # Heartbeat observability: monotonic timestamp of the last heartbeat # that actually reached the backend, plus a handle on the heartbeat # thread itself. A silently-dead heartbeat thread once left the # executor running-but-OFFLINE (observed 2026-09-04). self.last_heartbeat_at: Optional[float] = None self._hb_thread: Optional[threading.Thread] = None if executor_id is not None: self.executor_id = executor_id else: self.executor_id = "motorcad-executor-%s-%s" % (os.getpid(), uuid.uuid4().hex[:4]) def fetch_pending_tasks(self) -> List[Dict[str, Any]]: """Fetch claimable tasks from Web backend. Web's start-simulation marks tasks as 'dispatched' immediately, while tasks created via the tasks API stay 'pending'. The executor claims BOTH states so every task gets picked up regardless of creation path. """ if requests is None: return self._scan_local_task_files() tasks = [] for st in ("pending", "dispatched"): try: resp = requests.get( f"{self.web_base_url}/api/tasks", params={"status": st, "limit": 20}, timeout=10, ) if resp.status_code == 200: data = resp.json() tasks.extend(data.get("tasks", [])) except Exception as e: if self.on_error: self.on_error(f"Fetch tasks failed: {str(e)}") # De-duplicate by task_id (keep first occurrence) seen = set() result = [] for t in tasks: tid = t.get("task_id") if tid and tid not in seen: seen.add(tid) result.append(t) return result def _hydrate_task(self, task: Dict[str, Any]) -> Dict[str, Any]: """Fetch full task payload (parameters + plan_data) from Web backend. The list API only returns task metadata; the actual parameter sets live in the task.json file exposed by the download endpoint. """ if requests is None: return task tid = task.get("task_id") if not tid: return task try: resp = requests.get( f"{self.web_base_url}/api/tasks/{tid}/download", timeout=10, ) if resp.status_code == 200: full = resp.json() if isinstance(full, dict): if full.get("parameters"): task = {**task, **full} elif task.get("_local_file"): # local file fallback try: with open(task["_local_file"], "r", encoding="utf-8") as f: local = json.load(f) if local.get("parameters"): task = {**task, **local} except Exception: pass except Exception as e: if self.on_error: self.on_error(f"Hydrate task {tid} failed: {str(e)}") return task def _scan_local_task_files(self) -> List[Dict[str, Any]]: """Scan local task directory for task files (fallback mode). Only picks up *_task.json files. Completed tasks are renamed to *_task.done.json to prevent infinite re-execution (B7 fix). """ tasks = [] for fname in os.listdir(self.task_dir): if fname.endswith("_task.json") and not fname.endswith("_task.done.json"): fpath = os.path.join(self.task_dir, fname) try: with open(fpath, "r", encoding="utf-8") as f: task = json.load(f) task["_local_file"] = fpath tasks.append(task) except Exception: continue return tasks def _mark_local_task_done(self, task: Dict[str, Any]) -> None: """Rename completed local task file to prevent re-execution (B7 fix).""" fpath = task.get("_local_file") if fpath and os.path.exists(fpath): done_path = fpath.replace("_task.json", "_task.done.json") try: os.rename(fpath, done_path) except Exception as e: if self.on_error: self.on_error(f"Failed to mark task done: {str(e)}") def dispatch_task(self, task_id: str) -> bool: """Mark task as dispatched on Web backend.""" if requests is None: return True try: resp = requests.post( f"{self.web_base_url}/api/tasks/{task_id}/dispatch", timeout=10, ) return resp.status_code in (200, 201) except Exception as e: if self.on_error: self.on_error(f"Dispatch task {task_id} failed: {str(e)}") return False def report_progress( self, task_id: str, current_point: int, total_points: int, current_params: Optional[Dict[str, Any]] = None, elapsed_time: Optional[float] = None, ) -> bool: """Report simulation progress to Web backend.""" if requests is None: if self.on_progress: self.on_progress(task_id, current_point, total_points) return True try: payload = { "current_point": current_point, "total_points": total_points, "current_params": current_params, "elapsed_time": elapsed_time, } resp = requests.post( f"{self.web_base_url}/api/tasks/{task_id}/progress", json=payload, timeout=10, ) return resp.status_code == 200 except Exception as e: if self.on_error: self.on_error(f"Report progress failed: {str(e)}") return False def _is_task_cancelled(self, task_id: str) -> bool: """Best-effort check whether the web side cancelled this task. Polls GET /api/tasks/{task_id} once per point. Network or backend failures return False on purpose: a transient error must never abort a running simulation. Local mode (no requests) cannot be cancelled remotely, so it always returns False. """ if requests is None or not self.web_base_url: return False try: resp = requests.get( f"{self.web_base_url}/api/tasks/{task_id}", timeout=5 ) if resp.status_code == 200: return resp.json().get("status") == "cancelled" except Exception: pass return False def report_results( self, task_id: str, results: List[Dict[str, Any]], metrics: Optional[Dict[str, Any]] = None, logs: Optional[str] = None, duration: Optional[float] = None, status: str = "completed", ) -> bool: """Report final results to Web backend.""" if requests is None: if self.on_complete: self.on_complete(task_id, results, metrics) return True try: payload = { "results": results, "metrics": metrics, "logs": logs, "duration": duration, "status": status, } resp = requests.post( f"{self.web_base_url}/api/tasks/{task_id}/results", json=payload, timeout=30, ) return resp.status_code == 200 except Exception as e: if self.on_error: self.on_error(f"Report results failed: {str(e)}") return False def _report_to_adaptive_loop( self, task: Dict[str, Any], results: List[Dict[str, Any]] ) -> None: """Feed an adaptive_batch task's results back into its loop. Maps each point result to {point_id, metrics, status} and posts to the loop's report-results endpoint. Best-effort: a failure here must not break normal task reporting (results are already stored on the task). """ if requests is None: return if task.get("task_type") != "adaptive_batch": return loop_id = task.get("loop_id") if not loop_id: return point_results = [] for r in results: pid = r.get("point_id") if pid is None: continue point_results.append({ "point_id": pid, "metrics": r.get("metrics") or {}, "status": "ok" if r.get("status") == "OK" else "failed", }) if not point_results: return try: resp = requests.post( f"{self.web_base_url}/api/adaptive/loops/{loop_id}/report-results", json={"point_results": point_results}, timeout=120, ) if self.on_progress: self.on_progress( f"Adaptive loop {loop_id}: reported {len(point_results)} " f"points (HTTP {resp.status_code})" ) except Exception as e: if self.on_error: self.on_error(f"Adaptive loop report failed ({loop_id}): {str(e)}") def execute_task(self, task: Dict[str, Any]) -> None: """Execute a single simulation task. This is a template method. Override _run_simulation_point in subclasses to implement actual Motor-CAD simulation. """ task = self._hydrate_task(task) task_id = task.get("task_id", str(uuid.uuid4())[:8]) parameters = task.get("parameters", []) total_points = len(parameters) results = [] start_time = time.time() self._current_task = task self._task_cancelled = False # P6: task-level thermal mode switch (off/steady/coupled). Read once # per task; every point uses it via _run_simulation_point. self._current_thermal_mode = task.get("thermal_mode") # start-simulation already marks a task 'dispatched' at creation, while # tasks created via the tasks API stay 'pending'. Only claim (dispatch) # a task that is still pending; re-dispatching an already-dispatched # task is rejected by the backend (pending -> dispatched only) and # would otherwise be misreported as "not claimable". claimed = True if task.get("status") == "pending": claimed = self.dispatch_task(task_id) if not claimed: # Another instance already claimed this task; skip it so # parallel executors never duplicate the same simulation. if self.on_error: self.on_error("Task %s not claimable (claimed/network); skip" % task_id) self._current_task = None return for idx, params in enumerate(parameters): if self._stop_event.is_set(): break # Web-side cancellation: the user clicked stop on the UI, which # flips the task status to 'cancelled'. Abort between points so a # cancelled task does not keep consuming Motor-CAD time; points # already finished are still reported below. if self._is_task_cancelled(task_id): self._task_cancelled = True if self.on_progress: self.on_progress(task_id, idx, total_points) break elapsed = time.time() - start_time self.report_progress(task_id, idx, total_points, params, elapsed) try: point_result = self._run_simulation_point(params, idx) point_result["point_index"] = idx if "point_id" in params: point_result["point_id"] = params["point_id"] point_result["params"] = params results.append(point_result) except Exception as e: # A2 fix: failed points are recorded as failed, NOT mock data failed_result = { "point_index": idx, "params": params, "status": "FAILED", "error": str(e), } if "point_id" in params: failed_result["point_id"] = params["point_id"] results.append(failed_result) if self.on_error: self.on_error(f"Point {idx} failed: {str(e)}") duration = time.time() - start_time metrics = self._compute_metrics(results) # Status reflects actual outcome: completed/cancelled/failed if self._stop_event.is_set() or self._task_cancelled: status = "cancelled" elif any(r.get("status") == "FAILED" for r in results): status = "completed_with_errors" if any( r.get("status") == "OK" for r in results ) else "failed" else: status = "completed" self.report_results(task_id, results, metrics, None, duration, status) self.report_progress(task_id, total_points, total_points, None, duration) # Adaptive-loop bridge: an adaptive_batch task belongs to a loop; feed # per-point results back to /adaptive/loops/{loop_id}/report-results so # the search advances without manual intervention (P3-M5 gap closure). self._report_to_adaptive_loop(task, results) # B7 fix: mark local task file as done to prevent re-execution if requests is None: self._mark_local_task_done(task) self._current_task = None if self.on_complete: self.on_complete(task_id, results, metrics) def _run_simulation_point(self, params: Dict[str, Any], index: int) -> Dict[str, Any]: """Run a single simulation point. Override in subclass. Mock data is ONLY returned when enable_mock=True (explicit opt-in). Mock results are tagged with source="mock" so they can never be confused with real simulation data (A2 fix). """ if not self.enable_mock: raise RuntimeError( "No simulation backend configured. " "Use MotorCADTaskExecutor for real Motor-CAD simulation, " "or set enable_mock=True for testing." ) import random rng = random.Random(index + hash(json.dumps(params, sort_keys=True)) % 10000) airgap = params.get("airgap_mm", 1.0) current = params.get("current_a", 15.0) return { "tavg_nm": round(current * 2.5 / (airgap ** 0.5) + rng.gauss(0, 0.3), 4), "efficiency_pct": round(88 + rng.gauss(0, 2), 2), "total_losses_w": round(50 + rng.gauss(0, 10), 2), "winding_temp_c": round(90 + rng.gauss(0, 10), 1), "status": "OK", "source": "mock", } def _compute_metrics(self, results: List[Dict[str, Any]]) -> Dict[str, Any]: """Compute aggregated metrics from results.""" ok_results = [r for r in results if r.get("status") == "OK"] if not ok_results: return { "total_points": len(results), "successful_points": 0, "failed_points": len(results), } metrics = { "total_points": len(results), "successful_points": len(ok_results), "failed_points": len(results) - len(ok_results), } for key in ["tavg_nm", "efficiency_pct", "total_losses_w", "winding_temp_c"]: values = [r[key] for r in ok_results if key in r] if values: metrics[f"{key}_min"] = min(values) metrics[f"{key}_max"] = max(values) metrics[f"{key}_mean"] = round(sum(values) / len(values), 4) return metrics def _send_heartbeat(self) -> None: """Register this executor with the Web backend (online status).""" if requests is None: return try: status = "running" if self._current_task is not None else "idle" current_task = None progress = None if self._current_task is not None: current_task = self._current_task.get("task_id") total = self._current_task.get("total_points") or 0 done = self._current_task.get("completed_points") or 0 progress = { "completed_points": done, "total_points": total, } requests.post( f"{self.web_base_url}/api/executor/heartbeat", json={ "executor_id": self.executor_id, "status": status, "current_task": current_task, "progress": progress, }, timeout=5, ) self.last_heartbeat_at = time.monotonic() except Exception: # Heartbeat failures are non-fatal pass def start_polling(self, interval: int = 5) -> List[threading.Thread]: """Start background threads: one polls/executes tasks, one heartbeats. Heartbeat runs on its own thread so a long-running Motor-CAD point (~2 min each) never starves the heartbeat - otherwise the backend would mark this executor offline mid-task (observed 2026-09-04). Returns BOTH threads so the caller can watch them; a silently dead thread must surface as a loud failure, not a zombie executor. """ self._running = True self._stop_event.clear() def heartbeat_loop(): while self._running and not self._stop_event.is_set(): try: self._send_heartbeat() except Exception as e: # _send_heartbeat already swallows request errors; this # guards the unexpected so the thread never dies silently. if self.on_error: self.on_error(f"Heartbeat loop error: {str(e)}") self._stop_event.wait(interval) def poll_loop(): while self._running and not self._stop_event.is_set(): try: tasks = self.fetch_pending_tasks() for task in tasks: if self._stop_event.is_set(): break self.execute_task(task) # Post-task GC breadcrumb: COM objects released at GC # time can hang while holding the GIL, freezing every # thread (zombie executor observed 2026-09-04). Collect # at a known point so a hang here is pinpointed by the # surrounding log lines instead of being invisible. if self.on_progress: self.on_progress("post-task-gc:begin") gc.collect() if self.on_progress: self.on_progress("post-task-gc:end") except Exception as e: if self.on_error: self.on_error(f"Poll loop error: {str(e)}") self._stop_event.wait(interval) hb_thread = threading.Thread(target=heartbeat_loop, daemon=True) hb_thread.start() self._hb_thread = hb_thread thread = threading.Thread(target=poll_loop, daemon=True) thread.start() return [thread, hb_thread] def stop(self): """Stop the executor.""" self._running = False self._stop_event.set() class MotorCADTaskExecutor(TaskExecutor): """Task executor backed by a registered simulation-tool adapter. Uses afmcore.adapters.get_adapter(tool) so the executor never hard-codes a specific solver. Default tool "motorcad" wraps RobustMotorCADSolver (open_new_instance, set_visible, baseline reload per point, popup suppression, write-back verification, per-point disk flush). Result mapping: adapter returns {metrics, status, error, ...}; the metrics dict is flattened to the point's top level so downstream aggregation (TaskExecutor._compute_metrics) keeps working unchanged. """ def __init__(self, *args, model_path: Optional[str] = None, tool: str = "motorcad", enable_thermal: bool = False, ambient_temperature: Optional[float] = None, **kwargs): # Mock fallback is disabled by default for real solver adapter. kwargs.setdefault("enable_mock", False) super().__init__(*args, **kwargs) self.model_path = model_path self.tool = tool # P5-M6: pass through to the adapter so each EM point can also run a # steady-state thermal solve and merge thermal metrics. self.enable_thermal = bool(enable_thermal) # P5-M6 thermal boundary: Ambient_Temperature override (degC). self.ambient_temperature = ambient_temperature self._adapter = None def _ensure_adapter(self): """Lazily create the tool adapter via the platform registry.""" if self._adapter is not None: return self._adapter _root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) _src = os.path.join(_root, "src") if _src not in sys.path: sys.path.insert(0, _src) from afmcore.adapters import get_adapter # Dynamically import the adapter module matching self.tool so it # self-registers in ADAPTER_REGISTRY. Unknown tools rely on # pre-registered adapters (caller may have imported them). if self.tool == "motorcad": import afmcore.adapters.motorcad # noqa: F401 elif self.tool == "maxwell": import afmcore.adapters.maxwell # noqa: F401 elif self.tool == "jmag": import afmcore.adapters.jmag # noqa: F401 if not self.model_path: raise RuntimeError("model_path is required for MotorCADTaskExecutor") output_dir = os.path.join( _root, "output", "task_%s" % datetime.now().strftime("%Y%m%d_%H%M%S") ) self._adapter = get_adapter( self.tool, model_path=self.model_path, output_dir=output_dir, enable_thermal=self.enable_thermal, ambient_temperature=self.ambient_temperature, ) self._adapter.connect() return self._adapter def _run_simulation_point(self, params: Dict[str, Any], index: int) -> Dict[str, Any]: """Run one point through the adapter and flatten metrics to top level. The adapter owns the robust protocol (baseline reload, write-back verification, export parsing). A non-OK point raises so execute_task records status=FAILED (no mock fallback). When enable_mock=True the base-class mock implementation is used instead, so no Motor-CAD instance is launched at all (P5-M2). """ if self.enable_mock: return super()._run_simulation_point(params, index) adapter = self._ensure_adapter() result = adapter.run_point( self.model_path, params=params, output_dir=os.path.dirname(os.path.dirname(os.path.abspath(__file__))), tag=str(index), thermal_mode=getattr(self, "_current_thermal_mode", None), ) if result.get("status") != "OK": raise RuntimeError( result.get("error") or ("Simulation failed (adapter status=%s)" % result.get("status")) ) metrics = result.get("metrics") or {} point = dict(metrics) point["status"] = "OK" point["metrics"] = metrics point["error"] = result.get("error") point["solve_time_s"] = result.get("solve_time_s") # Lossless full export (all Motor-CAD fields with units) rides along # for the complete-data Excel/CSV export; stored in raw_json column. point["raw_flat"] = result.get("raw_flat", []) return point def cleanup(self): """Disconnect the adapter and release the tool instance.""" if self._adapter is not None: try: self._adapter.disconnect() except Exception: pass self._adapter = None if __name__ == "__main__": # Standalone test: run executor with mock data (explicit) executor = TaskExecutor( web_base_url=os.environ.get("WEB_BASE_URL", "http://127.0.0.1:8000"), on_progress=lambda tid, cur, tot: print(f"[{tid}] Progress: {cur}/{tot}"), on_complete=lambda tid, res, met: print(f"[{tid}] Complete: {len(res)} points"), on_error=lambda msg: print(f"ERROR: {msg}"), enable_mock=True, ) print("Task executor started (mock mode). Press Ctrl+C to stop.") try: threads = executor.start_polling(interval=5) while any(t.is_alive() for t in threads): time.sleep(1) except KeyboardInterrupt: executor.stop() print("Executor stopped.")