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

"""AI Result Analyst + Multi-Fidelity Calibration + Confidence Grading (P3-M4).

Analyzes simulation results using Kimi k3 model, with:
- Multi-fidelity result calibration (L0-L4 bias correction)
- Confidence grade assessment (A-D)
- Six convergence criteria checks
- Natural language result interpretation
"""
import json
import math
from pathlib import Path
from typing import Dict, List, Optional, Any, Tuple
from datetime import datetime

from ..config import PROMPTS_DIR
from ..services.ai_client import get_kimi_client


# Fidelity level definitions (per third-party review)
FIDELITY_LEVELS = {
    "L0": {"name": "Analytic Pre-screen", "bias_pct": 20.0, "cost": 0.01},
    "L1": {"name": "Magnetic Equivalent Circuit", "bias_pct": 15.0, "cost": 0.1},
    "L2": {"name": "2D FEA (Maxwell)", "bias_pct": 8.0, "cost": 1.0},
    "L3": {"name": "3D FEA (Motor-CAD)", "bias_pct": 3.0, "cost": 5.0},
    "L4": {"name": "3D Transient + Thermal Coupled", "bias_pct": 1.0, "cost": 20.0},
}


class ConvergenceChecker:
    """Six convergence criteria checker (per third-party review)."""

    @staticmethod
    def check_objective_stability(values: List[float], window: int = 5, threshold: float = 0.01) -> Dict[str, Any]:
        """Criterion 1: Objective function change rate < threshold."""
        if len(values) < window:
            return {"passed": False, "reason": "insufficient_data", "value": None}
        recent = values[-window:]
        if max(recent) == 0:
            return {"passed": True, "reason": "zero_objective", "value": 0.0}
        change_rate = abs(max(recent) - min(recent)) / abs(max(recent))
        return {"passed": change_rate < threshold, "reason": "change_rate", "value": round(change_rate, 4)}

    @staticmethod
    def check_optimum_stability(locations: List[Dict[str, float]], window: int = 5, threshold: float = 0.1) -> Dict[str, Any]:
        """Criterion 2: Optimum location stability."""
        if len(locations) < window:
            return {"passed": False, "reason": "insufficient_data", "value": None}
        recent = locations[-window:]
        # Calculate average distance between consecutive optimum locations
        distances = []
        for i in range(1, len(recent)):
            dist_sq = 0.0
            for key in recent[i]:
                if key in recent[i-1]:
                    dist_sq += (recent[i][key] - recent[i-1][key]) ** 2
            distances.append(math.sqrt(dist_sq))
        if not distances:
            return {"passed": True, "reason": "single_location", "value": 0.0}
        avg_dist = sum(distances) / len(distances)
        return {"passed": avg_dist < threshold, "reason": "location_drift", "value": round(avg_dist, 4)}

    @staticmethod
    def check_surrogate_error(predictions: List[float], actuals: List[float], threshold: float = 0.05) -> Dict[str, Any]:
        """Criterion 3: Surrogate model prediction error < 5%."""
        if len(predictions) < 3 or len(predictions) != len(actuals):
            return {"passed": False, "reason": "insufficient_data", "value": None}
        errors = []
        for p, a in zip(predictions, actuals):
            if abs(a) > 1e-10:
                errors.append(abs(p - a) / abs(a))
        if not errors:
            return {"passed": False, "reason": "zero_actuals", "value": None}
        avg_error = sum(errors) / len(errors)
        return {"passed": avg_error < threshold, "reason": "avg_relative_error", "value": round(avg_error, 4)}

    @staticmethod
    def check_constraint_satisfaction(feasible_count: int, total_count: int, threshold: float = 0.95) -> Dict[str, Any]:
        """Criterion 4: Constraint satisfaction rate > 95%."""
        if total_count == 0:
            return {"passed": False, "reason": "no_data", "value": None}
        rate = feasible_count / total_count
        return {"passed": rate > threshold, "reason": "feasibility_rate", "value": round(rate, 4)}

    @staticmethod
    def check_sample_density(n_samples: int, n_dimensions: int, threshold: int = 10) -> Dict[str, Any]:
        """Criterion 5: Sample density sufficient (>10 points per dimension)."""
        if n_dimensions == 0:
            return {"passed": False, "reason": "no_dimensions", "value": None}
        density = n_samples / n_dimensions
        return {"passed": density >= threshold, "reason": "points_per_dimension", "value": round(density, 1)}

    @staticmethod
    def check_physical_consistency(metrics: Dict[str, float], bounds: Dict[str, Tuple[float, float]]) -> Dict[str, Any]:
        """Criterion 6: Physical consistency check."""
        violations = []
        for key, (low, high) in bounds.items():
            if key in metrics:
                value = metrics[key]
                if value < low or value > high:
                    violations.append(f"{key}={value} outside [{low}, {high}]")
        return {"passed": len(violations) == 0, "reason": "physical_bounds", "violations": violations}


