Folzo/Effi
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: Rename class to Tools for OpenWebUI compatibility

Browse Source
This commit is contained in:
Folzo
2026-04-12 13:39:12 +02:00
parent b42cfcabce
commit 3dbd2bcc17
1 changed files with 159 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/TOOL-BESZEL-OPENWEBUI.md
+159 -149
View File
@@ -8,7 +8,7 @@ Permet à Effi d'accéder aux métriques système de FolzoCorp, EffiServer et NA
1. **Admin Panel** → **Tools** → **Create Tool** 1. **Admin Panel** → **Tools** → **Create Tool**
2. **Name**: `Beszel System Monitor` 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")` 3. **Description**: `Monitor system resources via Beszel. Usage: get_all_stats(), get_system_stats("folzocorp"), check_alerts("folzocorp"), get_gpu_status("folzocorp"), can_run_comfyui()`
4. **Coller le code ci-dessous** 4. **Coller le code ci-dessous**
5. **Save and Activate** 5. **Save and Activate**
@@ -24,17 +24,16 @@ Les variables d'environnement à configurer:
```python ```python
import os import os
import requests import requests
from typing import Optional, Dict, Any, List
class BeszelClient: class Tools:
def __init__(self): def __init__(self):
self.url = os.getenv("BESZEL_URL", "http://192.168.1.25:8090") self.url = os.getenv("BESZEL_URL", "http://192.168.1.25:8090")
self.username = os.getenv("BESZEL_USER", "admin") self.username = os.getenv("BESZEL_USER", "admin")
self.password = os.getenv("BESZEL_PASSWORD", "") self.password = os.getenv("BESZEL_PASSWORD", "")
self.token = None self.token = None
def login(self) -> bool: def _login(self):
try: try:
response = requests.post( response = requests.post(
f"{self.url}/api/auth/login", f"{self.url}/api/auth/login",
@@ -48,8 +47,8 @@ class BeszelClient:
except Exception: except Exception:
return False return False
def get_systems(self) -> Optional[List[Dict]]: def _get_systems(self):
if not self.token and not self.login(): if not self.token and not self._login():
return None return None
try: try:
headers = {"Authorization": f"Bearer {self.token}"} headers = {"Authorization": f"Bearer {self.token}"}
@@ -60,8 +59,8 @@ class BeszelClient:
except Exception: except Exception:
return None return None
def get_system_stats(self, system_id: str) -> Optional[Dict[str, Any]]: def _get_system_stats(self, system_id):
if not self.token and not self.login(): if not self.token and not self._login():
return None return None
try: try:
headers = {"Authorization": f"Bearer {self.token}"} headers = {"Authorization": f"Bearer {self.token}"}
@@ -76,7 +75,7 @@ class BeszelClient:
except Exception: except Exception:
return None return None
def format_stats_text(self, stats: Dict, system_name: str = "System") -> str: def _format_stats_text(self, stats, system_name="System"):
lines = [] lines = []
lines.append(f"=== {system_name} ===") lines.append(f"=== {system_name} ===")
lines.append(f"CPU: {stats.get('cpu', 0):.1f}%") lines.append(f"CPU: {stats.get('cpu', 0):.1f}%")
@@ -119,7 +118,7 @@ class BeszelClient:
return "\n".join(lines) return "\n".join(lines)
def check_thresholds(self, stats: Dict, thresholds: Optional[Dict] = None) -> List[str]: def _check_thresholds(self, stats, thresholds=None):
if thresholds is None: if thresholds is None:
thresholds = {'cpu': 80, 'memory': 90, 'vram': 90} thresholds = {'cpu': 80, 'memory': 90, 'vram': 90}
@@ -144,159 +143,170 @@ class BeszelClient:
alerts.append(f"⚠️ GPU {gpu_name} utilization elevated: {util}%") alerts.append(f"⚠️ GPU {gpu_name} utilization elevated: {util}%")
return alerts return alerts
def get_all_stats():
"""Get stats from all monitored systems."""
client = BeszelClient()
systems = client.get_systems()
if not systems: def get_all_stats(self):
return "[ERROR] Cannot connect to Beszel or no systems found" """Get stats from all monitored systems."""
systems = self._get_systems()
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: if not systems:
results.append(client.format_stats_text(stats, sys_name)) return "[ERROR] Cannot connect to Beszel or no systems found"
else:
results.append(f"=== {sys_name} ===\n[ERROR] Cannot get stats") results = []
for sys_info in systems:
return "\n\n".join(results) sys_id = sys_info.get('id')
sys_name = sys_info.get('name', sys_id)
stats = self._get_system_stats(sys_id)
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: if stats:
return client.format_stats_text(stats, sys_info.get('name', system_name)) results.append(self._format_stats_text(stats, sys_name))
return f"[ERROR] Cannot get stats for {system_name}" else:
results.append(f"=== {sys_name} ===\n[ERROR] Cannot get stats")
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(): return "\n\n".join(results)
continue
def get_system_stats(self, system_name):
"""Get stats for a specific system. Options: folzocorp, effiserver, synology-nas"""
systems = self._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 = self._get_system_stats(sys_info.get('id'))
if stats:
return self._format_stats_text(stats, sys_info.get('name', system_name))
return f"[ERROR] Cannot get stats for {system_name}"
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] available = [s.get('name', s.get('id')) for s in systems]
return f"[ERROR] System '{system_name}' not found. Available: {', '.join(available)}" return f"[ERROR] System '{system_name}' not found. Available: {', '.join(available)}"
stats = client.get_system_stats(target_id) def check_alerts(self, system_name="all"):
if not stats: """Check for resource alerts. Use 'all' or specific system name."""
return f"[ERROR] Cannot get stats for {system_name}" systems = self._get_systems()
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}:") if not systems:
lines.append(f" VRAM: {vram_used:.2f}GB / {vram_total:.2f}GB ({vram_pct:.1f}%)") return "[ERROR] Cannot connect to Beszel"
lines.append(f" GPU Utilization: {gpu_data.get('utilization_gpu', 0)}%")
if 'temperature' in gpu_data: alerts = []
lines.append(f" Temperature: {gpu_data.get('temperature', 0)}°C")
if 'power_draw' in gpu_data: for sys_info in systems:
lines.append(f" Power: {gpu_data.get('power_draw', 0)}W") 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 = self._get_system_stats(sys_id)
if stats:
system_alerts = self._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)
return "\n".join(lines) def get_gpu_status(self, system_name="folzocorp"):
"""Get GPU status for a specific system. Most useful for folzocorp (AMD RX 7800 XT)."""
systems = self._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 = self._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(self):
"""Check if there's enough RAM and VRAM to run ComfyUI."""
stats = self._get_system_stats("folzocorp")
if not stats:
return "[ERROR] Cannot get FolzoCorp stats"
memory_pct = stats.get('memory', 0)
vram_info = None
vram_pct = 0
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 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"\nConclusion: {'✅ Can run ComfyUI' if can_run else '⚠️ Resources low, may be slow'}")
return "\n".join(lines)
def get_all_stats():
tools = Tools()
return tools.get_all_stats()
def get_system_stats(system_name):
tools = Tools()
return tools.get_system_stats(system_name)
def check_alerts(system_name="all"):
tools = Tools()
return tools.check_alerts(system_name)
def get_gpu_status(system_name="folzocorp"):
tools = Tools()
return tools.get_gpu_status(system_name)
def can_run_comfyui(): def can_run_comfyui():
"""Check if there's enough RAM and VRAM to run ComfyUI.""" tools = Tools()
client = BeszelClient() return tools.can_run_comfyui()
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 ## Usage Examples