pcb-afm-simulation-system/web/backend/app/services/analytics.py

"""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