#!/usr/bin/env python3 """ GENESIS STATUS DASHBOARD ======================== Displays real-time health status of all Genesis components. Usage: python status_dashboard.py # Show status python status_dashboard.py --json # Output as JSON python status_dashboard.py --watch # Continuous monitoring """ import json import os import sys from datetime import datetime from pathlib import Path from typing import Dict, Any, List # Genesis root directory GENESIS_ROOT = Path(__file__).parent.parent class StatusDashboard: """Genesis system health dashboard.""" def __init__(self): self.status = {} self.checks = [ ("Core Imports", self._check_core_imports), ("Hyperdrive Controller", self._check_hyperdrive), ("Gemini Executor", self._check_gemini), ("Task Queue", self._check_tasks), ("Memory System", self._check_memory), ("Configuration", self._check_config), ("AIVA Connection", self._check_aiva), ] def _check_core_imports(self) -> Dict[str, Any]: """Check if core modules import successfully.""" modules = { "genesis_kernel": False, "genesis_heartbeat": False, "continuous_evolution": False, "self_learning": False, } for module in modules: try: __import__(module) modules[module] = True except ImportError: pass success = sum(modules.values()) total = len(modules) return { "status": "healthy" if success == total else "degraded" if success > 0 else "failed", "healthy_count": success, "total_count": total, "modules": modules } def _check_hyperdrive(self) -> Dict[str, Any]: """Check hyperdrive controller availability.""" try: from hyperdrive_controller import HyperdriveController, AgentSelector selector = AgentSelector() test = selector.classify_task("test task") return { "status": "healthy", "controller_available": True, "selector_available": True, "test_classification": test.recommended_agent.value } except Exception as e: return { "status": "failed", "error": str(e)[:100] } def _check_gemini(self) -> Dict[str, Any]: """Check Gemini executor availability.""" try: os.environ['GEMINI_API_KEY'] = 'AIzaSyALfbAdHfJ6aRnqNyiTRmKmGVoena1JsdU' from gemini_executor import GeminiExecutor executor = GeminiExecutor() # Get budget status budget = executor.get_budget_status() return { "status": "healthy" if executor.api_key else "degraded", "api_key_configured": bool(executor.api_key), "budget_remaining": budget.get("remaining", 0), "iterations_used": budget.get("iterations", 0) } except Exception as e: return { "status": "failed", "error": str(e)[:100] } def _check_tasks(self) -> Dict[str, Any]: """Check task queue status.""" tasks_path = GENESIS_ROOT / "loop" / "tasks.json" if not tasks_path.exists(): return { "status": "missing", "error": "tasks.json not found" } try: with open(tasks_path) as f: data = json.load(f) stories = data.get("stories", []) completed = len([s for s in stories if s.get("passes")]) pending = len([s for s in stories if not s.get("passes")]) return { "status": "healthy" if pending > 0 else "complete", "total_tasks": len(stories), "completed": completed, "pending": pending, "project": data.get("project", "Unknown"), "phase": data.get("phase", 1) } except Exception as e: return { "status": "error", "error": str(e)[:100] } def _check_memory(self) -> Dict[str, Any]: """Check memory system components.""" memory_files = { "episodic": GENESIS_ROOT / "data" / "episodic_memory.db", "axioms": GENESIS_ROOT / "data" / "axioms.json", "heartbeat_state": GENESIS_ROOT / "data" / "heartbeat_state.json", "kernel_state": GENESIS_ROOT / "data" / "kernel_state.json", } available = {} for name, path in memory_files.items(): available[name] = path.exists() healthy = sum(available.values()) return { "status": "healthy" if healthy >= 2 else "degraded" if healthy > 0 else "missing", "files_available": available, "count": healthy } def _check_config(self) -> Dict[str, Any]: """Check configuration files.""" config_files = { "genesis_config": GENESIS_ROOT / "config" / "genesis_config.json", "unceasing_motion": GENESIS_ROOT / "config" / "unceasing_motion.json", "swarm_config": GENESIS_ROOT / "config" / "swarm_config.json", } available = {} for name, path in config_files.items(): available[name] = path.exists() return { "status": "healthy" if all(available.values()) else "degraded", "files": available } def _check_aiva(self) -> Dict[str, Any]: """Check AIVA connection (Elestio).""" # Check if AIVA endpoint is configured aiva_host = "152.53.201.152" aiva_port = 23405 try: import socket sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(2) result = sock.connect_ex((aiva_host, aiva_port)) sock.close() if result == 0: return { "status": "healthy", "host": aiva_host, "port": aiva_port, "reachable": True } else: return { "status": "unreachable", "host": aiva_host, "port": aiva_port, "reachable": False } except Exception as e: return { "status": "error", "error": str(e)[:100] } def run_checks(self) -> Dict[str, Any]: """Run all health checks.""" results = { "timestamp": datetime.now().isoformat(), "genesis_root": str(GENESIS_ROOT), "checks": {} } overall_health = [] for name, check_func in self.checks: try: result = check_func() results["checks"][name] = result overall_health.append(result.get("status", "unknown")) except Exception as e: results["checks"][name] = { "status": "error", "error": str(e)[:100] } overall_health.append("error") # Calculate overall status if all(s == "healthy" for s in overall_health): results["overall_status"] = "HEALTHY" elif any(s in ["failed", "error"] for s in overall_health): results["overall_status"] = "DEGRADED" else: results["overall_status"] = "PARTIAL" return results def display(self, as_json: bool = False): """Display dashboard status.""" results = self.run_checks() if as_json: print(json.dumps(results, indent=2)) return # Pretty print print("\n" + "=" * 60) print("GENESIS STATUS DASHBOARD") print("=" * 60) print(f"Timestamp: {results['timestamp']}") print(f"Overall Status: {results['overall_status']}") print("-" * 60) for name, data in results["checks"].items(): status = data.get("status", "unknown") icon = {"healthy": "[OK]", "degraded": "[!!]", "failed": "[XX]", "complete": "[OK]"}.get(status, "[??]") print(f"\n{icon} {name}") for key, value in data.items(): if key != "status": print(f" {key}: {value}") print("\n" + "=" * 60) def main(): import argparse parser = argparse.ArgumentParser(description="Genesis Status Dashboard") parser.add_argument("--json", action="store_true", help="Output as JSON") parser.add_argument("--watch", action="store_true", help="Continuous monitoring") args = parser.parse_args() dashboard = StatusDashboard() if args.watch: import time while True: os.system('clear' if os.name == 'posix' else 'cls') dashboard.display(as_json=args.json) time.sleep(5) else: dashboard.display(as_json=args.json) if __name__ == "__main__": main()