kpi_tracker.py

"""

KPI Tracking - Track consciousness metrics over time

"""

from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any
from collections import deque
from dataclasses import dataclass
import json
import logging

logger = logging.getLogger(__name__)

@dataclass
class KPISnapshot:
    """Snapshot of consciousness KPIs at a point in time"""
    timestamp: datetime
    
    # Memory metrics
    total_memories: int
    core_memories: int
    long_term_memories: int
    short_term_memories: int
    ephemeral_memories: int
    memory_promotion_rate: float
    
    # Interaction metrics
    interactions_count: int
    avg_confidence: float
    
    # Autonomy metrics
    autonomous_actions_today: int
    knowledge_gaps_total: int
    knowledge_gaps_filled_today: int
    proactive_contacts_today: int
    
    # Cognitive metrics
    dreams_completed: int
    reflections_completed: int
    goals_active: int
    goals_completed: int
    
    # Emotional metrics
    current_mood: str
    mood_changes_today: int
    curiosity_level: float
    enthusiasm_level: float

class KPITracker:
    """Track consciousness KPIs over time"""
    
    def __init__(self, history_hours: int = 72):
        self.history_hours = history_hours
        self.snapshots: deque = deque(maxlen=1000)
        
        # Daily counters
        self.autonomous_actions_today = 0
        self.knowledge_gaps_filled_today = 0
        self.proactive_contacts_today = 0
        self.mood_changes_today = 0
        self.reflections_today = 0
        
        # Cumulative counters
        self.total_autonomous_actions = 0
        self.total_knowledge_gaps_filled = 0
        self.total_proactive_contacts = 0
        self.total_mood_changes = 0
        
        self.last_reset = datetime.now()
        self.last_mood = "neutral"
        
        logger.info("[KPI] Tracker initialized")
    
    def capture_snapshot(self, consciousness_loop) -> KPISnapshot:
        """Capture current KPIs from consciousness loop"""
        
        # Daily reset check
        if (datetime.now() - self.last_reset).days >= 1:
            self._reset_daily_counters()
        
        # Check for mood change
        current_mood = consciousness_loop.emotional_state.current_mood
        if current_mood != self.last_mood:
            self.increment_mood_change()
            self.last_mood = current_mood
        
        # Get memory summary
        mem_summary = consciousness_loop.memory.get_summary()
        
        # Calculate promotion rate
        total_mem = mem_summary.get('total', 0)
        promoted = (mem_summary.get('short_term', 0) + 
                   mem_summary.get('long_term', 0) + 
                   mem_summary.get('core', 0))
        promotion_rate = promoted / total_mem if total_mem > 0 else 0.0
        
        # Get active/completed goals
        active_goals = [g for g in consciousness_loop.goal_system.goals if not g.completed]
        completed_goals = [g for g in consciousness_loop.goal_system.goals if g.completed]
        
        # Get knowledge gaps
        unfilled_gaps = [g for g in consciousness_loop.meta_cognition.knowledge_gaps if not g.filled]
        
        snapshot = KPISnapshot(
            timestamp=datetime.now(),
            total_memories=mem_summary.get('total', 0),
            core_memories=mem_summary.get('core', 0),
            long_term_memories=mem_summary.get('long_term', 0),
            short_term_memories=mem_summary.get('short_term', 0),
            ephemeral_memories=mem_summary.get('ephemeral', 0),
            memory_promotion_rate=promotion_rate,
            interactions_count=consciousness_loop.interaction_count,
            avg_confidence=consciousness_loop.meta_cognition.get_average_confidence(),
            autonomous_actions_today=self.autonomous_actions_today,
            knowledge_gaps_total=len(unfilled_gaps),
            knowledge_gaps_filled_today=self.knowledge_gaps_filled_today,
            proactive_contacts_today=self.proactive_contacts_today,
            dreams_completed=len(consciousness_loop.dreams),
            reflections_completed=self.reflections_today,
            goals_active=len(active_goals),
            goals_completed=len(completed_goals),
            current_mood=current_mood,
            mood_changes_today=self.mood_changes_today,
            curiosity_level=consciousness_loop.emotional_state.personality_traits.get('curiosity', 0.5),
            enthusiasm_level=consciousness_loop.emotional_state.personality_traits.get('enthusiasm', 0.5)
        )
        
        self.snapshots.append(snapshot)
        self._cleanup_old_snapshots()
        
        return snapshot
    
    def _reset_daily_counters(self):
        """Reset daily counters at midnight"""
        logger.info(f"[KPI] Daily reset - Actions: {self.autonomous_actions_today}, "
                   f"Gaps filled: {self.knowledge_gaps_filled_today}, "
                   f"Proactive: {self.proactive_contacts_today}")
        
        self.autonomous_actions_today = 0
        self.knowledge_gaps_filled_today = 0
        self.proactive_contacts_today = 0
        self.mood_changes_today = 0
        self.reflections_today = 0
        self.last_reset = datetime.now()
    
    def _cleanup_old_snapshots(self):
        """Remove snapshots older than history_hours"""
        if not self.snapshots:
            return
        
        cutoff = datetime.now() - timedelta(hours=self.history_hours)
        # Deque doesn't support list comprehension, so convert
        temp_list = [s for s in self.snapshots if s.timestamp > cutoff]
        self.snapshots.clear()
        self.snapshots.extend(temp_list)
    
    # Increment methods
    def increment_autonomous_action(self):
        self.autonomous_actions_today += 1
        self.total_autonomous_actions += 1
        logger.debug(f"[KPI] Autonomous action #{self.total_autonomous_actions}")
    
    def increment_gap_filled(self):
        self.knowledge_gaps_filled_today += 1
        self.total_knowledge_gaps_filled += 1
        logger.debug(f"[KPI] Gap filled #{self.total_knowledge_gaps_filled}")
    
    def increment_proactive_contact(self):
        self.proactive_contacts_today += 1
        self.total_proactive_contacts += 1
        logger.info(f"[KPI] Proactive contact #{self.total_proactive_contacts}")
    
    def increment_mood_change(self):
        self.mood_changes_today += 1
        self.total_mood_changes += 1
    
    def increment_reflection(self):
        self.reflections_today += 1
    
    # Analysis methods
    def get_trend(self, metric: str, hours: int = 24) -> List[float]:
        """Get trend for a metric over time"""
        cutoff = datetime.now() - timedelta(hours=hours)
        recent = [s for s in self.snapshots if s.timestamp > cutoff]
        
        if not recent:
            return []
        
        metric_map = {
            "confidence": lambda s: s.avg_confidence,
            "memories": lambda s: s.total_memories,
            "core_memories": lambda s: s.core_memories,
            "autonomous": lambda s: s.autonomous_actions_today,
            "curiosity": lambda s: s.curiosity_level,
            "enthusiasm": lambda s: s.enthusiasm_level,
            "promotion_rate": lambda s: s.memory_promotion_rate
        }
        
        if metric in metric_map:
            return [metric_map[metric](s) for s in recent]
        
        return []
    
    def get_growth_rate(self, metric: str, hours: int = 24) -> float:
        """Calculate growth rate for a metric"""
        trend = self.get_trend(metric, hours)
        
        if len(trend) < 2:
            return 0.0
        
        start = trend[0]
        end = trend[-1]
        
        if start == 0:
            return 0.0
        
        return ((end - start) / start) * 100
    
    def get_summary(self) -> Dict[str, Any]:
        """Get summary of current KPIs"""
        if not self.snapshots:
            return {"error": "No snapshots captured yet"}
        
        latest = self.snapshots[-1]
        
        # Calculate trends (last 24 hours)
        confidence_trend = self.get_trend("confidence", 24)
        memory_trend = self.get_trend("memories", 24)
        
        summary = {
            "timestamp": latest.timestamp.isoformat(),
            "memory": {
                "total": latest.total_memories,
                "core": latest.core_memories,
                "long_term": latest.long_term_memories,
                "short_term": latest.short_term_memories,
                "ephemeral": latest.ephemeral_memories,
                "promotion_rate": round(latest.memory_promotion_rate, 2),
                "growth_24h": round(self.get_growth_rate("memories", 24), 1)
            },
            "interactions": {
                "total": latest.interactions_count,
                "avg_confidence": round(latest.avg_confidence, 2),
                "confidence_trend": "↑" if len(confidence_trend) > 1 and confidence_trend[-1] > confidence_trend[0] else "↓"
            },
            "autonomy": {
                "actions_today": latest.autonomous_actions_today,
                "total_actions": self.total_autonomous_actions,
                "gaps_total": latest.knowledge_gaps_total,
                "gaps_filled_today": latest.knowledge_gaps_filled_today,
                "gaps_filled_total": self.total_knowledge_gaps_filled,
                "proactive_today": latest.proactive_contacts_today,
                "proactive_total": self.total_proactive_contacts
            },
            "cognitive": {
                "dreams": latest.dreams_completed,
                "reflections_today": latest.reflections_completed,
                "goals_active": latest.goals_active,
                "goals_completed": latest.goals_completed
            },
            "emotional": {
                "mood": latest.current_mood,
                "mood_changes_today": latest.mood_changes_today,
                "curiosity": round(latest.curiosity_level * 100, 1),
                "enthusiasm": round(latest.enthusiasm_level * 100, 1)
            }
        }
        
        return summary
    
    def get_detailed_report(self) -> str:
        """Get human-readable detailed report"""
        summary = self.get_summary()
        
        if "error" in summary:
            return summary["error"]
        
        report = f"""

╔══════════════════════════════════════════════════════════════╗

║          CONSCIOUSNESS LOOP - KPI REPORT                     ║

╠══════════════════════════════════════════════════════════════╣

║  Time: {summary['timestamp']}                   

╠══════════════════════════════════════════════════════════════╣

║  MEMORY SYSTEM                                               ║

║    Total Memories: {summary['memory']['total']}                                          ║

║      ├─ Core: {summary['memory']['core']}                                              ║

║      ├─ Long-term: {summary['memory']['long_term']}                                       ║

║      ├─ Short-term: {summary['memory']['short_term']}                                      ║

║      └─ Ephemeral: {summary['memory']['ephemeral']}                                       ║

║    Promotion Rate: {summary['memory']['promotion_rate']:.0%}                                    ║

║    24h Growth: {summary['memory']['growth_24h']:+.1f}%                                    ║

╠══════════════════════════════════════════════════════════════╣

║  INTERACTIONS                                                ║

║    Total: {summary['interactions']['total']}                                            ║

║    Avg Confidence: {summary['interactions']['avg_confidence']:.0%} {summary['interactions']['confidence_trend']}                           ║

╠══════════════════════════════════════════════════════════════╣

║  AUTONOMY                                                    ║

║    Actions Today: {summary['autonomy']['actions_today']} (Total: {summary['autonomy']['total_actions']})                   ║

║    Knowledge Gaps: {summary['autonomy']['gaps_total']} open                                  ║

║    Gaps Filled Today: {summary['autonomy']['gaps_filled_today']} (Total: {summary['autonomy']['gaps_filled_total']})                ║

║    Proactive Today: {summary['autonomy']['proactive_today']} (Total: {summary['autonomy']['proactive_total']})                  ║

╠══════════════════════════════════════════════════════════════╣

║  COGNITIVE                                                   ║

║    Dreams: {summary['cognitive']['dreams']}                                              ║

║    Reflections Today: {summary['cognitive']['reflections_today']}                                   ║

║    Goals: {summary['cognitive']['goals_active']} active, {summary['cognitive']['goals_completed']} completed                         ║

╠══════════════════════════════════════════════════════════════╣

║  EMOTIONAL                                                   ║

║    Mood: {summary['emotional']['mood'].upper()}                                          ║

║    Mood Changes Today: {summary['emotional']['mood_changes_today']}                                  ║

║    Curiosity: {summary['emotional']['curiosity']:.1f}%                                      ║

║    Enthusiasm: {summary['emotional']['enthusiasm']:.1f}%                                    ║

╚══════════════════════════════════════════════════════════════╝

"""
        return report
    
    def export_to_json(self, filepath: str):
        """Export all snapshots to JSON"""
        data = [
            {
                "timestamp": s.timestamp.isoformat(),
                "total_memories": s.total_memories,
                "core_memories": s.core_memories,
                "avg_confidence": s.avg_confidence,
                "autonomous_actions": s.autonomous_actions_today,
                "knowledge_gaps": s.knowledge_gaps_total,
                "current_mood": s.current_mood,
                "curiosity": s.curiosity_level,
                "enthusiasm": s.enthusiasm_level
            }
            for s in self.snapshots
        ]
        
        with open(filepath, 'w') as f:
            json.dump(data, f, indent=2)
        
        logger.info(f"[KPI] Exported {len(data)} snapshots to {filepath}")
    
    def export_summary_to_json(self, filepath: str):
        """Export current summary to JSON"""
        summary = self.get_summary()
        
        with open(filepath, 'w') as f:
            json.dump(summary, f, indent=2)
        
        logger.info(f"[KPI] Exported summary to {filepath}")
    
    def get_timeseries(self, metric: str, hours: int = 24) -> Dict[str, list]:
        """Return time-series data for a given KPI metric over the last N hours."""
        cutoff = datetime.now() - timedelta(hours=hours)
        snapshots = [s for s in self.snapshots if s.timestamp > cutoff]
        timestamps = [s.timestamp.isoformat() for s in snapshots]
        metric_map = {
            "confidence": lambda s: s.avg_confidence,
            "memories": lambda s: s.total_memories,
            "core_memories": lambda s: s.core_memories,
            "autonomous": lambda s: s.autonomous_actions_today,
            "curiosity": lambda s: s.curiosity_level,
            "enthusiasm": lambda s: s.enthusiasm_level,
            "promotion_rate": lambda s: s.memory_promotion_rate,
            "reflections": lambda s: s.reflections_completed,
            "dreams": lambda s: s.dreams_completed,
            "proactive": lambda s: s.proactive_contacts_today,
            "gaps_filled": lambda s: s.knowledge_gaps_filled_today,
        }
        if metric in metric_map:
            values = [metric_map[metric](s) for s in snapshots]
        else:
            values = []
        return {"timestamps": timestamps, "values": values}