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, }