"""Simulation plan API router (CRUD + download + upload results + start simulation).""" import json import math import uuid from datetime import datetime from typing import Optional, Literal from fastapi import APIRouter, Depends, HTTPException, UploadFile, File from pydantic import BaseModel from sqlalchemy.orm import Session from ..database import get_db from ..metrics_constants import METRIC_KEYS from ..models.project import Project from ..models.simulation_plan import SimulationPlan from ..models.simulation_result import SimulationResult from ..schemas.simulation_plan import ( PlanCreate, PlanUpdate, PlanResponse, PlanListResponse, PlanDownloadResponse, ) from ..schemas.simulation_result import ResultListResponse from ..services.task_manager import get_task_manager router = APIRouter(prefix="/api/plans", tags=["plans"]) @router.get("/variable-catalog") def get_variable_catalog(topology: str = "SSSR"): """Return topology-aware variable catalog for frontend scan-variable selectors. Returns the fixed-parameter template (with motorcad_var resolved) and the set of known Motor-CAD variable names for the given topology. Frontend should use this to populate scan-variable dropdowns and prevent users from entering invalid variable names. Args: topology: Motor topology (SSSR/AFIR/RFM). Defaults to SSSR. """ from ..services.fixed_params_template import FIXED_PARAM_TEMPLATES from ..services.topology_variable_map import ( get_known_variables, normalize_topology, resolve_variable, ) topo = normalize_topology(topology) # Build template with resolved motorcad_var for this topology. # For params where motorcad_var is None but the name is a known alias, # resolve it. Otherwise keep name as the variable name if known. template = [] for p in FIXED_PARAM_TEMPLATES: row = dict(p) mc_var = row.get("motorcad_var") if not mc_var: # Try to resolve from alias map resolved, was_alias = resolve_variable(row["name"], topo) if was_alias: row["motorcad_var"] = resolved else: row["motorcad_var"] = row["name"] template.append(row) known_vars = sorted(get_known_variables(topo)) return { "topology": topo, "template": template, "known_variables": known_vars, "total_known": len(known_vars), "total_template": len(template), } def _generate_plan_id() -> str: """Generate a unique plan ID with timestamp + random suffix to avoid collisions.""" return f"SP-{datetime.now().strftime('%Y%m%d-%H%M%S')}-{uuid.uuid4().hex[:6]}" def _plan_to_response(db: Session, plan: SimulationPlan) -> PlanResponse: result_count = db.query(SimulationResult).filter(SimulationResult.plan_id == plan.id).count() variables_summary = {} try: variables_summary = json.loads(plan.variables_summary) if plan.variables_summary else {} except (json.JSONDecodeError, TypeError): pass return PlanResponse( id=plan.id, project_id=plan.project_id, name=plan.name, plan_id=plan.plan_id, status=plan.status, plan_data=plan.get_plan_dict(), variables_summary=variables_summary, estimated_points=plan.estimated_points or 0, estimated_time_min=plan.estimated_time_min or 0, notes=plan.notes or "", result_count=result_count, created_at=plan.created_at, updated_at=plan.updated_at, ) @router.get("", response_model=PlanListResponse) def list_plans( project_id: int | None = None, skip: int = 0, limit: int = 50, db: Session = Depends(get_db), ): """List simulation plans, optionally filtered by project.""" query = db.query(SimulationPlan) if project_id: query = query.filter(SimulationPlan.project_id == project_id) total = query.count() plans = query.order_by(SimulationPlan.updated_at.desc()).offset(skip).limit(limit).all() return PlanListResponse(total=total, items=[_plan_to_response(db, p) for p in plans]) @router.post("", response_model=PlanResponse, status_code=201) def create_plan(data: PlanCreate, db: Session = Depends(get_db)): """Create a new simulation plan.""" project = db.query(Project).filter(Project.id == data.project_id).first() if not project: raise HTTPException(status_code=404, detail="Project not found") plan_id = _generate_plan_id() plan_data = data.plan_data plan_data["plan_id"] = plan_id # Persist the project boundary conditions (normalized to canonical BC # keys) when the caller did not supply them, so the plan-detail boundary # display has data and the plan is traceable to its BC (P1-1). if not plan_data.get("boundary_conditions"): from ..services.bc_fields import normalize_bc plan_data["boundary_conditions"] = normalize_bc(project.get_boundary_conditions()) # Source tracking: manually created plans inherit all BC from the project # (user-specified). AI-generated plans tag sources in ai_plan.py instead. if not plan_data.get("bc_meta"): plan_data["bc_meta"] = { k: {"source": "user"} for k in (plan_data.get("boundary_conditions") or {}) } # Auto-fill the base model from the topology registry when neither the # caller nor the project provided a model path. if not (plan_data.get("model_path") or "").strip(): from src.afmcore.topology import default_model_for fallback = default_model_for(plan_data.get("topology") or project.topology or "") if fallback: plan_data["model_path"] = fallback # Validate against the single-source plan schema (draft-stage: model # path may be configured later). from src.plan_schema import validate_plan_dict _ok, _errs = validate_plan_dict(plan_data, require_model_path=False) if not _ok: raise HTTPException( status_code=400, detail="Invalid plan_data: " + "; ".join(_errs), ) # Extract variables summary for display variables_summary = {} estimated_points = 1 for var in plan_data.get("variables", []): name = var.get("name", "unknown") values = var.get("values", []) variables_summary[name] = { "unit": var.get("unit", ""), "values": values, "count": len(values), } estimated_points *= len(values) if values else 1 plan = SimulationPlan( project_id=data.project_id, name=data.name, plan_id=plan_id, status="draft", estimated_points=estimated_points, estimated_time_min=estimated_points * 3, # ~3 min per point estimate notes=data.notes, ) plan.set_plan_dict(plan_data) plan.variables_summary = json.dumps(variables_summary, ensure_ascii=False) db.add(plan) db.commit() db.refresh(plan) return _plan_to_response(db, plan) @router.get("/{plan_id}", response_model=PlanResponse) def get_plan(plan_id: int, db: Session = Depends(get_db)): """Get a plan by ID.""" plan = db.query(SimulationPlan).filter(SimulationPlan.id == plan_id).first() if not plan: raise HTTPException(status_code=404, detail="Plan not found") return _plan_to_response(db, plan) @router.put("/{plan_id}", response_model=PlanResponse) def update_plan(plan_id: int, data: PlanUpdate, db: Session = Depends(get_db)): """Update a plan.""" plan = db.query(SimulationPlan).filter(SimulationPlan.id == plan_id).first() if not plan: raise HTTPException(status_code=404, detail="Plan not found") update_data = data.model_dump(exclude_unset=True) if "plan_data" in update_data: from src.plan_schema import validate_plan_dict _ok, _errs = validate_plan_dict( update_data["plan_data"], require_model_path=False ) if not _ok: raise HTTPException( status_code=400, detail="Invalid plan_data: " + "; ".join(_errs), ) plan.set_plan_dict(update_data.pop("plan_data")) for key, value in update_data.items(): setattr(plan, key, value) db.commit() db.refresh(plan) return _plan_to_response(db, plan) @router.delete("/{plan_id}", status_code=204) def delete_plan(plan_id: int, db: Session = Depends(get_db)): """Delete a plan and its results.""" plan = db.query(SimulationPlan).filter(SimulationPlan.id == plan_id).first() if not plan: raise HTTPException(status_code=404, detail="Plan not found") db.delete(plan) db.commit() return None # --------------------------------------------------------------------------- # API integration endpoints (for local executor) # --------------------------------------------------------------------------- @router.get("/{plan_id}/download", response_model=PlanDownloadResponse) def download_plan(plan_id: int, db: Session = Depends(get_db)): """Download a plan as simulation_plan.json (read-only, F5 fix). This is the API endpoint used by the local execution system to fetch plans. Returns the raw plan JSON in the exact format expected by src/plan_schema.py. Status changes must go through POST /{plan_id}/start-execution. """ plan = db.query(SimulationPlan).filter(SimulationPlan.id == plan_id).first() if not plan: raise HTTPException(status_code=404, detail="Plan not found") return PlanDownloadResponse(plan_id=plan.plan_id, plan_data=plan.get_plan_dict()) @router.post("/{plan_id}/start-execution") def start_execution(plan_id: int, db: Session = Depends(get_db)): """Mark a plan as executing (explicit state transition, F5 fix).""" plan = db.query(SimulationPlan).filter(SimulationPlan.id == plan_id).first() if not plan: raise HTTPException(status_code=404, detail="Plan not found") if plan.status in ("draft", "confirmed"): plan.status = "executing" db.commit() return {"plan_id": plan.plan_id, "status": plan.status} @router.get("/by-plan-id/{plan_uuid}/download", response_model=PlanDownloadResponse) def download_plan_by_uuid(plan_uuid: str, db: Session = Depends(get_db)): """Download a plan by its plan_id string (read-only, F5 fix).""" plan = db.query(SimulationPlan).filter(SimulationPlan.plan_id == plan_uuid).first() if not plan: raise HTTPException(status_code=404, detail="Plan not found") return PlanDownloadResponse(plan_id=plan.plan_id, plan_data=plan.get_plan_dict()) @router.post("/{plan_id}/upload-results", status_code=201) async def upload_results( plan_id: int, file: UploadFile = File(...), db: Session = Depends(get_db), ): """Upload scan_results.csv from local executor. Parses the CSV and stores each row as a SimulationResult. This is the API endpoint used by the local execution system to push results back. """ import csv import io plan = db.query(SimulationPlan).filter(SimulationPlan.id == plan_id).first() if not plan: raise HTTPException(status_code=404, detail="Plan not found") # Clear existing results for this plan (re-upload replaces) db.query(SimulationResult).filter(SimulationResult.plan_id == plan_id).delete() content = await file.read() text = content.decode("utf-8-sig") reader = csv.DictReader(io.StringIO(text)) # A1 fix: use shared metric constants (single source of truth) metric_keys = METRIC_KEYS standard_keys = {"run_index", "status", "seconds", "error"} count = 0 for row in reader: params = {} metrics = {} for key, val in row.items(): if key in standard_keys or val == "" or val is None: continue try: fval = float(val) except (ValueError, TypeError): continue if key in metric_keys: metrics[key] = fval else: params[key] = fval result = SimulationResult( plan_id=plan_id, run_index=int(row.get("run_index", count + 1)), status=row.get("status", "OK"), solve_time_s=float(row.get("seconds", 0) or 0), error_message=row.get("error", ""), ) result.set_params(params) result.set_metrics(metrics) db.add(result) count += 1 # Update plan status plan.status = "completed" db.commit() return {"message": f"Uploaded {count} results", "count": count, "plan_id": plan.plan_id} @router.get("/{plan_id}/results", response_model=ResultListResponse) def get_plan_results(plan_id: int, db: Session = Depends(get_db)): """Get all results for a plan.""" plan = db.query(SimulationPlan).filter(SimulationPlan.id == plan_id).first() if not plan: raise HTTPException(status_code=404, detail="Plan not found") results = ( db.query(SimulationResult) .filter(SimulationResult.plan_id == plan_id) .order_by(SimulationResult.run_index.asc()) .all() ) from ..schemas.simulation_result import ResultResponse items = [ ResultResponse( id=r.id, plan_id=r.plan_id, run_index=r.run_index, status=r.status, solve_time_s=r.solve_time_s, params=r.get_params(), metrics=r.get_metrics(), error_message=r.error_message or "", created_at=r.created_at, ) for r in results ] return ResultListResponse(total=len(items), items=items) @router.get("/{plan_id}/export-xlsx") def export_plan_results_xlsx(plan_id: int, db: Session = Depends(get_db)): """Export plan simulation results as a formatted .xlsx workbook. Format follows the torqrippswap scan_results.xlsx convention: two-row header (merged category groups + column headers), freeze panes, autofilter. Columns cover scan info, scan variables, electromagnetic metrics, and thermal metrics (only columns present in the data are exported, so thermal columns appear only for thermal_mode != off). """ import tempfile from urllib.parse import quote from fastapi.responses import FileResponse from starlette.background import BackgroundTask from ..metrics_constants import METRIC_KEYS as _METRIC_KEYS # noqa: F401 from afmcore.xlsx_report import write_summary_xlsx plan = db.query(SimulationPlan).filter(SimulationPlan.id == plan_id).first() if not plan: raise HTTPException(status_code=404, detail="Plan not found") results = ( db.query(SimulationResult) .filter(SimulationResult.plan_id == plan_id) .order_by(SimulationResult.run_index.asc()) .all() ) if not results: raise HTTPException(status_code=400, detail="No simulation results to export") col_keys, col_categories, col_headers, data_rows = _collect_export_rows( plan, results ) ts = datetime.now().strftime("%Y%m%d_%H%M%S") tmp = tempfile.NamedTemporaryFile( prefix=f"scan_results_{ts}_", suffix=".xlsx", delete=False ) tmp.close() write_summary_xlsx(tmp.name, col_keys, col_categories, col_headers, data_rows) plan_name = (plan.name or f"plan_{plan_id}").strip() or f"plan_{plan_id}" # RFC 5987: ASCII fallback + UTF-8 filename* for Chinese plan names. ascii_fallback = f"scan_results_{plan_id}_{ts}.xlsx" utf8_name = f"scan_results_{plan_name}_{ts}.xlsx" disposition = ( f"attachment; filename=\"{ascii_fallback}\"; " f"filename*=UTF-8''{quote(utf8_name)}" ) return FileResponse( tmp.name, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": disposition}, background=BackgroundTask(os_remove_quiet, tmp.name), ) def os_remove_quiet(path: str) -> None: """Best-effort temp-file cleanup after FileResponse streaming.""" import os try: os.remove(path) except OSError: pass def _collect_export_rows(plan, results): """Shared row/column assembly for xlsx and csv exports. Returns (col_keys, col_categories, col_headers, data_rows). Columns: scan info (fixed) -> scan params -> normalized EM/thermal key metrics -> the lossless raw export (every Motor-CAD field, grouped by its export section, header = Chinese name + [unit]). """ from afmcore.xlsx_report import build_export_columns param_labels: dict = {} try: plan_dict = plan.get_plan_dict() for var in plan_dict.get("variables", []) or []: name = var.get("name") if not name: continue cn = var.get("name_cn") or name unit = var.get("unit") or "" param_labels[name] = f"{cn}[{unit}]" if unit else cn except Exception: param_labels = {} param_names: list = [] present_metrics: set = set() data_rows: list = [] raw_units: dict = {} # "section|name" -> unit (first-seen) for r in results: params = r.get_params() metrics = r.get_metrics() for name in params: if name not in param_names: param_names.append(name) present_metrics.update(metrics.keys()) row = { "run_index": r.run_index, "status": r.status, "solve_time_s": r.solve_time_s, "error": r.error_message or "", } row.update(params) row.update(metrics) # Lossless raw archive -> flat row values keyed by section|name. for entry in r.get_raw(): sec = entry.get("section", "") nm = entry.get("name", "") if not nm: continue row[f"__raw__{sec}|{nm}"] = entry.get("value", "") uk = f"{sec}|{nm}" if uk not in raw_units: raw_units[uk] = entry.get("unit") or "" data_rows.append(row) present_metrics = {k for k in present_metrics if isinstance(k, str)} col_keys, col_categories, col_headers = build_export_columns( param_names, sorted(present_metrics), param_labels ) # Raw columns: union over all rows, first-seen order, grouped by the # Motor-CAD export section (Chinese category names). Thermal sections # were prefixed "Thermal-" by the solver; display as "热仿真-". seen_raw: set = set() for row in data_rows: for key in row: if not key.startswith("__raw__") or key in seen_raw: continue seen_raw.add(key) sec, _, nm = key[len("__raw__"):].partition("|") entry_unit = raw_units.get(f"{sec}|{nm}", "") category = sec.replace("Thermal-", "热仿真-") category = category.replace("Node Temperatures", "节点温度") header = f"{nm}[{entry_unit}]" if entry_unit else nm # Disambiguate duplicate headers (same field name exported in # two sections, e.g. 系统效率 in 驱动 and 电磁). if header in col_headers: header = f"{category}-{header}" col_keys.append(key) col_categories.append(category) col_headers.append(header) return col_keys, col_categories, col_headers, data_rows @router.get("/{plan_id}/export-csv") def export_plan_results_csv(plan_id: int, db: Session = Depends(get_db)): """Export plan simulation results as CSV (same columns as the xlsx export). Single header row (Chinese labels with units); utf-8-sig BOM so Excel opens the file with correct encoding. """ import csv import io from urllib.parse import quote from fastapi.responses import Response plan = db.query(SimulationPlan).filter(SimulationPlan.id == plan_id).first() if not plan: raise HTTPException(status_code=404, detail="Plan not found") results = ( db.query(SimulationResult) .filter(SimulationResult.plan_id == plan_id) .order_by(SimulationResult.run_index.asc()) .all() ) if not results: raise HTTPException(status_code=400, detail="No simulation results to export") col_keys, _cats, col_headers, data_rows = _collect_export_rows(plan, results) buf = io.StringIO() writer = csv.writer(buf) writer.writerow(col_headers) for row in data_rows: writer.writerow([row.get(k, "") for k in col_keys]) ts = datetime.now().strftime("%Y%m%d_%H%M%S") plan_name = (plan.name or f"plan_{plan_id}").strip() or f"plan_{plan_id}" ascii_fallback = f"scan_results_{plan_id}_{ts}.csv" utf8_name = f"scan_results_{plan_name}_{ts}.csv" disposition = ( f"attachment; filename=\"{ascii_fallback}\"; " f"filename*=UTF-8''{quote(utf8_name)}" ) return Response( content=buf.getvalue().encode("utf-8-sig"), media_type="text/csv", headers={"Content-Disposition": disposition}, ) # --------------------------------------------------------------------------- # One-click simulation start (P0-4) # --------------------------------------------------------------------------- def _expand_plan_to_parameters(plan_data: dict) -> list[dict]: """Expand plan variables into full parameter sets (Cartesian product). Fixed params are canonicalized against the template: the template is the source of truth for the real Motor-CAD variable name (motorcad_var), while the plan's stored value overrides the template default. Only params with a real motorcad_var are written to the simulation. Scan variables are resolved through: 1. The template (if name matches a template param, use its motorcad_var) 2. The topology-aware alias map (RFM names remapped to AFM names) 3. Fallback: use the name as-is (caller should validate). Returns list of param dicts for the task executor. """ from ..services.fixed_params_template import FIXED_PARAM_TEMPLATES from ..services.topology_variable_map import ( resolve_variable, normalize_topology, ) topology = normalize_topology(plan_data.get("topology")) template_by_name = {p["name"].lower(): p for p in FIXED_PARAM_TEMPLATES} plan_fps = plan_data.get("fixed_params", []) or [] seen = set() merged = [] for fp in plan_fps: if not (isinstance(fp, dict) and fp.get("name")): continue name = fp["name"] key = name.lower() if key in seen: continue seen.add(key) tmpl = template_by_name.get(key) val = fp.get("value") if tmpl: row = dict(tmpl) # includes motorcad_var (baseline-tuned default) # Only user-explicitly-modified values override the baseline # default. AI-suggested values (or legacy params without source) # keep the template default so old plans run with valid geometry. if fp.get("source") == "user" and val is not None and val != "": row["value"] = val merged.append(row) else: # Param not in template: try topology alias resolution. # If the alias map resolves it, use the resolved name as # motorcad_var. Otherwise keep None (will not be written). row = dict(fp) resolved, was_alias = resolve_variable(name, topology) if was_alias: row["motorcad_var"] = resolved else: row["motorcad_var"] = None merged.append(row) # Template params missing from the plan (fill with template defaults) for tmpl in FIXED_PARAM_TEMPLATES: if tmpl["name"].lower() not in seen: merged.append(dict(tmpl)) seen.add(tmpl["name"].lower()) fixed = {} for fp in merged: var = fp.get("motorcad_var") val = fp.get("value") if var and val is not None and val != "": try: fixed[var] = float(val) except (TypeError, ValueError): fixed[var] = val variables = plan_data.get("variables", []) if not variables: return [dict(fixed)] # Collect value lists for each variable, resolving the variable name # through template -> topology alias map -> fallback to raw name. var_value_lists = [] for v in variables: if not isinstance(v, dict): continue name = v.get("name", "") if not name: continue # Resolve the Motor-CAD variable name for this scan variable. tmpl = template_by_name.get(name.lower()) if tmpl and tmpl.get("motorcad_var"): resolved_name = tmpl["motorcad_var"] else: resolved_name, _was_alias = resolve_variable(name, topology) values = v.get("values", []) if not values and v.get("start") is not None and v.get("stop") is not None and v.get("step"): start, stop, step = float(v["start"]), float(v["stop"]), float(v["step"]) count = int(math.floor((stop - start) / step + 1e-9)) + 1 values = [round(start + i * step, 6) for i in range(count)] if values and abs(values[-1] - stop) > 1e-9: values.append(round(stop, 6)) if values: var_value_lists.append((resolved_name, values)) if not var_value_lists: return [dict(fixed)] # Cartesian product def _cartesian(idx: int, current: dict) -> list[dict]: if idx >= len(var_value_lists): return [dict(current)] name, vals = var_value_lists[idx] result = [] for val in vals: current[name] = val result.extend(_cartesian(idx + 1, current)) return result return _cartesian(0, dict(fixed)) @router.get("/{plan_id}/preflight") def preflight_check(plan_id: int, db: Session = Depends(get_db)): """Pre-flight checklist before starting a simulation. Returns machine-readable checks (key/status/data only; the frontend maps keys to localized labels and messages). status: pass | warn | fail. Any 'fail' blocks starting; 'warn' is advisory and does not block. """ import os from ..config import PROJECT_ROOT plan = db.query(SimulationPlan).filter(SimulationPlan.id == plan_id).first() if not plan: raise HTTPException(status_code=404, detail="Plan not found") plan_data = plan.get_plan_dict() checks = [] # 1. Model file must exist. model_path may be repo-relative (e.g. # "models/xxx.mot") or absolute; resolve relative paths against the repo # root so the check matches how the executor locates the model. model_path = (plan_data.get("model_path") or "").strip() resolved = model_path if model_path and not os.path.isabs(model_path): resolved = os.path.join(str(PROJECT_ROOT), model_path) # Expose the topology's default base model so the UI can offer one-click # repair when the check fails. from src.afmcore.topology import default_model_for default_model = default_model_for(plan_data.get("topology") or "") if not model_path: checks.append({ "key": "model_path", "status": "fail", "value": "", "fixable": bool(default_model), "default_model": default_model, }) elif not os.path.exists(resolved): checks.append({ "key": "model_path", "status": "fail", "value": model_path, "fixable": bool(default_model), "default_model": default_model, }) else: checks.append({"key": "model_path", "status": "pass", "value": model_path}) # 2. Fixed params whose Motor-CAD variable name is unverified (no mapping). unverified = [ fp["name"] for fp in (plan_data.get("fixed_params") or []) if isinstance(fp, dict) and fp.get("name") and not fp.get("motorcad_var") ] checks.append({ "key": "unverified_vars", "status": "warn" if unverified else "pass", "items": unverified, }) # 3. At least one scan variable with values. variables = plan_data.get("variables") or [] valid_vars = [ v for v in variables if isinstance(v, dict) and v.get("name") and (v.get("values") or []) ] checks.append({ "key": "scan_vars", "status": "pass" if valid_vars else "fail", "count": len(valid_vars), }) # 4. Point-count estimate (advisory when large). total = 1 for v in valid_vars: total *= len(v.get("values") or [1]) checks.append({ "key": "point_count", "status": "warn" if total > 200 else "pass", "count": total, }) # 5. Local executor online (advisory: tasks can queue while offline). try: exec_status = get_task_manager().get_executor_status() online = sum(1 for e in exec_status.get("executors", []) if e.get("online")) except Exception: online = 0 checks.append({ "key": "executor", "status": "pass" if online > 0 else "warn", "online": online, }) ok = not any(c["status"] == "fail" for c in checks) return {"ok": ok, "checks": checks} class StartSimulationRequest(BaseModel): """Optional one-click start options (task-level thermal switch).""" thermal_mode: Literal["off", "steady", "coupled"] = "steady" @router.post("/{plan_id}/start-simulation") def start_simulation( plan_id: int, request: Optional[StartSimulationRequest] = None, db: Session = Depends(get_db), ): """One-click start: expand plan to parameters, create task, dispatch. Automatically: 1. Expands variables into Cartesian product parameter sets 2. Merges fixed_params into each parameter set 3. Creates a Task linked to this plan 4. Marks task as dispatched (local executor picks it up) 5. Updates plan status to 'executing' Returns the created task info. """ thermal_mode = request.thermal_mode if request is not None else "steady" plan = db.query(SimulationPlan).filter(SimulationPlan.id == plan_id).first() if not plan: raise HTTPException(status_code=404, detail="Plan not found") plan_data = plan.get_plan_dict() # Runtime fallback: legacy plans may carry an empty model_path (created # before the topology default-model auto-fill). Auto-fill from the # topology registry and persist so the plan becomes self-contained. if not (plan_data.get("model_path") or "").strip(): from src.afmcore.topology import default_model_for fallback = default_model_for(plan_data.get("topology") or "") if fallback: plan_data["model_path"] = fallback plan.set_plan_dict(plan_data) db.commit() parameters = _expand_plan_to_parameters(plan_data) if not parameters: raise HTTPException(status_code=400, detail="Plan has no valid parameters to simulate") # --- Topology-aware variable name validation (P1-bugfix: plan 23) --- # Reject unknown variable names BEFORE creating the task, with suggested # alternatives. This prevents silent Motor-CAD "Could not find variable" # failures that waste 15+ minutes of simulation time. from ..services.topology_variable_map import ( is_known_variable, suggest_alternative, normalize_topology, ) topo = normalize_topology(plan_data.get("topology")) if parameters: all_var_names = list(parameters[0].keys()) unknown = [] for vname in all_var_names: if not is_known_variable(vname, topo): suggestion = suggest_alternative(vname, topo) unknown.append({ "variable": vname, "suggestion": suggestion, }) if unknown: detail_lines = [ f"Unknown Motor-CAD variable(s) for topology {topo}. " "These will cause 'Could not find variable' errors in Motor-CAD.", ] for u in unknown: if u["suggestion"]: detail_lines.append( f" - '{u['variable']}' -> did you mean '{u['suggestion']}'?" ) else: detail_lines.append( f" - '{u['variable']}' (no close match found; verify against .mot model)" ) raise HTTPException(status_code=400, detail="\n".join(detail_lines)) # --- End variable name validation --- # Create task via task manager manager = get_task_manager() task = manager.create_task( plan_id=plan_id, plan_data=plan_data, parameters=parameters, task_name=f"{plan.name}_run", priority=5, created_by="web", thermal_mode=thermal_mode, ) # Dispatch immediately try: manager.dispatch_task(task["task_id"]) except ValueError: pass # Already dispatched or other state issue # Update plan status plan.status = "executing" db.commit() return { "task_id": task["task_id"], "task_name": task["task_name"], "plan_id": plan.plan_id, "total_points": len(parameters), "status": "dispatched", "message": f"Simulation started with {len(parameters)} points. Local executor will pick it up.", } @router.get("/{plan_id}/active-task") def get_active_task(plan_id: int, db: Session = Depends(get_db)): """Get the most recent active task for a plan (for progress display).""" plan = db.query(SimulationPlan).filter(SimulationPlan.id == plan_id).first() if not plan: raise HTTPException(status_code=404, detail="Plan not found") manager = get_task_manager() tasks = manager.list_tasks(plan_id=plan_id, limit=1) if tasks.get("tasks"): return tasks["tasks"][0] return None # --------------------------------------------------------------------------- # AI Analysis & Iteration (P1-8) # --------------------------------------------------------------------------- @router.post("/{plan_id}/ai-analyze") def ai_analyze_results(plan_id: int, db: Session = Depends(get_db)): """Analyze simulation results using AI and return insights. Returns key metrics summary, parameter sensitivity, anomaly detection, and recommendations for next iteration. """ plan = db.query(SimulationPlan).filter(SimulationPlan.id == plan_id).first() if not plan: raise HTTPException(status_code=404, detail="Plan not found") results = ( db.query(SimulationResult) .filter(SimulationResult.plan_id == plan_id) .order_by(SimulationResult.run_index.asc()) .all() ) if not results: raise HTTPException(status_code=400, detail="No simulation results to analyze") # Compute basic statistics ok_results = [r for r in results if r.status == "OK"] metrics_list = [r.get_metrics() for r in ok_results] params_list = [r.get_params() for r in ok_results] if not metrics_list: return {"summary": "All points failed", "ok_count": 0, "total": len(results)} # Compute metric stats def _stats(key: str) -> dict: vals = [m.get(key) for m in metrics_list if m.get(key) is not None] if not vals: return {} return { "min": min(vals), "max": max(vals), "avg": sum(vals) / len(vals), "count": len(vals), } metric_stats = { "tavg_nm": _stats("tavg_nm"), "ripple_pct": _stats("ripple_pct"), "efficiency_pct": _stats("efficiency_pct"), "total_losses_w": _stats("total_losses_w"), } # Find best point by efficiency best_idx = -1 best_eff = -1 for i, m in enumerate(metrics_list): eff = m.get("efficiency_pct", 0) if eff and eff > best_eff: best_eff = eff best_idx = i best_point = None if best_idx >= 0: best_point = { "run_index": ok_results[best_idx].run_index, "params": params_list[best_idx], "metrics": metrics_list[best_idx], } # Simple parameter sensitivity (correlation-like) sensitivity = {} param_keys = set() for p in params_list: param_keys.update(p.keys()) for pk in param_keys: vals = [p.get(pk) for p in params_list if p.get(pk) is not None] if len(vals) < 2: continue effs = [metrics_list[i].get("efficiency_pct", 0) for i, p in enumerate(params_list) if p.get(pk) is not None] if len(effs) < 2: continue # Simple: range of efficiency vs range of param p_range = max(vals) - min(vals) e_range = max(effs) - min(effs) if p_range > 0: sensitivity[pk] = round(e_range / p_range, 4) # Boundary condition check plan_data = plan.get_plan_dict() bc = plan_data.get("acceptance_criteria", {}) constraints = bc.get("hard_constraints", []) satisfied = [] violated = [] for c in constraints: # Simple parse: "metric >= value" or "metric <= value" parts = c.replace(">=", ">=").replace("<=", "<=").split() if len(parts) >= 3: metric, op, val = parts[0], parts[1], float(parts[2]) stat = metric_stats.get(metric, {}) if stat: if op == ">=" and stat.get("max", 0) >= val: satisfied.append(c) elif op == "<=" and stat.get("min", 999) <= val: satisfied.append(c) else: violated.append(c) return { "total_points": len(results), "ok_count": len(ok_results), "failed_count": len(results) - len(ok_results), "metric_stats": metric_stats, "best_point": best_point, "sensitivity": sensitivity, "constraints_satisfied": satisfied, "constraints_violated": violated, "recommendations": _generate_recommendations(metric_stats, sensitivity, best_point, violated), } def _generate_recommendations(metric_stats: dict, sensitivity: dict, best_point: dict, violated: list) -> list[str]: """Generate simple recommendations based on analysis results.""" recs = [] eff = metric_stats.get("efficiency_pct", {}) ripple = metric_stats.get("ripple_pct", {}) if eff and eff.get("max", 0) < 90: recs.append("\u6700\u9ad8\u6548\u7387\u4f4e\u4e8e90%\uff0c\u5efa\u8bae\u51cf\u5c0f\u6c14\u9699\u6216\u589e\u52a0\u78c1\u94a2\u539a\u5ea6\u4ee5\u63d0\u5347\u8f6c\u77e9\u5bc6\u5ea6") if ripple and ripple.get("min", 100) > 5: recs.append("\u8f6c\u77e9\u8109\u52a8\u504f\u9ad8(>5%)\uff0c\u5efa\u8bae\u626b\u63cf\u78c1\u94a2\u6781\u5f27\u89d2(Magnet_Arc)\u4f18\u5316\u8109\u52a8") if sensitivity: top_sens = sorted(sensitivity.items(), key=lambda x: abs(x[1]), reverse=True)[:3] for pk, sv in top_sens: direction = "\u589e\u5927" if sv > 0 else "\u51cf\u5c0f" recs.append(f"{pk}\u5bf9\u6548\u7387\u5f71\u54cd\u663e\u8457(\u7075\u654f\u5ea6={sv})\uff0c\u5efa\u8bae\u4e0b\u4e00\u8f6e{direction}\u8be5\u53c2\u6570\u8303\u56f4") if best_point: recs.append(f"\u5f53\u524d\u6700\u4f18\u70b9: \u6548\u7387{best_point['metrics'].get('efficiency_pct', '?')}%, \u5efa\u8bae\u4ee5\u8be5\u70b9\u53c2\u6570\u4e3a\u4e2d\u5fc3\u7f29\u5c0f\u641c\u7d22\u8303\u56f4") if violated: recs.append(f"\u6709{len(violated)}\u9879\u7ea6\u675f\u672a\u6ee1\u8db3\uff0c\u5efa\u8bae\u8c03\u6574\u626b\u63cf\u8303\u56f4\u6216\u56fa\u5b9a\u53c2\u6570") if not recs: recs.append("\u7ed3\u679c\u826f\u597d\uff0c\u5efa\u8bae\u4ee5\u5f53\u524d\u6700\u4f18\u70b9\u4e3a\u4e2d\u5fc3\u8fdb\u884c\u7cbe\u7ec6\u5316\u626b\u63cf") return recs @router.post("/{plan_id}/generate-iteration") def generate_iteration_plan(plan_id: int, db: Session = Depends(get_db)): """Generate next iteration plan based on current results. Uses AI analysis to adjust scan ranges and creates a new plan with parent_plan_id linking to the current plan. """ plan = db.query(SimulationPlan).filter(SimulationPlan.id == plan_id).first() if not plan: raise HTTPException(status_code=404, detail="Plan not found") results = ( db.query(SimulationResult) .filter(SimulationResult.plan_id == plan_id) .order_by(SimulationResult.run_index.asc()) .all() ) if not results: raise HTTPException(status_code=400, detail="No simulation results for iteration") plan_data = plan.get_plan_dict() ok_results = [r for r in results if r.status == "OK"] if not ok_results: raise HTTPException(status_code=400, detail="No successful results for iteration") # Find best point best = max(ok_results, key=lambda r: r.get_metrics().get("efficiency_pct", 0)) best_params = best.get_params() best_metrics = best.get_metrics() # Generate new variables: narrow ranges around best point new_variables = [] for v in plan_data.get("variables", []): name = v.get("name", "") if name in best_params: best_val = best_params[name] step = v.get("step", 0.1) # Narrow to +/- 2 steps around best new_start = round(best_val - 2 * step, 6) new_stop = round(best_val + 2 * step, 6) # Ensure within physical bounds new_start = max(new_start, v.get("start", new_start)) new_stop = min(new_stop, v.get("stop", new_stop)) values = [] if new_stop > new_start and step > 0: count = int((new_stop - new_start) / step) + 1 values = [round(new_start + i * step, 6) for i in range(count)] new_variables.append({ **v, "start": new_start, "stop": new_stop, "values": values, }) else: new_variables.append(v) # Create new plan import uuid as _uuid new_plan_id = f"SP-{datetime.now().strftime('%Y%m%d-%H%M%S')}-{_uuid.uuid4().hex[:6]}" iteration = (plan_data.get("iteration", 1) or 1) + 1 new_plan_data = { **plan_data, "plan_id": new_plan_id, "iteration": iteration, "parent_plan_id": plan.plan_id, "variables": new_variables, "ai_reasoning": f"Iteration #{iteration}: Narrowed search around best point " f"(eff={best_metrics.get('efficiency_pct', '?')}%, " f"torque={best_metrics.get('tavg_nm', '?')}Nm). " f"Previous best params: {best_params}", } estimated_points = 1 for v in new_variables: estimated_points *= len(v.get("values", [])) if v.get("values") else 1 variables_summary = {} for v in new_variables: variables_summary[v["name"]] = { "unit": v.get("unit", ""), "values": v.get("values", []), "count": len(v.get("values", [])), } new_plan = SimulationPlan( project_id=plan.project_id, name=f"{plan.name}_iter{iteration}", plan_id=new_plan_id, status="draft", estimated_points=estimated_points, estimated_time_min=estimated_points * 3, notes=f"Iteration #{iteration} from plan {plan.plan_id}. Best eff={best_metrics.get('efficiency_pct', '?')}%", ) new_plan.set_plan_dict(new_plan_data) new_plan.variables_summary = json.dumps(variables_summary, ensure_ascii=False) db.add(new_plan) db.commit() db.refresh(new_plan) return { "id": new_plan.id, "plan_id": new_plan.plan_id, "name": new_plan.name, "iteration": iteration, "parent_plan_id": plan.plan_id, "estimated_points": estimated_points, "best_point": { "run_index": best.run_index, "params": best_params, "metrics": best_metrics, }, "message": f"\u8fed\u4ee3\u65b9\u6848\u5df2\u751f\u6210\uff0c\u56f4\u7ed5\u6700\u4f18\u70b9\u7f29\u5c0f\u641c\u7d22\u8303\u56f4\uff0c\u5171{estimated_points}\u4e2a\u6570\u636e\u70b9", }