class MultiFidelityCalibrator:
    """Multi-fidelity result calibration (L0-L4 bias correction).

    Per third-party review: different fidelity levels have known biases.
    Low-fidelity results are corrected using calibration factors derived
    from paired high-low fidelity comparisons.
    """

    def __init__(self):
        # Default calibration factors (will be refined with actual data)
        # Format: {metric: {fidelity: correction_factor}}
        self.calibration_factors = {
            "tavg_nm": {"L0": 1.15, "L1": 1.10, "L2": 1.05, "L3": 1.01, "L4": 1.00},
            "efficiency_pct": {"L0": 0.95, "L1": 0.97, "L2": 0.99, "L3": 0.995, "L4": 1.00},
            "total_losses_w": {"L0": 0.85, "L1": 0.90, "L2": 0.95, "L3": 0.98, "L4": 1.00},
        }
        self.calibration_data: List[Dict[str, Any]] = []

    def calibrate(self, metric: str, value: float, fidelity: str) -> Dict[str, Any]:
        """Calibrate a low-fidelity result to high-fidelity equivalent.

        Args:
            metric: Metric name (tavg_nm, efficiency_pct, etc.)
            value: Raw value from simulation
            fidelity: Fidelity level (L0-L4)

        Returns:
            Dict with calibrated value, correction factor, and uncertainty
        """
        factors = self.calibration_factors.get(metric, {})
        factor = factors.get(fidelity, 1.0)
        calibrated_value = value * factor

        # Uncertainty based on fidelity level bias
        fidelity_info = FIDELITY_LEVELS.get(fidelity, {"bias_pct": 10.0})
        uncertainty_pct = fidelity_info["bias_pct"]
        uncertainty = abs(calibrated_value) * uncertainty_pct / 100.0

        return {
            "metric": metric,
            "raw_value": value,
            "fidelity": fidelity,
            "calibrated_value": round(calibrated_value, 4),
            "correction_factor": factor,
            "uncertainty_pct": uncertainty_pct,
            "uncertainty_abs": round(uncertainty, 4),
            "calibrated": fidelity != "L4",
        }

    def update_calibration(self, low_fidelity_results: Dict[str, float],
                            high_fidelity_results: Dict[str, float],
                            low_fidelity: str, high_fidelity: str = "L4") -> None:
        """Update calibration factors using paired low-high fidelity results.

        Args:
            low_fidelity_results: Results from low-fidelity simulation
            high_fidelity_results: Results from high-fidelity simulation
            low_fidelity: Low fidelity level
            high_fidelity: High fidelity level (reference)
        """
        for metric, low_value in low_fidelity_results.items():
            if metric in high_fidelity_results and high_fidelity_results[metric] != 0:
                factor = high_fidelity_results[metric] / low_value
                if metric not in self.calibration_factors:
                    self.calibration_factors[metric] = {}
                # Exponential moving average update
                old_factor = self.calibration_factors[metric].get(low_fidelity, 1.0)
                new_factor = 0.7 * old_factor + 0.3 * factor
                self.calibration_factors[metric][low_fidelity] = round(new_factor, 4)

        self.calibration_data.append({
            "timestamp": datetime.now().isoformat(),
            "low_fidelity": low_fidelity,
            "high_fidelity": high_fidelity,
            "metrics": list(low_fidelity_results.keys()),
        })


