""" 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}