|
|
@@ -0,0 +1,395 @@
|
|
|
+"""Analytics service for experience library and simulation results.
|
|
|
+
|
|
|
+Provides:
|
|
|
+- Experience library statistics (count, topology distribution, metric ranges)
|
|
|
+- Similar case retrieval (parameter distance matching)
|
|
|
+- Result trend analysis (parameter vs metric scatter/line data)
|
|
|
+- Pareto frontier (efficiency vs torque, losses vs torque)
|
|
|
+- Parameter sensitivity ranking (correlation-based)
|
|
|
+
|
|
|
+All source is ASCII.
|
|
|
+"""
|
|
|
+
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import math
|
|
|
+from typing import Any
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+# Metric definitions (for display and analysis)
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+
|
|
|
+METRIC_DEFS = {
|
|
|
+ "tavg_nm": {"label": "Average Torque", "unit": "Nm", "higher_is_better": True},
|
|
|
+ "ripple_pct": {"label": "Torque Ripple", "unit": "%", "higher_is_better": False},
|
|
|
+ "efficiency_pct": {"label": "Efficiency", "unit": "%", "higher_is_better": True},
|
|
|
+ "total_losses_w": {"label": "Total Losses", "unit": "W", "higher_is_better": False},
|
|
|
+ "copper_loss_w": {"label": "Copper Loss", "unit": "W", "higher_is_better": False},
|
|
|
+ "iron_loss_w": {"label": "Iron Loss", "unit": "W", "higher_is_better": False},
|
|
|
+ "magnet_loss_w": {"label": "Magnet Loss", "unit": "W", "higher_is_better": False},
|
|
|
+ "back_emf_v": {"label": "Back EMF", "unit": "V", "higher_is_better": True},
|
|
|
+ "output_power_w": {"label": "Output Power", "unit": "W", "higher_is_better": True},
|
|
|
+ "input_power_w": {"label": "Input Power", "unit": "W", "higher_is_better": True},
|
|
|
+ "no_load_speed_rpm": {"label": "No-Load Speed", "unit": "rpm", "higher_is_better": True},
|
|
|
+}
|
|
|
+
|
|
|
+
|
|
|
+def get_metric_defs() -> list[dict[str, Any]]:
|
|
|
+ """Return metric definitions for frontend display."""
|
|
|
+ return [
|
|
|
+ {"key": k, "label": v["label"], "unit": v["unit"], "higher_is_better": v["higher_is_better"]}
|
|
|
+ for k, v in METRIC_DEFS.items()
|
|
|
+ ]
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+# Experience library statistics
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+
|
|
|
+def compute_experience_stats(cases: list[dict[str, Any]]) -> dict[str, Any]:
|
|
|
+ """Compute aggregate statistics from experience cases.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ cases: List of experience case dicts (from _case_to_dict).
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ Stats dict with total, topology distribution, metric ranges, param coverage.
|
|
|
+ """
|
|
|
+ total = len(cases)
|
|
|
+ if total == 0:
|
|
|
+ return {
|
|
|
+ "total": 0,
|
|
|
+ "topology_distribution": {},
|
|
|
+ "metric_ranges": {},
|
|
|
+ "param_coverage": {},
|
|
|
+ "avg_rating": 0,
|
|
|
+ }
|
|
|
+
|
|
|
+ # Topology distribution
|
|
|
+ topology_dist: dict[str, int] = {}
|
|
|
+ for c in cases:
|
|
|
+ topo = c.get("topology", "Unknown")
|
|
|
+ topology_dist[topo] = topology_dist.get(topo, 0) + 1
|
|
|
+
|
|
|
+ # Metric ranges
|
|
|
+ metric_ranges: dict[str, dict[str, float]] = {}
|
|
|
+ for c in cases:
|
|
|
+ metrics = c.get("metrics", {}) or {}
|
|
|
+ for key, val in metrics.items():
|
|
|
+ if not isinstance(val, (int, float)) or math.isnan(val):
|
|
|
+ continue
|
|
|
+ if key not in metric_ranges:
|
|
|
+ metric_ranges[key] = {"min": val, "max": val, "sum": 0.0, "count": 0}
|
|
|
+ r = metric_ranges[key]
|
|
|
+ r["min"] = min(r["min"], val)
|
|
|
+ r["max"] = max(r["max"], val)
|
|
|
+ r["sum"] += val
|
|
|
+ r["count"] += 1
|
|
|
+
|
|
|
+ # Compute averages
|
|
|
+ for key, r in metric_ranges.items():
|
|
|
+ r["avg"] = round(r["sum"] / r["count"], 4) if r["count"] > 0 else 0
|
|
|
+ del r["sum"]
|
|
|
+
|
|
|
+ # Parameter coverage (which params appear in how many cases)
|
|
|
+ param_coverage: dict[str, int] = {}
|
|
|
+ for c in cases:
|
|
|
+ params = c.get("params", {}) or {}
|
|
|
+ for key in params:
|
|
|
+ param_coverage[key] = param_coverage.get(key, 0) + 1
|
|
|
+
|
|
|
+ # Average rating
|
|
|
+ ratings = [c.get("rating", 0) for c in cases if c.get("rating", 0) > 0]
|
|
|
+ avg_rating = round(sum(ratings) / len(ratings), 2) if ratings else 0
|
|
|
+
|
|
|
+ return {
|
|
|
+ "total": total,
|
|
|
+ "topology_distribution": topology_dist,
|
|
|
+ "metric_ranges": metric_ranges,
|
|
|
+ "param_coverage": param_coverage,
|
|
|
+ "avg_rating": avg_rating,
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+# Similar case retrieval
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+
|
|
|
+def find_similar_cases(
|
|
|
+ target_params: dict[str, float],
|
|
|
+ cases: list[dict[str, Any]],
|
|
|
+ top_k: int = 5,
|
|
|
+ tolerance: float = 0.3,
|
|
|
+) -> list[dict[str, Any]]:
|
|
|
+ """Find experience cases similar to target parameters.
|
|
|
+
|
|
|
+ Uses normalized Euclidean distance on shared parameters.
|
|
|
+ Only cases sharing at least one parameter with the target are considered.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ target_params: Dict of parameter_name -> value.
|
|
|
+ cases: List of experience case dicts.
|
|
|
+ top_k: Number of top matches to return.
|
|
|
+ tolerance: Maximum normalized distance to consider (0.0-1.0).
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ List of similar cases with added 'similarity_score' field (0-1, higher=more similar).
|
|
|
+ """
|
|
|
+ if not target_params or not cases:
|
|
|
+ return []
|
|
|
+
|
|
|
+ # Compute value ranges for normalization
|
|
|
+ all_params: dict[str, list[float]] = {}
|
|
|
+ for c in cases:
|
|
|
+ params = c.get("params", {}) or {}
|
|
|
+ for key, val in params.items():
|
|
|
+ if isinstance(val, (int, float)):
|
|
|
+ all_params.setdefault(key, []).append(float(val))
|
|
|
+
|
|
|
+ param_ranges: dict[str, float] = {}
|
|
|
+ for key, vals in all_params.items():
|
|
|
+ rng = max(vals) - min(vals) if len(vals) > 1 else 1.0
|
|
|
+ param_ranges[key] = rng if rng > 0 else 1.0
|
|
|
+
|
|
|
+ scored = []
|
|
|
+ for c in cases:
|
|
|
+ params = c.get("params", {}) or {}
|
|
|
+ shared = [k for k in target_params if k in params]
|
|
|
+ if not shared:
|
|
|
+ continue
|
|
|
+
|
|
|
+ # Normalized Euclidean distance
|
|
|
+ dist_sq = 0.0
|
|
|
+ for key in shared:
|
|
|
+ tv = float(target_params[key])
|
|
|
+ cv = float(params[key])
|
|
|
+ rng = param_ranges.get(key, 1.0)
|
|
|
+ dist_sq += ((tv - cv) / rng) ** 2
|
|
|
+ dist = math.sqrt(dist_sq / len(shared))
|
|
|
+
|
|
|
+ if dist > tolerance:
|
|
|
+ continue
|
|
|
+
|
|
|
+ similarity = max(0.0, 1.0 - dist)
|
|
|
+ result = dict(c)
|
|
|
+ result["similarity_score"] = round(similarity, 4)
|
|
|
+ result["shared_params"] = shared
|
|
|
+ scored.append(result)
|
|
|
+
|
|
|
+ scored.sort(key=lambda x: x["similarity_score"], reverse=True)
|
|
|
+ return scored[:top_k]
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+# Result trend analysis
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+
|
|
|
+def compute_trend_data(
|
|
|
+ results: list[dict[str, Any]],
|
|
|
+ param_key: str,
|
|
|
+ metric_key: str,
|
|
|
+) -> dict[str, Any]:
|
|
|
+ """Compute trend data for a parameter vs metric scatter plot.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ results: List of simulation result dicts (with params and metrics).
|
|
|
+ param_key: Parameter name for X axis.
|
|
|
+ metric_key: Metric name for Y axis.
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ Dict with x_key, y_key, points (list of [x, y] sorted by x), and stats.
|
|
|
+ """
|
|
|
+ points = []
|
|
|
+ for r in results:
|
|
|
+ if r.get("status") != "OK":
|
|
|
+ continue
|
|
|
+ params = r.get("params", {}) or {}
|
|
|
+ metrics = r.get("metrics", {}) or {}
|
|
|
+ if param_key in params and metric_key in metrics:
|
|
|
+ try:
|
|
|
+ x = float(params[param_key])
|
|
|
+ y = float(metrics[metric_key])
|
|
|
+ if not math.isnan(x) and not math.isnan(y):
|
|
|
+ points.append([x, y])
|
|
|
+ except (ValueError, TypeError):
|
|
|
+ continue
|
|
|
+
|
|
|
+ points.sort(key=lambda p: p[0])
|
|
|
+
|
|
|
+ # Basic stats
|
|
|
+ if points:
|
|
|
+ ys = [p[1] for p in points]
|
|
|
+ stats = {
|
|
|
+ "count": len(points),
|
|
|
+ "min": round(min(ys), 4),
|
|
|
+ "max": round(max(ys), 4),
|
|
|
+ "avg": round(sum(ys) / len(ys), 4),
|
|
|
+ }
|
|
|
+ else:
|
|
|
+ stats = {"count": 0, "min": 0, "max": 0, "avg": 0}
|
|
|
+
|
|
|
+ return {
|
|
|
+ "x_key": param_key,
|
|
|
+ "y_key": metric_key,
|
|
|
+ "x_label": METRIC_DEFS.get(param_key, {}).get("label", param_key),
|
|
|
+ "y_label": METRIC_DEFS.get(metric_key, {}).get("label", metric_key),
|
|
|
+ "y_unit": METRIC_DEFS.get(metric_key, {}).get("unit", ""),
|
|
|
+ "points": points,
|
|
|
+ "stats": stats,
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+# Pareto frontier
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+
|
|
|
+def compute_pareto_frontier(
|
|
|
+ results: list[dict[str, Any]],
|
|
|
+ x_metric: str = "total_losses_w",
|
|
|
+ y_metric: str = "efficiency_pct",
|
|
|
+) -> dict[str, Any]:
|
|
|
+ """Compute Pareto frontier for two metrics.
|
|
|
+
|
|
|
+ A point is Pareto-optimal if no other point is better in both metrics
|
|
|
+ (considering higher_is_better for each).
|
|
|
+
|
|
|
+ Args:
|
|
|
+ results: List of simulation result dicts.
|
|
|
+ x_metric: Metric for X axis (typically losses, lower is better).
|
|
|
+ y_metric: Metric for Y axis (typically efficiency, higher is better).
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ Dict with all_points, pareto_points, and labels.
|
|
|
+ """
|
|
|
+ x_higher_better = METRIC_DEFS.get(x_metric, {}).get("higher_is_better", True)
|
|
|
+ y_higher_better = METRIC_DEFS.get(y_metric, {}).get("higher_is_better", True)
|
|
|
+
|
|
|
+ all_points = []
|
|
|
+ for r in results:
|
|
|
+ if r.get("status") != "OK":
|
|
|
+ continue
|
|
|
+ metrics = r.get("metrics", {}) or {}
|
|
|
+ if x_metric in metrics and y_metric in metrics:
|
|
|
+ try:
|
|
|
+ x = float(metrics[x_metric])
|
|
|
+ y = float(metrics[y_metric])
|
|
|
+ if not math.isnan(x) and not math.isnan(y):
|
|
|
+ all_points.append({"x": x, "y": y, "params": r.get("params", {})})
|
|
|
+ except (ValueError, TypeError):
|
|
|
+ continue
|
|
|
+
|
|
|
+ # Find Pareto-optimal points
|
|
|
+ pareto = []
|
|
|
+ for i, p in enumerate(all_points):
|
|
|
+ dominated = False
|
|
|
+ for j, q in enumerate(all_points):
|
|
|
+ if i == j:
|
|
|
+ continue
|
|
|
+ # q dominates p if q is better or equal in both, and strictly better in at least one
|
|
|
+ x_better = (q["x"] >= p["x"]) if x_higher_better else (q["x"] <= p["x"])
|
|
|
+ y_better = (q["y"] >= p["y"]) if y_higher_better else (q["y"] <= p["y"])
|
|
|
+ x_strict = (q["x"] > p["x"]) if x_higher_better else (q["x"] < p["x"])
|
|
|
+ y_strict = (q["y"] > p["y"]) if y_higher_better else (q["y"] < p["y"])
|
|
|
+ if x_better and y_better and (x_strict or y_strict):
|
|
|
+ dominated = True
|
|
|
+ break
|
|
|
+ if not dominated:
|
|
|
+ pareto.append(p)
|
|
|
+
|
|
|
+ # Sort pareto by x
|
|
|
+ pareto.sort(key=lambda p: p["x"])
|
|
|
+
|
|
|
+ return {
|
|
|
+ "x_metric": x_metric,
|
|
|
+ "y_metric": y_metric,
|
|
|
+ "x_label": METRIC_DEFS.get(x_metric, {}).get("label", x_metric),
|
|
|
+ "y_label": METRIC_DEFS.get(y_metric, {}).get("label", y_metric),
|
|
|
+ "x_unit": METRIC_DEFS.get(x_metric, {}).get("unit", ""),
|
|
|
+ "y_unit": METRIC_DEFS.get(y_metric, {}).get("unit", ""),
|
|
|
+ "all_points": [{"x": p["x"], "y": p["y"]} for p in all_points],
|
|
|
+ "pareto_points": [{"x": p["x"], "y": p["y"], "params": p["params"]} for p in pareto],
|
|
|
+ "total_count": len(all_points),
|
|
|
+ "pareto_count": len(pareto),
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+# Parameter sensitivity ranking
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
+
|
|
|
+def compute_sensitivity(
|
|
|
+ results: list[dict[str, Any]],
|
|
|
+ metric_key: str,
|
|
|
+) -> list[dict[str, Any]]:
|
|
|
+ """Compute parameter sensitivity ranking for a target metric.
|
|
|
+
|
|
|
+ Uses Pearson correlation coefficient between each parameter and the metric.
|
|
|
+ Absolute value indicates sensitivity strength; sign indicates direction.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ results: List of simulation result dicts.
|
|
|
+ metric_key: Target metric key.
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ List of {param, correlation, abs_correlation, direction} sorted by abs correlation.
|
|
|
+ """
|
|
|
+ # Collect parameter values and metric values
|
|
|
+ param_values: dict[str, list[float]] = {}
|
|
|
+ metric_values: list[float] = []
|
|
|
+
|
|
|
+ for r in results:
|
|
|
+ if r.get("status") != "OK":
|
|
|
+ continue
|
|
|
+ metrics = r.get("metrics", {}) or {}
|
|
|
+ if metric_key not in metrics:
|
|
|
+ continue
|
|
|
+ try:
|
|
|
+ mv = float(metrics[metric_key])
|
|
|
+ except (ValueError, TypeError):
|
|
|
+ continue
|
|
|
+ if math.isnan(mv):
|
|
|
+ continue
|
|
|
+
|
|
|
+ params = r.get("params", {}) or {}
|
|
|
+ for key, val in params.items():
|
|
|
+ try:
|
|
|
+ pv = float(val)
|
|
|
+ if not math.isnan(pv):
|
|
|
+ param_values.setdefault(key, []).append(pv)
|
|
|
+ except (ValueError, TypeError):
|
|
|
+ continue
|
|
|
+ metric_values.append(mv)
|
|
|
+
|
|
|
+ n = len(metric_values)
|
|
|
+ if n < 3:
|
|
|
+ return []
|
|
|
+
|
|
|
+ sensitivities = []
|
|
|
+ for param, pvals in param_values.items():
|
|
|
+ if len(pvals) != n:
|
|
|
+ continue # Must have same number of samples
|
|
|
+
|
|
|
+ # Pearson correlation
|
|
|
+ mean_p = sum(pvals) / n
|
|
|
+ mean_m = sum(metric_values) / n
|
|
|
+ cov = sum((pvals[i] - mean_p) * (metric_values[i] - mean_m) for i in range(n))
|
|
|
+ var_p = sum((p - mean_p) ** 2 for p in pvals)
|
|
|
+ var_m = sum((m - mean_m) ** 2 for m in metric_values)
|
|
|
+
|
|
|
+ if var_p == 0 or var_m == 0:
|
|
|
+ corr = 0.0
|
|
|
+ else:
|
|
|
+ corr = cov / math.sqrt(var_p * var_m)
|
|
|
+
|
|
|
+ direction = "positive" if corr >= 0 else "negative"
|
|
|
+ sensitivities.append({
|
|
|
+ "param": param,
|
|
|
+ "correlation": round(corr, 4),
|
|
|
+ "abs_correlation": round(abs(corr), 4),
|
|
|
+ "direction": direction,
|
|
|
+ "sample_count": n,
|
|
|
+ })
|
|
|
+
|
|
|
+ sensitivities.sort(key=lambda x: x["abs_correlation"], reverse=True)
|
|
|
+ return sensitivities
|
carlin