goal_formulator.py

"""
Goal Formulator — analyzes capabilities, identifies gaps, defines improvement goals.
"""

import logging
import time
from collections import deque
from typing import Any, Dict, List, Optional

logger = logging.getLogger("nima_unified.training.goal_formulator")


class GoalFormulator:
    """
    Formulates improvement goals by analyzing capabilities, identifying gaps,
    and defining objectives based on system performance and feedback.
    """

    def __init__(self):
        self.performance_history: deque = deque(maxlen=100)
        self.capability_map: Dict[str, float] = {}
        self.gap_analysis: List[Dict[str, Any]] = []
        self.goals: List[Dict[str, Any]] = []
        logger.info("GoalFormulator initialized")

    def analyze_capabilities(self, capabilities: Dict[str, float]) -> Dict[str, Any]:
        """Analyze current capabilities and identify gaps."""
        self.capability_map.update(capabilities)

        gaps = {}
        priorities = []

        for cap, score in capabilities.items():
            if score < 0.5:
                gaps[cap] = {"current": score, "priority": "HIGH"}
                priorities.append((cap, score, "HIGH"))
            elif score < 0.8:
                gaps[cap] = {"current": score, "priority": "MEDIUM"}
                priorities.append((cap, score, "MEDIUM"))
            else:
                gaps[cap] = {"current": score, "priority": "LOW"}

        priorities.sort(key=lambda x: (x[2] != "HIGH", x[2] != "MEDIUM", x[1]))

        analysis = {
            "timestamp": time.time(),
            "total_capabilities": len(capabilities),
            "average_maturity": sum(capabilities.values()) / len(capabilities) if capabilities else 0.0,
            "gaps": gaps,
            "priority_order": [p[0] for p in priorities],
        }

        self.gap_analysis.append(analysis)
        logger.debug(f"Capability analysis: {len(priorities)} gaps identified")
        return analysis

    def formulate_goals(
        self,
        gap_analysis: Dict[str, Any],
        performance_feedback: Optional[Dict[str, Any]] = None,
    ) -> List[Dict[str, Any]]:
        """Formulate improvement goals from gap analysis and feedback."""
        goals = []
        for gap in gap_analysis.get("priority_order", []):
            gap_info = gap_analysis["gaps"][gap]
            current_score = gap_info["current"]
            goal = {
                "capability": gap,
                "current_score": current_score,
                "target_score": min(1.0, current_score + 0.3),
                "priority": gap_info["priority"],
                "formulated_at": time.time(),
                "target_completion": time.time() + (3600 if gap_info["priority"] == "HIGH" else 7200),
            }
            goals.append(goal)

        self.goals.extend(goals)
        logger.info(f"Formulated {len(goals)} improvement goals")
        return goals