class ConfidenceGrader:
    """Confidence grade assessor (A-D per third-party review)."""

    @staticmethod
    def grade(
        fidelity: str,
        n_samples: int,
        n_dimensions: int,
        convergence_passed: int,
        total_criteria: int = 6,
        anomalies: int = 0,
    ) -> Dict[str, Any]:
        """Assign confidence grade A-D based on multiple factors.

        Args:
            fidelity: Simulation fidelity level (L0-L4)
            n_samples: Number of simulation samples
            n_dimensions: Number of scan dimensions
            convergence_passed: Number of convergence criteria passed
            total_criteria: Total number of convergence criteria
            anomalies: Number of detected anomalies

        Returns:
            Dict with grade, score, and reasoning
        """
        score = 0.0
        reasons = []

        # Fidelity contribution (40%)
        fidelity_scores = {"L4": 40, "L3": 32, "L2": 24, "L1": 12, "L0": 4}
        fidelity_score = fidelity_scores.get(fidelity, 0)
        score += fidelity_score
        reasons.append(f"fidelity={fidelity} ({fidelity_score}/40)")

        # Sample density contribution (25%)
        if n_dimensions > 0:
            density = n_samples / n_dimensions
            density_score = min(25, density * 2.5)
            score += density_score
            reasons.append(f"sample_density={density:.1f} pts/dim ({density_score:.1f}/25)")

        # Convergence contribution (25%)
        conv_score = (convergence_passed / total_criteria) * 25 if total_criteria > 0 else 0
        score += conv_score
        reasons.append(f"convergence={convergence_passed}/{total_criteria} ({conv_score:.1f}/25)")

        # Anomaly penalty (10%)
        anomaly_penalty = min(10, anomalies * 3)
        score -= anomaly_penalty
        if anomalies > 0:
            reasons.append(f"anomalies={anomalies} (-{anomaly_penalty})")

        # Assign grade
        if score >= 85:
            grade = "A"
        elif score >= 70:
            grade = "B"
        elif score >= 50:
            grade = "C"
        else:
            grade = "D"

        return {
            "grade": grade,
            "score": round(score, 1),
            "max_score": 100,
            "reasons": reasons,
            "fidelity": fidelity,
            "n_samples": n_samples,
            "n_dimensions": n_dimensions,
            "convergence_passed": convergence_passed,
            "anomalies": anomalies,
        }


