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

"""Experience Library AI Enhancer (P3-M5).

Uses AI to extract design knowledge, rules, and insights from
historical simulation results, enhancing the experience library.
"""
import json
from pathlib import Path
from typing import Dict, List, Optional, Any
from datetime import datetime

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


class ExperienceEnhancer:
    """AI-powered experience library enhancer.

    Extracts:
    - Design rules and heuristics from successful cases
    - Failure patterns and constraint boundaries
    - Parameter sensitivity rankings
    - Optimal design regions
    - Comparative insights across topologies
    """

    def __init__(self):
        self.ai_client = get_kimi_client()
        self._prompt_template = None

    def _load_prompt(self) -> str:
        if self._prompt_template is None:
            prompt_path = PROMPTS_DIR / "experience" / "extract.txt"
            if prompt_path.exists():
                with open(prompt_path, "r", encoding="utf-8") as f:
                    self._prompt_template = f.read()
            else:
                self._prompt_template = self._default_prompt()
        return self._prompt_template

    def _default_prompt(self) -> str:
        return """\u4f60\u662f\u8f74\u5411\u78c1\u901a\u7535\u673a\u8bbe\u8ba1\u77e5\u8bc6\u63d0\u53d6\u4e13\u5bb6\u3002\u4ece\u4eff\u771f\u7ed3\u679c\u6570\u636e\u4e2d\u63d0\u53d6\u8bbe\u8ba1\u89c4\u5219\u3001\u5931\u8d25\u6a21\u5f0f\u3001\u53c2\u6570\u654f\u611f\u6027\u548c\u6700\u4f18\u8bbe\u8ba1\u533a\u57df\u3002\u8f93\u51faJSON\u683c\u5f0f\u3002"""

    def extract_insights(
        self,
        results: List[Dict[str, Any]],
        project_context: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        """Extract design insights from a batch of simulation results.

        Args:
            results: List of simulation result dicts
            project_context: Optional project context

        Returns:
            Structured insights including rules, patterns, and recommendations
        """
        if not results:
            return {"error": "No results to analyze"}

        # Prepare condensed data
        condensed = self._condense_results(results)
        if project_context:
            condensed["project_context"] = project_context

        user_message = f"\u4ece\u4ee5\u4e0b\u8f74\u5411\u78c1\u901a\u7535\u673a\u4eff\u771f\u7ed3\u679c\u4e2d\u63d0\u53d6\u8bbe\u8ba1\u77e5\u8bc6\uff1a\n{json.dumps(condensed, ensure_ascii=False, indent=2)}"

        if not self.ai_client.is_configured:
            return self._fallback_insights(results)

        try:
            result = self.ai_client.chat_json(
                messages=[{"role": "user", "content": user_message}],
                system_prompt=self._load_prompt(),
                max_tokens=KIMI_MAX_TOKENS,
            )
            insights = result.get("parsed_json", {})
            if not insights:
                raw = result.get("raw_content", result.get("content", ""))
                try:
                    start = raw.find("{")
                    end = raw.rfind("}") + 1
                    if start >= 0 and end > start:
                        insights = json.loads(raw[start:end])
                except (json.JSONDecodeError, Exception):
                    insights = {"summary": raw[:500]}
            insights["extracted_at"] = datetime.now().isoformat()
            insights["n_results_analyzed"] = len(results)
            insights["usage"] = result.get("usage", {})
            return insights
        except Exception as e:
            return {"error": f"AI insight extraction failed: {str(e)}", **self._fallback_insights(results)}

    def _condense_results(self, results: List[Dict[str, Any]]) -> Dict[str, Any]:
        """Condense results for AI processing (avoid token overflow)."""
        # Extract key metrics. Input params may arrive under Motor-CAD names
        # (Airgap, RMSCurrent); translate them to the BC-style keys below via
        # the shared single-source map so sensitivity analysis sees inputs.
        from src.afmcore.l0.prescreening import MOTORCAD_TO_L0
        keys = ["tavg_nm", "efficiency_pct", "total_losses_w", "winding_temp_c",
                "airgap_mm", "magnet_thickness_mm", "current_a", "speed_rpm",
                "outer_diameter_mm", "inner_diameter_mm", "feasible"]
        metrics = []
        for r in results:
            r_norm = dict(r)
            for mc_name, l0_name in MOTORCAD_TO_L0.items():
                if mc_name in r_norm and l0_name not in r_norm:
                    r_norm[l0_name] = r_norm[mc_name]
            m = {}
            for key in keys:
                if key in r_norm:
                    m[key] = r_norm[key]
            if m:
                metrics.append(m)

        # Sort by torque and take top/bottom 10 + random 10
        sorted_by_torque = sorted(metrics, key=lambda x: x.get("tavg_nm", 0), reverse=True)
        top_10 = sorted_by_torque[:10]
        bottom_10 = sorted_by_torque[-10:] if len(sorted_by_torque) > 20 else []

        # Calculate basic statistics
        all_torque = [m.get("tavg_nm", 0) for m in metrics if "tavg_nm" in m]
        all_efficiency = [m.get("efficiency_pct", 0) for m in metrics if "efficiency_pct" in m]

        stats = {}
        if all_torque:
            stats["torque"] = {
                "min": min(all_torque), "max": max(all_torque),
                "mean": sum(all_torque) / len(all_torque),
            }
        if all_efficiency:
            stats["efficiency"] = {
                "min": min(all_efficiency), "max": max(all_efficiency),
                "mean": sum(all_efficiency) / len(all_efficiency),
            }

        return {
            "total_points": len(results),
            "statistics": stats,
            "top_performers": top_10,
            "bottom_performers": bottom_10,
        }

    def _fallback_insights(self, results: List[Dict[str, Any]]) -> Dict[str, Any]:
        """Generate basic insights without AI."""
        feasible = [r for r in results if r.get("feasible", True)]
        infeasible = [r for r in results if not r.get("feasible", True)]

        insights = {
            "summary": f"Analyzed {len(results)} points ({len(feasible)} feasible, {len(infeasible)} infeasible).",
            "feasibility_rate": len(feasible) / len(results) if results else 0,
            "design_rules": [],
            "failure_patterns": [],
            "extracted_at": datetime.now().isoformat(),
            "n_results_analyzed": len(results),
            "ai_used": False,
        }

        # Simple correlation analysis
        if len(feasible) >= 5:
            # Find parameters that correlate with high torque
            for param in ["airgap_mm", "magnet_thickness_mm", "current_a"]:
                values = [(r.get(param, 0), r.get("tavg_nm", 0)) for r in feasible if param in r]
                if len(values) >= 5:
                    values.sort(key=lambda x: x[1], reverse=True)
                    top_params = [v[0] for v in values[:len(values)//3]]
                    avg_top = sum(top_params) / len(top_params) if top_params else 0
                    all_avg = sum(v[0] for v in values) / len(values)
                    if abs(avg_top - all_avg) > 0.1 * abs(all_avg) if all_avg else False:
                        direction = "higher" if avg_top > all_avg else "lower"
                        insights["design_rules"].append(
                            f"High torque designs tend to have {direction} {param} (avg top={avg_top:.2f} vs all={all_avg:.2f})"
                        )

        return insights

    def generate_experience_entry(
        self,
        insights: Dict[str, Any],
        project_name: str,
        topology: str,
    ) -> Dict[str, Any]:
        """Generate a structured experience library entry from insights.

        Args:
            insights: Extracted insights
            project_name: Project name
            topology: Motor topology

        Returns:
            Structured experience entry ready for database storage
        """
        return {
            "title": f"{project_name} - {topology} Design Insights",
            "topology": topology,
            "project": project_name,
            "summary": insights.get("summary", ""),
            "design_rules": insights.get("design_rules", []),
            "failure_patterns": insights.get("failure_patterns", []),
            "parameter_sensitivity": insights.get("parameter_sensitivity", {}),
            "optimal_regions": insights.get("optimal_regions", []),
            "confidence": insights.get("confidence_grade", "C"),
            "source": "ai_extracted",
            "created_at": datetime.now().isoformat(),
            "tags": [topology, "ai-insights", project_name],
        }


# Global singleton
_enhancer: Optional[ExperienceEnhancer] = None


def get_experience_enhancer() -> ExperienceEnhancer:
    """Get or create global ExperienceEnhancer singleton."""
    global _enhancer
    if _enhancer is None:
        _enhancer = ExperienceEnhancer()
    return _enhancer