140 lines
4.1 KiB
Python
140 lines
4.1 KiB
Python
import time
|
|
import shutil
|
|
import subprocess
|
|
from typing import Any, Dict, Optional, Tuple
|
|
|
|
_last_cpu: Optional[Tuple[float, float]] = None
|
|
_last_net: Optional[Tuple[float, float, float]] = None
|
|
|
|
|
|
def _read_cpu_times() -> Tuple[float, float]:
|
|
with open("/proc/stat", "r", encoding="utf-8") as handle:
|
|
line = handle.readline()
|
|
parts = line.strip().split()
|
|
if not parts or parts[0] != "cpu":
|
|
return 0.0, 0.0
|
|
values = [float(p) for p in parts[1:]]
|
|
total = sum(values)
|
|
idle = values[3] if len(values) > 3 else 0.0
|
|
return total, idle
|
|
|
|
|
|
def _cpu_percent() -> float:
|
|
global _last_cpu
|
|
total, idle = _read_cpu_times()
|
|
if _last_cpu is None:
|
|
_last_cpu = (total, idle)
|
|
return 0.0
|
|
last_total, last_idle = _last_cpu
|
|
_last_cpu = (total, idle)
|
|
delta_total = total - last_total
|
|
delta_idle = idle - last_idle
|
|
if delta_total <= 0:
|
|
return 0.0
|
|
return max(0.0, min(100.0, (delta_total - delta_idle) / delta_total * 100.0))
|
|
|
|
|
|
def _read_meminfo() -> Dict[str, float]:
|
|
data: Dict[str, float] = {}
|
|
with open("/proc/meminfo", "r", encoding="utf-8") as handle:
|
|
for line in handle:
|
|
key, value = line.split(":", 1)
|
|
parts = value.strip().split()
|
|
if not parts:
|
|
continue
|
|
data[key] = float(parts[0])
|
|
return data
|
|
|
|
|
|
def _read_net_bytes() -> Tuple[float, float]:
|
|
rx_total = 0.0
|
|
tx_total = 0.0
|
|
with open("/proc/net/dev", "r", encoding="utf-8") as handle:
|
|
for line in handle:
|
|
if ":" not in line:
|
|
continue
|
|
iface, stats = line.split(":", 1)
|
|
iface = iface.strip()
|
|
if iface == "lo":
|
|
continue
|
|
fields = stats.split()
|
|
if len(fields) < 16:
|
|
continue
|
|
rx_total += float(fields[0])
|
|
tx_total += float(fields[8])
|
|
return rx_total, tx_total
|
|
|
|
|
|
def _net_mbps() -> Tuple[float, float]:
|
|
global _last_net
|
|
now = time.time()
|
|
rx, tx = _read_net_bytes()
|
|
if _last_net is None:
|
|
_last_net = (now, rx, tx)
|
|
return 0.0, 0.0
|
|
last_time, last_rx, last_tx = _last_net
|
|
_last_net = (now, rx, tx)
|
|
delta_t = now - last_time
|
|
if delta_t <= 0:
|
|
return 0.0, 0.0
|
|
rx_mbps = (rx - last_rx) * 8.0 / (delta_t * 1_000_000.0)
|
|
tx_mbps = (tx - last_tx) * 8.0 / (delta_t * 1_000_000.0)
|
|
return max(0.0, rx_mbps), max(0.0, tx_mbps)
|
|
|
|
|
|
def _gpu_metrics() -> Dict[str, Any]:
|
|
if not shutil.which("nvidia-smi"):
|
|
return {
|
|
"gpu_vram_total_mb": None,
|
|
"gpu_vram_used_percent": None,
|
|
"gpu_present": False,
|
|
}
|
|
try:
|
|
output = subprocess.check_output(
|
|
[
|
|
"nvidia-smi",
|
|
"--query-gpu=memory.total,memory.used",
|
|
"--format=csv,noheader,nounits",
|
|
],
|
|
text=True,
|
|
).strip()
|
|
if not output:
|
|
raise ValueError("empty nvidia-smi output")
|
|
total_str, used_str = output.split(",", 1)
|
|
total_mb = float(total_str.strip())
|
|
used_mb = float(used_str.strip())
|
|
used_percent = 0.0 if total_mb == 0 else used_mb / total_mb * 100.0
|
|
return {
|
|
"gpu_vram_total_mb": total_mb,
|
|
"gpu_vram_used_percent": used_percent,
|
|
"gpu_present": True,
|
|
}
|
|
except Exception:
|
|
return {
|
|
"gpu_vram_total_mb": None,
|
|
"gpu_vram_used_percent": None,
|
|
"gpu_present": False,
|
|
}
|
|
|
|
|
|
def get_system_metrics() -> Dict[str, Any]:
|
|
meminfo = _read_meminfo()
|
|
total_kb = meminfo.get("MemTotal", 0.0)
|
|
available_kb = meminfo.get("MemAvailable", 0.0)
|
|
used_kb = max(0.0, total_kb - available_kb)
|
|
ram_total_mb = total_kb / 1024.0
|
|
ram_used_percent = 0.0 if total_kb == 0 else used_kb / total_kb * 100.0
|
|
|
|
cpu_percent = _cpu_percent()
|
|
rx_mbps, tx_mbps = _net_mbps()
|
|
gpu = _gpu_metrics()
|
|
|
|
return {
|
|
"cpu_percent": cpu_percent,
|
|
"ram_total_mb": ram_total_mb,
|
|
"ram_used_percent": ram_used_percent,
|
|
"net_rx_mbps": rx_mbps,
|
|
"net_tx_mbps": tx_mbps,
|
|
**gpu,
|
|
}
|