class AIResultAnalyst:
    """AI-powered simulation result analyst."""

    def __init__(self):
        self.ai_client = get_kimi_client()
        self.calibrator = MultiFidelityCalibrator()
        self.convergence = ConvergenceChecker()
        self.grader = ConfidenceGrader()
        self._prompt_template = None

    def _load_prompt(self) -> str:
        if self._prompt_template is None:
            prompt_path = PROMPTS_DIR / "result_analysis" / "analyze.txt"
            if prompt_path.exists():
                with open(prompt_path, "r", encoding="utf-8") as f:
                    self._prompt_template = f.read()
            else:
                self._prompt_template = "你是电机仿真结果分析专家。分析提供的仿真数据，输出JSON格式的分析报告。"
        return self._prompt_template

    def analyze(
        self,
        results: List[Dict[str, Any]],
        targets: Optional[Dict[str, float]] = None,
        fidelity: str = "L3",
        scan_parameters: Optional[List[str]] = None,
    ) -> Dict[str, Any]:
        """Analyze simulation results with AI and quantitative methods.

        Args:
            results: List of simulation result dicts
            targets: Target values for key metrics
            fidelity: Simulation fidelity level
            scan_parameters: List of scanned parameter names

        Returns:
            Comprehensive analysis report
        """
        if not results:
            return {"error": "No results to analyze"}

        # Quantitative analysis
        quantitative = self._quantitative_analysis(results, targets, fidelity, scan_parameters)

        # AI analysis
        ai_analysis = {}
        if self.ai_client.is_configured:
            ai_analysis = self._ai_analysis(results, targets, quantitative)

        # Combine
        return {
            "summary": ai_analysis.get("summary", quantitative.get("summary", "")),
            "quantitative": quantitative,
            "ai_analysis": ai_analysis,
            "confidence": quantitative.get("confidence", {}),
            "convergence": quantitative.get("convergence", {}),
            "calibration": quantitative.get("calibration", {}),
            "analyzed_at": datetime.now().isoformat(),
            "n_results": len(results),
            "fidelity": fidelity,
        }

    def _quantitative_analysis(
        self,
        results: List[Dict[str, Any]],
        targets: Optional[Dict[str, float]],
        fidelity: str,
        scan_parameters: Optional[List[str]],
    ) -> Dict[str, Any]:
        """Perform quantitative analysis without AI."""
        # Extract key metrics
        metrics_list = []
        for r in results:
            metric = {}
            for key in ["tavg_nm", "efficiency_pct", "total_losses_w", "winding_temp_c", "magnet_temp_c"]:
                if key in r:
                    metric[key] = float(r[key])
            if metric:
                metrics_list.append(metric)

        # Calculate statistics
        stats = {}
        for key in set().union(*(m.keys() for m in metrics_list)) if metrics_list else []:
            values = [m[key] for m in metrics_list if key in m]
            if values:
                stats[key] = {
                    "min": round(min(values), 4),
                    "max": round(max(values), 4),
                    "mean": round(sum(values) / len(values), 4),
                    "std": round(math.sqrt(sum((v - sum(values)/len(values))**2 for v in values) / len(values)), 4) if len(values) > 1 else 0,
                    "n": len(values),
                }

        # Target comparison
        target_comparison = {}
        if targets:
            for metric, target in targets.items():
                if metric in stats:
                    actual = stats[metric]["max"] if "torque" in metric or "efficiency" in metric else stats[metric]["min"]
                    target_comparison[metric] = {
                        "target": target,
                        "best_achieved": actual,
                        "margin": round(actual - target, 4),
                        "status": "pass" if (actual >= target if "torque" in metric or "efficiency" in metric else actual <= target) else "fail",
                    }

        # Calibration
        calibration = {}
        for metric in ["tavg_nm", "efficiency_pct", "total_losses_w"]:
            if metric in stats:
                calibration[metric] = self.calibrator.calibrate(metric, stats[metric]["mean"], fidelity)

        # Convergence checks
        convergence = {}
        objective_values = [m.get("tavg_nm", 0) for m in metrics_list if "tavg_nm" in m]
        convergence["objective_stability"] = self.convergence.check_objective_stability(objective_values)

        feasible_count = sum(1 for r in results if r.get("feasible", True))
        convergence["constraint_satisfaction"] = self.convergence.check_constraint_satisfaction(
            feasible_count, len(results)
        )

        n_dims = len(scan_parameters) if scan_parameters else 1
        convergence["sample_density"] = self.convergence.check_sample_density(len(results), n_dims)

        # Physical consistency
        physical_bounds = {
            "efficiency_pct": (0, 100),
            "tavg_nm": (0, 1000),
            "winding_temp_c": (-50, 300),
        }
        best_metrics = {}
        for key in stats:
            best_metrics[key] = stats[key]["max"] if key in ["tavg_nm", "efficiency_pct"] else stats[key]["min"]
        convergence["physical_consistency"] = self.convergence.check_physical_consistency(best_metrics, physical_bounds)

        # Count passed criteria
        passed = sum(1 for v in convergence.values() if v.get("passed", False))
        convergence["summary"] = {"passed": passed, "total": 6, "status": "converged" if passed >= 4 else "insufficient"}

        # Confidence grade
        anomalies = len(convergence["physical_consistency"].get("violations", []))
        confidence = self.grader.grade(
            fidelity=fidelity,
            n_samples=len(results),
            n_dimensions=n_dims,
            convergence_passed=passed,
            anomalies=anomalies,
        )

        # Summary
        best_torque = stats.get("tavg_nm", {}).get("max", "N/A")
        best_efficiency = stats.get("efficiency_pct", {}).get("max", "N/A")
        summary = f"Analyzed {len(results)} points. Best torque={best_torque}Nm, best efficiency={best_efficiency}%. Confidence: {confidence['grade']}."

        return {
            "summary": summary,
            "statistics": stats,
            "target_comparison": target_comparison,
            "calibration": calibration,
            "convergence": convergence,
            "confidence": confidence,
        }

    def _ai_analysis(
        self,
        results: List[Dict[str, Any]],
        targets: Optional[Dict[str, float]],
        quantitative: Dict[str, Any],
    ) -> Dict[str, Any]:
        """Call AI for natural language analysis."""
        # Prepare condensed data for AI
        condensed = {
            "n_points": len(results),
            "statistics": quantitative.get("statistics", {}),
            "targets": targets or {},
            "best_results": [],
        }

        # Include best 5 results
        sorted_results = sorted(
            results,
            key=lambda r: r.get("tavg_nm", 0),
            reverse=True,
        )[:5]
        condensed["best_results"] = sorted_results

        user_message = f"分析以下轴向磁通电机仿真结果：\n{json.dumps(condensed, ensure_ascii=False, indent=2)}"

        try:
            result = self.ai_client.chat_json(
                messages=[{"role": "user", "content": user_message}],
                system_prompt=self._load_prompt(),
                max_tokens=3000,
            )
            parsed = result.get("parsed_json", {})
            if not parsed:
                raw = result.get("raw_content", result.get("content", ""))
                # Try to extract JSON
                try:
                    start = raw.find("{")
                    end = raw.rfind("}") + 1
                    if start >= 0 and end > start:
                        parsed = json.loads(raw[start:end])
                except (json.JSONDecodeError, Exception):
                    parsed = {"summary": raw[:500]}
            return parsed
        except Exception as e:
            return {"error": f"AI analysis failed: {str(e)}", "summary": quantitative.get("summary", "")}


# Global singletons
_analyst: Optional[AIResultAnalyst] = None


def get_result_analyst() -> AIResultAnalyst:
    """Get or create global AIResultAnalyst singleton."""
    global _analyst
    if _analyst is None:
        _analyst = AIResultAnalyst()
    return _analyst