Dashboard pour monitorer une machine sous Ubuntu

monitor_dashboard.py

#!/bin/python3
# Paquets requis : nvidia-utils, python3, python3-pip, pip3, lm-sensors, nvme-cl, smartmontools, docker.io
# Libs python : 
# pip3 install psutil textual rich

# 'r' pour recharger, 'q' pour quitter
from textual.app import App, ComposeResult
from textual.containers import Horizontal, Vertical
from textual.widgets import Static
from textual.reactive import reactive
from rich.panel import Panel
from rich.text import Text
import psutil
import subprocess
import os
import platform
import socket


def get_disk_info():
    lines = []
    for part in psutil.disk_partitions():
        if 'snap' in part.mountpoint:
            continue
        usage = psutil.disk_usage(part.mountpoint)
        lines.append(f"{part.mountpoint}: {usage.free // (1024**3)}G libres / {usage.total // (1024**3)}G")
    return "\n".join(lines)


def get_memory_info():
    mem = psutil.virtual_memory()
    swap = psutil.swap_memory()
    return f"RAM: {mem.used // (1024**2)}M / {mem.total // (1024**2)}M\nSWAP: {swap.used // (1024**2)}M / {swap.total // (1024**2)}M"


def get_cpu_info():
    load1, load5, load15 = psutil.getloadavg()
    per_core = psutil.cpu_percent(percpu=True)
    per_core_info = "\n".join([f"Core {i}: {pct:.1f}%" for i, pct in enumerate(per_core)])
    return f"Load avg: {load1:.2f} {load5:.2f} {load15:.2f}\n{per_core_info}"


def get_gpu_info():
    try:
        output = subprocess.check_output(
            ['nvidia-smi', '--query-gpu=utilization.gpu,utilization.memory,memory.used,memory.total,temperature.gpu',
             '--format=csv,noheader,nounits'],
            encoding='utf-8'
        )
        lines = []
        for i, line in enumerate(output.strip().split('\n')):
            gpu_util, mem_util, mem_used, mem_total, temp = map(int, line.strip().split(', '))
            lines.append(f"GPU {i}: {gpu_util}% GPU | {mem_util}% MEM | {mem_used}/{mem_total} MiB | Temp: {temp}°C")
        return "\n".join(lines)
    except Exception:
        return "GPU: nvidia-smi non dispo"


def get_cpu_temp():
    try:
        temps = psutil.sensors_temperatures()
        if 'coretemp' in temps:
            core_temps = temps['coretemp']
            avg_temp = sum(t.current for t in core_temps if t.label.startswith('Core')) / len(core_temps)
            return f"Température CPU moyenne: {avg_temp:.1f}°C"
        return "Température CPU: non disponible"
    except Exception:
        return "Température CPU: erreur"


def get_docker_info():
    try:
        output = subprocess.check_output(['docker', 'ps', '-q'], encoding='utf-8')
        count = len([line for line in output.strip().split('\n') if line])
        return f"Containers Docker actifs: {count}"
    except Exception:
        return "Docker: non dispo"


def get_top_processes(limit=5):
    processes = []
    for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent']):
        try:
            info = proc.info
            if info['cpu_percent'] is None:
                continue
            processes.append(info)
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue

    processes = sorted(processes, key=lambda p: (p['cpu_percent'], p['memory_percent']), reverse=True)
    seen = set()
    top = []
    for proc in processes:
        name = proc['name']
        if name in seen:
            continue
        seen.add(name)
        top.append(f"{name} (PID {proc['pid']}): CPU {proc['cpu_percent']:.1f}%, MEM {proc['memory_percent']:.1f}%")
        if len(top) >= limit:
            break
    return "\n".join(top) if top else "Aucun processus trouvé."


def get_system_info():
    hostname = socket.gethostname()
    distro = platform.platform()
    kernel = platform.release()
    return f"🖥️ Machine: {hostname}\n🧬 Distribution: {distro}\n🧪 Noyau: {kernel}"


class InfoBox(Static):
    def __init__(self, title, getter, style):
        super().__init__()
        self.title = title
        self.getter = getter
        self.style = style

    def on_mount(self):
        self.update_panel()

    def update_panel(self):
        content = self.getter()
        self.update(Panel(Text(content, style=self.style), title=self.title))


class SystemDashboard(App):
    CSS = """
    Screen {
        layout: horizontal;
        padding: 1;
    }
    Vertical {
        width: 1fr;
        padding: 1;
    }
    Static {
        border: round white;
        padding: 1;
    }
    """

    refresh_interval = reactive(2.0)

    def compose(self) -> ComposeResult:
        self.box_sys = InfoBox("🛠 Système", get_system_info, "bright_white")
        self.box_disk = InfoBox("🗄 Disques", get_disk_info, "green")
        self.box_mem = InfoBox("🧠 Mémoire", get_memory_info, "cyan")
        self.box_docker = InfoBox("🐳 Docker", get_docker_info, "magenta")
        self.box_cpu = InfoBox("💻 CPU", get_cpu_info, "yellow")
        self.box_cputemp = InfoBox("🌡️ Temp. CPU", get_cpu_temp, "orange1")
        self.box_gpu = InfoBox("🎮 GPU NVIDIA", get_gpu_info, "red")
        self.box_proc = InfoBox("🔥 Top Processus", get_top_processes, "bright_blue")

        yield Horizontal(
            Vertical(self.box_sys, self.box_disk, self.box_mem, self.box_docker),
            Vertical(self.box_cpu, self.box_cputemp, self.box_gpu, self.box_proc),
        )

    def on_mount(self):
        self.set_interval(self.refresh_interval, self.refresh_all_boxes)

    def refresh_all_boxes(self):
        for box in [self.box_sys, self.box_disk, self.box_mem, self.box_docker, self.box_cpu, self.box_cputemp, self.box_gpu, self.box_proc]:
            box.update_panel()

    def on_key(self, event):
        if event.key == "r":
            self.refresh_all_boxes()
        elif event.key == "q":
            self.exit()


app = SystemDashboard()
app.run()