|
@@ -0,0 +1,477 @@
|
|
|
|
|
+"""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
|