import click from rich.console import Console from rich.table import Table from rich.progress import Progress import requests from datetime import datetime, timezone # Define constants BACKEND_BASE_URL = "http://localhost:8000" THINK_K8S_URL = f"{BACKEND_BASE_URL}/think/k8s" METAL_NODES_URL = f"{BACKEND_BASE_URL}/metal/data" VM_URL = f"{BACKEND_BASE_URL}/vm/data" K8S_URL = f"{BACKEND_BASE_URL}/k8s/data" console = Console() # Helper functions for formatting def calculate_time_on_duty_hours(hours): if hours < 1: minutes = round(hours * 60) return f"{minutes} minutes" if minutes > 1 else "less than a minute" elif hours < 24: return f"{round(hours)} hours" if hours > 1 else "1 hour" else: days = hours // 24 remaining_hours = hours % 24 if remaining_hours: return f"{int(days)} days {int(remaining_hours)} hours" return f"{int(days)} days" # Fetch and display metal nodes def display_metal_nodes(): table = Table(title="🖥️ Metal Nodes", style="bold green") table.add_column("ID", justify="right", style="cyan") table.add_column("Name", style="magenta") table.add_column("Location", style="white") table.add_column("Vendor", style="green") table.add_column("CPU", justify="right", style="yellow") table.add_column("Memory (GB)", justify="right", style="cyan") table.add_column("Storage", style="magenta") table.add_column("Time on Duty", justify="right", style="magenta") try: response = requests.get(METAL_NODES_URL) if response.status_code == 200: metal_nodes = response.json().get("metal_nodes", []) for node in metal_nodes: time_on_duty = calculate_time_on_duty_hours(node[7]) table.add_row( f"{node[0]}", node[1], node[2], node[3], f"{node[4]}", node[5], node[6], time_on_duty, ) else: console.print(f"[red]Failed to fetch metal nodes: {response.status_code}[/red]") except requests.RequestException as e: console.print(f"[red]Error fetching metal nodes: {e}[/red]") console.print(table) # Fetch and display virtual machines def display_virtual_machines(): table = Table(title="💻 Virtual Machines", style="bold blue") table.add_column("ID", justify="right", style="cyan") table.add_column("Name", style="magenta") table.add_column("Location", style="white") table.add_column("CPU", justify="right", style="yellow") table.add_column("Memory (GB)", justify="right", style="cyan") table.add_column("Storage", style="magenta") table.add_column("Type", style="green") try: response = requests.get(VM_URL) if response.status_code == 200: virtual_machines = response.json().get("virtual_machines", []) for vm in virtual_machines: table.add_row( f"{vm[0]}", vm[1], vm[2], f"{vm[3]}", vm[4], vm[5], vm[6], ) else: console.print(f"[red]Failed to fetch virtual machines: {response.status_code}[/red]") except requests.RequestException as e: console.print(f"[red]Error fetching virtual machines: {e}[/red]") console.print(table) # Fetch and display Kubernetes nodes def display_kubernetes_nodes(): table = Table(title="📦 Kubernetes Nodes", style="bold yellow") table.add_column("Node Name", style="white") table.add_column("CPU", justify="right", style="yellow") table.add_column("Memory (MiB)", justify="right", style="cyan") table.add_column("Storage (GB)", justify="right", style="green") table.add_column("Type", style="blue") table.add_column("Max Pods", justify="right", style="magenta") table.add_column("Time on Duty", justify="right", style="magenta") try: response = requests.get(K8S_URL) if response.status_code == 200: nodes = response.json().get("nodes", []) for node in nodes: table.add_row( node["node_name"], node["cpu"], f"{node['memory']}", f"{node['storage']}", node["instance_type"], node["pods_allocatable"], node.get("time_on_duty", "N/A"), ) else: console.print(f"[red]Failed to fetch Kubernetes nodes: {response.status_code}[/red]") except requests.RequestException as e: console.print(f"[red]Error fetching Kubernetes nodes: {e}[/red]") console.print(table) # Fetch and display AI summary def display_ai_summary(): with Progress() as progress: task = progress.add_task("[cyan]Fetching AI Summary...", total=100) try: for _ in range(10): # Simulate progress progress.update(task, advance=10) import time; time.sleep(0.1) response = requests.get(THINK_K8S_URL) progress.update(task, completed=100) if response.status_code == 200: data = response.json() summary = data.get("summary", "No summary provided.") console.print("\n[bold magenta]AI Summary of Kubernetes Cluster:[/bold magenta]") console.print(f"[green]{summary}[/green]\n") else: console.print(f"[red]Failed to fetch AI summary: {response.status_code}[/red]") except requests.RequestException as e: console.print(f"[red]Error fetching AI summary: {e}[/red]") # Click command for visual dashboard @click.command() @click.option('--summary', is_flag=True, help="Include AI summary in the dashboard.") def visual_dashboard(summary): """ Displays the visual dashboard with Metal Nodes, Virtual Machines, Kubernetes Nodes, and optionally AI Summary. """ console.print("✨ [bold green]Welcome to the Metal Check Dashboard![/bold green] ✨\n") display_metal_nodes() display_virtual_machines() display_kubernetes_nodes() if summary: display_ai_summary()