Effi/docs/TOOL-BESZEL-OPENWEBUI.md

Beszel Tool for OpenWebUI

Description

Permet à Effi d'accéder aux métriques système de FolzoCorp, EffiServer et NAS Synology via Beszel.

Installation

1. Admin Panel → Tools → Create Tool
2. Name: Beszel System Monitor
3. Description: Monitor system resources via Beszel. Usage: get_system_stats("folzocorp"), get_all_stats(), check_alerts("folzocorp"), get_gpu_status("folzocorp")
4. Coller le code ci-dessous
5. Save and Activate

Configuration

Les variables d'environnement à configurer:

• BESZEL_URL: URL du hub Beszel (défaut: http://192.168.1.25:8090)
• BESZEL_USER: Nom d'utilisateur Beszel
• BESZEL_PASSWORD: Mot de passe Beszel

Tool Code

import os
import requests
from typing import Optional, Dict, Any, List


class BeszelClient:
    def __init__(self):
        self.url = os.getenv("BESZEL_URL", "http://192.168.1.25:8090")
        self.username = os.getenv("BESZEL_USER", "admin")
        self.password = os.getenv("BESZEL_PASSWORD", "")
        self.token = None
    
    def login(self) -> bool:
        try:
            response = requests.post(
                f"{self.url}/api/auth/login",
                json={"identity": self.username, "password": self.password},
                timeout=10
            )
            if response.status_code == 200:
                self.token = response.json().get("token")
                return True
            return False
        except Exception:
            return False
    
    def get_systems(self) -> Optional[List[Dict]]:
        if not self.token and not self.login():
            return None
        try:
            headers = {"Authorization": f"Bearer {self.token}"}
            response = requests.get(f"{self.url}/api/systems", headers=headers, timeout=10)
            if response.status_code == 200:
                return response.json().get("items", [])
            return None
        except Exception:
            return None
    
    def get_system_stats(self, system_id: str) -> Optional[Dict[str, Any]]:
        if not self.token and not self.login():
            return None
        try:
            headers = {"Authorization": f"Bearer {self.token}"}
            response = requests.get(
                f"{self.url}/api/systems/{system_id}/stats",
                headers=headers,
                timeout=10
            )
            if response.status_code == 200:
                return response.json()
            return None
        except Exception:
            return None
    
    def format_stats_text(self, stats: Dict, system_name: str = "System") -> str:
        lines = []
        lines.append(f"=== {system_name} ===")
        lines.append(f"CPU: {stats.get('cpu', 0):.1f}%")
        lines.append(f"Mémoire: {stats.get('memory', 0):.1f}%")
        
        if 'swap' in stats:
            lines.append(f"Swap: {stats.get('swap', 0):.1f}%")
        
        if 'disk' in stats:
            for disk, info in stats['disk'].items():
                lines.append(f"Disque {disk}: {info.get('used_percent', 0):.1f}%")
        
        if 'load' in stats:
            load = stats['load']
            lines.append(f"Load: {load[0]:.2f} {load[1]:.2f} {load[2]:.2f}")
        
        if 'gpu' in stats and stats['gpu']:
            lines.append("")
            lines.append("GPU:")
            for gpu_name, gpu_data in stats['gpu'].items():
                vram_used = gpu_data.get('memory_used', 0) / (1024**3)
                vram_total = gpu_data.get('memory_total', 0) / (1024**3)
                vram_pct = (gpu_data.get('memory_used', 0) / gpu_data.get('memory_total', 1)) * 100
                lines.append(f"  {gpu_name}:")
                lines.append(f"    VRAM: {vram_used:.2f}GB / {vram_total:.2f}GB ({vram_pct:.1f}%)")
                lines.append(f"    Utilisation GPU: {gpu_data.get('utilization_gpu', 0)}%")
                if 'temperature' in gpu_data:
                    lines.append(f"    Température: {gpu_data.get('temperature', 0)}°C")
                if 'power_draw' in gpu_data:
                    lines.append(f"    Consommation: {gpu_data.get('power_draw', 0)}W")
        
        if 'containers' in stats and stats['containers']:
            lines.append("")
            lines.append("Containers Docker:")
            for container in stats['containers']:
                name = container.get('name', 'unknown')
                cpu = container.get('cpu_percent', 0)
                mem = container.get('memory_percent', 0)
                lines.append(f"  {name}: CPU {cpu:.1f}% MEM {mem:.1f}%")
        
        return "\n".join(lines)
    
    def check_thresholds(self, stats: Dict, thresholds: Optional[Dict] = None) -> List[str]:
        if thresholds is None:
            thresholds = {'cpu': 80, 'memory': 90, 'vram': 90}
        
        alerts = []
        
        cpu = stats.get('cpu', 0)
        if cpu > thresholds['cpu']:
            alerts.append(f"⚠️ CPU elevated: {cpu:.1f}%")
        
        memory = stats.get('memory', 0)
        if memory > thresholds['memory']:
            alerts.append(f"⚠️ Memory elevated: {memory:.1f}%")
        
        if 'gpu' in stats and stats['gpu']:
            for gpu_name, gpu_data in stats['gpu'].items():
                vram_pct = (gpu_data.get('memory_used', 0) / gpu_data.get('memory_total', 1)) * 100
                if vram_pct > thresholds['vram']:
                    alerts.append(f"⚠️ VRAM {gpu_name} elevated: {vram_pct:.1f}%")
                
                util = gpu_data.get('utilization_gpu', 0)
                if util > thresholds['vram']:
                    alerts.append(f"⚠️ GPU {gpu_name} utilization elevated: {util}%")
        
        return alerts


def get_all_stats():
    """Get stats from all monitored systems."""
    client = BeszelClient()
    systems = client.get_systems()
    
    if not systems:
        return "[ERROR] Cannot connect to Beszel or no systems found"
    
    results = []
    for sys_info in systems:
        sys_id = sys_info.get('id')
        sys_name = sys_info.get('name', sys_id)
        stats = client.get_system_stats(sys_id)
        
        if stats:
            results.append(client.format_stats_text(stats, sys_name))
        else:
            results.append(f"=== {sys_name} ===\n[ERROR] Cannot get stats")
    
    return "\n\n".join(results)


def get_system_stats(system_name: str):
    """Get stats for a specific system. Options: folzocorp, effiserver, synology-nas"""
    client = BeszelClient()
    systems = client.get_systems()
    
    if not systems:
        return "[ERROR] Cannot connect to Beszel"
    
    system_map = {
        'folzocorp': 'folzocorp',
        'effiserver': 'effiserver', 
        'synology': 'synology-nas',
        'nas': 'synology-nas'
    }
    
    target_id = system_map.get(system_name.lower(), system_name.lower())
    
    for sys_info in systems:
        if sys_info.get('id') == target_id or sys_info.get('name', '').lower() == system_name.lower():
            stats = client.get_system_stats(sys_info.get('id'))
            if stats:
                return client.format_stats_text(stats, sys_info.get('name', system_name))
            return f"[ERROR] Cannot get stats for {system_name}"
    
    available = [s.get('name', s.get('id')) for s in systems]
    return f"[ERROR] System '{system_name}' not found. Available: {', '.join(available)}"


def check_alerts(system_name: str = "all"):
    """Check for resource alerts. Use 'all' or specific system name."""
    client = BeszelClient()
    systems = client.get_systems()
    
    if not systems:
        return "[ERROR] Cannot connect to Beszel"
    
    alerts = []
    
    for sys_info in systems:
        sys_id = sys_info.get('id')
        sys_name = sys_info.get('name', sys_id)
        
        if system_name.lower() != 'all' and sys_name.lower() != system_name.lower():
            continue
        
        stats = client.get_system_stats(sys_id)
        if stats:
            system_alerts = client.check_thresholds(stats)
            if system_alerts:
                alerts.append(f"=== {sys_name} ===")
                alerts.extend(system_alerts)
    
    if not alerts:
        return "✅ All systems nominal"
    
    return "\n".join(alerts)


def get_gpu_status(system_name: str = "folzocorp"):
    """Get GPU status for a specific system. Most useful for folzocorp (AMD RX 7800 XT)."""
    client = BeszelClient()
    systems = client.get_systems()
    
    if not systems:
        return "[ERROR] Cannot connect to Beszel"
    
    target_id = None
    for sys_info in systems:
        if sys_info.get('id') == system_name.lower() or sys_info.get('name', '').lower() == system_name.lower():
            target_id = sys_info.get('id')
            break
    
    if not target_id:
        available = [s.get('name', s.get('id')) for s in systems]
        return f"[ERROR] System '{system_name}' not found. Available: {', '.join(available)}"
    
    stats = client.get_system_stats(target_id)
    if not stats:
        return f"[ERROR] Cannot get stats for {system_name}"
    
    if 'gpu' not in stats or not stats['gpu']:
        return f"[INFO] No GPU data for {system_name}"
    
    lines = [f"=== GPU Status: {system_name} ==="]
    for gpu_name, gpu_data in stats['gpu'].items():
        vram_used = gpu_data.get('memory_used', 0) / (1024**3)
        vram_total = gpu_data.get('memory_total', 0) / (1024**3)
        vram_pct = (gpu_data.get('memory_used', 0) / gpu_data.get('memory_total', 1)) * 100
        
        lines.append(f"\n{gpu_name}:")
        lines.append(f"  VRAM: {vram_used:.2f}GB / {vram_total:.2f}GB ({vram_pct:.1f}%)")
        lines.append(f"  GPU Utilization: {gpu_data.get('utilization_gpu', 0)}%")
        
        if 'temperature' in gpu_data:
            lines.append(f"  Temperature: {gpu_data.get('temperature', 0)}°C")
        if 'power_draw' in gpu_data:
            lines.append(f"  Power: {gpu_data.get('power_draw', 0)}W")
    
    return "\n".join(lines)


def can_run_comfyui():
    """Check if there's enough RAM and VRAM to run ComfyUI."""
    client = BeszelClient()
    
    stats = client.get_system_stats("folzocorp")
    if not stats:
        return "[ERROR] Cannot get FolzoCorp stats"
    
    memory_pct = stats.get('memory', 0)
    
    vram_info = None
    if 'gpu' in stats and stats['gpu']:
        for gpu_name, gpu_data in stats['gpu'].items():
            vram_total = gpu_data.get('memory_total', 0) / (1024**3)
            vram_free = gpu_data.get('memory_total', 0) - gpu_data.get('memory_used', 0)
            vram_free_gb = vram_free / (1024**3)
            vram_pct = (gpu_data.get('memory_used', 0) / gpu_data.get('memory_total', 1)) * 100
            vram_info = f"{gpu_name}: {vram_free_gb:.1f}GB free ({vram_pct:.1f}% used)"
    
    can_run = memory_pct < 85 and (not vram_info or vram_pct < 85)
    
    lines = ["=== ComfyUI Readiness Check ==="]
    lines.append(f"RAM: {100-memory_pct:.1f}% available ({memory_pct:.1f}% used)")
    if vram_info:
        lines.append(f"VRAM: {vram_info}")
    lines.append(f"\n结论: {'✅ Can run ComfyUI' if can_run else '⚠️ Resources low, may be slow'}")
    
    return "\n".join(lines)

Usage Examples

# Stats de tous les systèmes
get_all_stats()

# Stats d'un système spécifique
get_system_stats("folzocorp")
get_system_stats("effiserver")
get_system_stats("synology")

# Vérifier les alertes
check_alerts()
check_alerts("folzocorp")

# Status GPU (surtout utile pour FolzoCorp avec RX 7800 XT)
get_gpu_status("folzocorp")

# Vérifier si ComfyUI peut tourner
can_run_comfyui()

System IDs

System	ID	GPU
FolzoCorp	folzocorp	AMD RX 7800 XT
EffiServer	effiserver	None (Intel J5040)
NAS Synology	synology-nas	Intel HD Graphics (transcodage)

---

Created 12 April 2026