|
|
@@ -0,0 +1,325 @@
|
|
|
+"""Experience database for PCB axial flux motor automated simulation.
|
|
|
+
|
|
|
+Lightweight SQLite-based experience database for Phase 1:
|
|
|
+- Auto-accumulate simulation results after each scan
|
|
|
+- Similar case retrieval by topology + parameter proximity
|
|
|
+- Feedback: recommend next-round parameter ranges based on results
|
|
|
+
|
|
|
+All source is ASCII.
|
|
|
+"""
|
|
|
+
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import json
|
|
|
+import math
|
|
|
+import sqlite3
|
|
|
+from datetime import datetime
|
|
|
+from pathlib import Path
|
|
|
+from typing import Any
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+# Database schema
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+
|
|
|
+SCHEMA = """
|
|
|
+CREATE TABLE IF NOT EXISTS runs (
|
|
|
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
|
+ plan_id TEXT NOT NULL,
|
|
|
+ topology TEXT NOT NULL DEFAULT 'SSSR',
|
|
|
+ model_path TEXT,
|
|
|
+ timestamp TEXT NOT NULL,
|
|
|
+ status TEXT NOT NULL DEFAULT 'OK',
|
|
|
+ params_json TEXT NOT NULL,
|
|
|
+ metrics_json TEXT NOT NULL,
|
|
|
+ solve_time_s REAL,
|
|
|
+ notes TEXT
|
|
|
+);
|
|
|
+
|
|
|
+CREATE INDEX IF NOT EXISTS idx_runs_topology ON runs(topology);
|
|
|
+CREATE INDEX IF NOT EXISTS idx_runs_timestamp ON runs(timestamp);
|
|
|
+CREATE INDEX IF NOT EXISTS idx_runs_plan_id ON runs(plan_id);
|
|
|
+"""
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+# Experience database
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+
|
|
|
+class ExperienceDB:
|
|
|
+ """SQLite-based experience database for simulation results."""
|
|
|
+
|
|
|
+ def __init__(self, db_path: str | Path = "experience/experience.db"):
|
|
|
+ self.db_path = Path(db_path)
|
|
|
+ self.db_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
+ self._conn = sqlite3.connect(str(self.db_path))
|
|
|
+ self._conn.row_factory = sqlite3.Row
|
|
|
+ self._conn.executescript(SCHEMA)
|
|
|
+ self._conn.commit()
|
|
|
+
|
|
|
+ def close(self) -> None:
|
|
|
+ self._conn.close()
|
|
|
+
|
|
|
+ def __enter__(self):
|
|
|
+ return self
|
|
|
+
|
|
|
+ def __exit__(self, *args):
|
|
|
+ self.close()
|
|
|
+
|
|
|
+ # -- Insertion --
|
|
|
+
|
|
|
+ def add_run(
|
|
|
+ self,
|
|
|
+ plan_id: str,
|
|
|
+ params: dict[str, float],
|
|
|
+ metrics: dict[str, float],
|
|
|
+ topology: str = "SSSR",
|
|
|
+ model_path: str = "",
|
|
|
+ status: str = "OK",
|
|
|
+ solve_time_s: float = 0.0,
|
|
|
+ notes: str = "",
|
|
|
+ ) -> int:
|
|
|
+ """Add a single simulation run result. Returns the run ID."""
|
|
|
+ cursor = self._conn.execute(
|
|
|
+ """INSERT INTO runs (plan_id, topology, model_path, timestamp, status,
|
|
|
+ params_json, metrics_json, solve_time_s, notes)
|
|
|
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
|
+ (
|
|
|
+ plan_id,
|
|
|
+ topology,
|
|
|
+ model_path,
|
|
|
+ datetime.now().isoformat(timespec="seconds"),
|
|
|
+ status,
|
|
|
+ json.dumps(params),
|
|
|
+ json.dumps(metrics),
|
|
|
+ solve_time_s,
|
|
|
+ notes,
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ self._conn.commit()
|
|
|
+ return cursor.lastrowid
|
|
|
+
|
|
|
+ def add_runs_from_csv(self, csv_path: str | Path, plan_id: str,
|
|
|
+ topology: str = "SSSR", model_path: str = "") -> int:
|
|
|
+ """Add all runs from a scan results CSV. Returns number added."""
|
|
|
+ import csv as csv_mod
|
|
|
+ count = 0
|
|
|
+ with open(csv_path, "r", encoding="utf-8-sig", newline="") as f:
|
|
|
+ reader = csv_mod.DictReader(f)
|
|
|
+ for row in reader:
|
|
|
+ if row.get("status") != "OK":
|
|
|
+ continue
|
|
|
+ # Extract params (all columns that are not standard fields/metrics)
|
|
|
+ standard = {"run_index", "status", "seconds", "error"}
|
|
|
+ from .solver_core import METRIC_KEYS
|
|
|
+ metric_set = set(METRIC_KEYS)
|
|
|
+ params = {}
|
|
|
+ metrics = {}
|
|
|
+ for key, val in row.items():
|
|
|
+ if key in standard or not val:
|
|
|
+ continue
|
|
|
+ try:
|
|
|
+ fval = float(val)
|
|
|
+ except (ValueError, TypeError):
|
|
|
+ continue
|
|
|
+ if key in metric_set:
|
|
|
+ metrics[key] = fval
|
|
|
+ else:
|
|
|
+ params[key] = fval
|
|
|
+ self.add_run(
|
|
|
+ plan_id=plan_id, params=params, metrics=metrics,
|
|
|
+ topology=topology, model_path=model_path,
|
|
|
+ status=row.get("status", "OK"),
|
|
|
+ solve_time_s=float(row.get("seconds", 0) or 0),
|
|
|
+ )
|
|
|
+ count += 1
|
|
|
+ return count
|
|
|
+
|
|
|
+ # -- Query --
|
|
|
+
|
|
|
+ def get_run(self, run_id: int) -> dict[str, Any] | None:
|
|
|
+ """Get a single run by ID."""
|
|
|
+ row = self._conn.execute("SELECT * FROM runs WHERE id = ?", (run_id,)).fetchone()
|
|
|
+ return self._row_to_dict(row) if row else None
|
|
|
+
|
|
|
+ def get_all_runs(self, topology: str | None = None,
|
|
|
+ limit: int = 100) -> list[dict[str, Any]]:
|
|
|
+ """Get all runs, optionally filtered by topology."""
|
|
|
+ if topology:
|
|
|
+ rows = self._conn.execute(
|
|
|
+ "SELECT * FROM runs WHERE topology = ? ORDER BY timestamp DESC LIMIT ?",
|
|
|
+ (topology, limit),
|
|
|
+ ).fetchall()
|
|
|
+ else:
|
|
|
+ rows = self._conn.execute(
|
|
|
+ "SELECT * FROM runs ORDER BY timestamp DESC LIMIT ?",
|
|
|
+ (limit,),
|
|
|
+ ).fetchall()
|
|
|
+ return [self._row_to_dict(r) for r in rows]
|
|
|
+
|
|
|
+ def find_similar(
|
|
|
+ self,
|
|
|
+ params: dict[str, float],
|
|
|
+ topology: str = "SSSR",
|
|
|
+ top_n: int = 5,
|
|
|
+ tolerance: float = 0.5,
|
|
|
+ ) -> list[dict[str, Any]]:
|
|
|
+ """Find similar runs based on parameter proximity.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ params: target parameter values to match against.
|
|
|
+ topology: filter by topology.
|
|
|
+ top_n: number of results to return.
|
|
|
+ tolerance: relative tolerance for parameter matching (0.5 = 50%).
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ List of similar runs, sorted by similarity score (closest first).
|
|
|
+ """
|
|
|
+ candidates = self.get_all_runs(topology=topology, limit=500)
|
|
|
+ scored = []
|
|
|
+ for run in candidates:
|
|
|
+ run_params = run.get("params", {})
|
|
|
+ # Compute average relative distance across shared parameters
|
|
|
+ distances = []
|
|
|
+ shared = 0
|
|
|
+ for key, target_val in params.items():
|
|
|
+ if key in run_params and target_val != 0:
|
|
|
+ rel_dist = abs(run_params[key] - target_val) / abs(target_val)
|
|
|
+ distances.append(rel_dist)
|
|
|
+ shared += 1
|
|
|
+ if shared == 0:
|
|
|
+ continue
|
|
|
+ avg_dist = sum(distances) / len(distances)
|
|
|
+ if avg_dist <= tolerance:
|
|
|
+ scored.append((avg_dist, run))
|
|
|
+ scored.sort(key=lambda x: x[0])
|
|
|
+ return [run for _, run in scored[:top_n]]
|
|
|
+
|
|
|
+ def get_stats(self, topology: str | None = None) -> dict[str, Any]:
|
|
|
+ """Get database statistics."""
|
|
|
+ if topology:
|
|
|
+ total = self._conn.execute(
|
|
|
+ "SELECT COUNT(*) FROM runs WHERE topology = ?", (topology,)
|
|
|
+ ).fetchone()[0]
|
|
|
+ else:
|
|
|
+ total = self._conn.execute("SELECT COUNT(*) FROM runs").fetchone()[0]
|
|
|
+ plans = self._conn.execute(
|
|
|
+ "SELECT COUNT(DISTINCT plan_id) FROM runs"
|
|
|
+ ).fetchone()[0]
|
|
|
+ return {"total_runs": total, "distinct_plans": plans, "db_path": str(self.db_path)}
|
|
|
+
|
|
|
+ # -- Feedback / Recommendation --
|
|
|
+
|
|
|
+ def recommend_next_round(
|
|
|
+ self,
|
|
|
+ params: dict[str, float],
|
|
|
+ metrics: dict[str, float],
|
|
|
+ target_metric: str = "efficiency_pct",
|
|
|
+ direction: str = "maximize",
|
|
|
+ topology: str = "SSSR",
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ """Recommend next-round parameter ranges based on current results.
|
|
|
+
|
|
|
+ Uses simple trend analysis: find similar runs in the experience DB,
|
|
|
+ compare their metric values, and recommend parameter directions.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ params: current parameter values.
|
|
|
+ metrics: current metric values.
|
|
|
+ target_metric: metric to optimize.
|
|
|
+ direction: 'maximize' or 'minimize'.
|
|
|
+ topology: motor topology.
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ dict with recommendations for each parameter.
|
|
|
+ """
|
|
|
+ similar = self.find_similar(params, topology=topology, top_n=10, tolerance=1.0)
|
|
|
+ if not similar:
|
|
|
+ return {
|
|
|
+ "status": "insufficient_data",
|
|
|
+ "message": "Not enough similar runs in experience DB for recommendation. "
|
|
|
+ "Accumulate more simulation results first.",
|
|
|
+ "recommendations": {},
|
|
|
+ }
|
|
|
+
|
|
|
+ current_val = metrics.get(target_metric)
|
|
|
+ if current_val is None:
|
|
|
+ return {
|
|
|
+ "status": "missing_metric",
|
|
|
+ "message": f"Target metric '{target_metric}' not found in current results.",
|
|
|
+ "recommendations": {},
|
|
|
+ }
|
|
|
+
|
|
|
+ # Analyze each parameter: find runs where this parameter differs
|
|
|
+ recommendations = {}
|
|
|
+ for param_name, current_param_val in params.items():
|
|
|
+ # Collect (param_value, metric_value) pairs from similar runs
|
|
|
+ pairs = []
|
|
|
+ for run in similar:
|
|
|
+ if param_name in run["params"] and target_metric in run["metrics"]:
|
|
|
+ pairs.append((
|
|
|
+ run["params"][param_name],
|
|
|
+ run["metrics"][target_metric],
|
|
|
+ ))
|
|
|
+ if len(pairs) < 2:
|
|
|
+ recommendations[param_name] = {
|
|
|
+ "status": "insufficient_data",
|
|
|
+ "message": f"Not enough data points for parameter '{param_name}'.",
|
|
|
+ }
|
|
|
+ continue
|
|
|
+
|
|
|
+ # Simple linear trend: compute correlation direction
|
|
|
+ pairs.sort(key=lambda x: x[0])
|
|
|
+ n = len(pairs)
|
|
|
+ mean_x = sum(p[0] for p in pairs) / n
|
|
|
+ mean_y = sum(p[1] for p in pairs) / n
|
|
|
+ cov = sum((p[0] - mean_x) * (p[1] - mean_y) for p in pairs) / n
|
|
|
+ var_x = sum((p[0] - mean_x) ** 2 for p in pairs) / n
|
|
|
+ if var_x == 0:
|
|
|
+ recommendations[param_name] = {
|
|
|
+ "status": "no_variation",
|
|
|
+ "message": f"Parameter '{param_name}' has no variation in similar runs.",
|
|
|
+ }
|
|
|
+ continue
|
|
|
+ slope = cov / var_x # metric change per unit parameter change
|
|
|
+
|
|
|
+ # Recommend direction
|
|
|
+ if direction == "maximize":
|
|
|
+ recommend_increase = slope > 0
|
|
|
+ else:
|
|
|
+ recommend_increase = slope < 0
|
|
|
+
|
|
|
+ # Suggest new range: current value +/- 20%, biased toward recommendation
|
|
|
+ delta = abs(current_param_val) * 0.2 if current_param_val != 0 else 0.5
|
|
|
+ if recommend_increase:
|
|
|
+ new_range = [current_param_val, current_param_val + delta * 2]
|
|
|
+ else:
|
|
|
+ new_range = [max(0, current_param_val - delta * 2), current_param_val]
|
|
|
+
|
|
|
+ recommendations[param_name] = {
|
|
|
+ "status": "ok",
|
|
|
+ "current_value": current_param_val,
|
|
|
+ "trend_slope": round(slope, 6),
|
|
|
+ "recommend_increase": recommend_increase,
|
|
|
+ "suggested_range": [round(new_range[0], 4), round(new_range[1], 4)],
|
|
|
+ "data_points": n,
|
|
|
+ }
|
|
|
+
|
|
|
+ return {
|
|
|
+ "status": "ok",
|
|
|
+ "target_metric": target_metric,
|
|
|
+ "direction": direction,
|
|
|
+ "current_value": current_val,
|
|
|
+ "similar_runs_used": len(similar),
|
|
|
+ "recommendations": recommendations,
|
|
|
+ }
|
|
|
+
|
|
|
+ # -- Internal --
|
|
|
+
|
|
|
+ def _row_to_dict(self, row: sqlite3.Row) -> dict[str, Any]:
|
|
|
+ d = dict(row)
|
|
|
+ d["params"] = json.loads(d.get("params_json", "{}"))
|
|
|
+ d["metrics"] = json.loads(d.get("metrics_json", "{}"))
|
|
|
+ del d["params_json"]
|
|
|
+ del d["metrics_json"]
|
|
|
+ return d
|
javen.ye