Spaces:
Running
Running
| """ | |
| 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__) | |
| 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} |