metalcheck-cli/cli/visual.py

from rich.console import Console
from rich.table import Table
from rich.progress import Progress
import requests
import click
from datetime import datetime, timezone

console = Console()

# Helper function to set backend URLs
def get_backend_urls(base_url):
    return {
        "THINK_K8S_URL": f"{base_url}/think/k8s",
        "METAL_NODES_URL": f"{base_url}/metal/data",
        "VM_URL": f"{base_url}/vm/data",
        "K8S_URL": f"{base_url}/k8s/data",
    }

# 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(metal_nodes_url):
    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(vm_url):
    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(k8s_url):
    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(think_k8s_url):
    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.")
@click.pass_context
def visual_dashboard(ctx, summary):
    """
    Displays the visual dashboard with Metal Nodes, Virtual Machines,
    Kubernetes Nodes, and optionally AI Summary.
    """
    base_url = ctx.obj["BASE_URL"]
    urls = get_backend_urls(base_url)

    console.print("✨ [bold green]Welcome to the Metal Check Dashboard![/bold green] ✨\n")
    display_metal_nodes(urls["METAL_NODES_URL"])
    display_virtual_machines(urls["VM_URL"])
    display_kubernetes_nodes(urls["K8S_URL"])

    if summary:
        display_ai_summary(urls["THINK_K8S_URL"])