302 lines
11 KiB
Python
302 lines
11 KiB
Python
"""
|
|
Linux AMD GPU Backend
|
|
=====================
|
|
Reads GPU stats from Linux sysfs (/sys/class/drm/) and hwmon for AMD GPUs.
|
|
Zero external dependencies — uses standard library only.
|
|
|
|
Data sources:
|
|
- GPU utilization: /sys/class/drm/card*/device/gpu_busy_percent
|
|
- VRAM: /sys/class/drm/card*/device/mem_info_vram_total
|
|
/sys/class/drm/card*/device/mem_info_vram_used
|
|
- Temperature: /sys/class/drm/card*/device/hwmon/hwmon*/temp1_input
|
|
- Core clock: /sys/class/drm/card*/device/pp_dpm_sclk (average from PP table)
|
|
- Power: /sys/class/drm/card*/device/hwmon/hwmon*/power1_average
|
|
|
|
APU Support:
|
|
- APUs with shared memory: memory_total=0, memory_shared=system_RAM
|
|
- Detects APU by checking VRAM size < 512MB
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
import time
|
|
from typing import Optional
|
|
|
|
from .base import MonitorBackend, HardwareStats
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class LinuxAMDGPUBackend(MonitorBackend):
|
|
"""AMD GPU monitoring on Linux via sysfs + hwmon"""
|
|
name = "linux-amdgpu-sysfs"
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.vendor = "amd"
|
|
self._card_paths: list[str] = []
|
|
self._vram_total: int = 0
|
|
self._is_apu: bool = False
|
|
self._system_ram_total: int = 0
|
|
self._hwmon_paths: list[str] = []
|
|
self._temp_cache_timestamp: float = 0
|
|
self._cached_temp: float = 0.0
|
|
self._TEMP_CACHE_SECONDS = 5.0
|
|
|
|
def initialize(self) -> bool:
|
|
import platform
|
|
if platform.system().lower() != "linux":
|
|
return False
|
|
|
|
try:
|
|
self._find_amd_cards()
|
|
if not self._card_paths:
|
|
return False
|
|
|
|
self._detect_vram()
|
|
self._detect_hwmon()
|
|
self._detect_system_ram()
|
|
|
|
self.gpu_count = len(self._card_paths)
|
|
self.gpu_names = [f"AMD GPU {i}" for i in range(self.gpu_count)]
|
|
self.available = True
|
|
|
|
logger.info(
|
|
f"Linux AMD: {self.gpu_count} GPU(s), "
|
|
f"VRAM={self._vram_total / (1024*1024):.0f}MB, "
|
|
f"APU={self._is_apu}"
|
|
)
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Linux AMD init error: {e}")
|
|
return False
|
|
|
|
def _find_amd_cards(self):
|
|
"""Find AMD GPU card directories in /sys/class/drm/"""
|
|
drm_path = "/sys/class/drm"
|
|
if not os.path.exists(drm_path):
|
|
return
|
|
|
|
for entry in sorted(os.listdir(drm_path)):
|
|
if not entry.startswith("card") or "-" in entry:
|
|
continue
|
|
|
|
card_path = os.path.join(drm_path, entry)
|
|
|
|
# Check vendor
|
|
vendor_file = os.path.join(card_path, "device", "vendor")
|
|
if os.path.exists(vendor_file):
|
|
try:
|
|
with open(vendor_file, 'r') as f:
|
|
vendor_id = f.read().strip().lower()
|
|
if '1002' in vendor_id or 'amd' in vendor_id:
|
|
self._card_paths.append(card_path)
|
|
except Exception:
|
|
pass
|
|
|
|
# Fallback: check for amdgpu driver
|
|
if not self._card_paths or card_path != self._card_paths[-1]:
|
|
if os.path.exists(os.path.join(card_path, "device", "gpu_busy_percent")):
|
|
if card_path not in self._card_paths:
|
|
self._card_paths.append(card_path)
|
|
|
|
def _detect_vram(self):
|
|
"""Read VRAM total from sysfs. Detect APU shared memory."""
|
|
vram_total = 0
|
|
for card_path in self._card_paths:
|
|
vram_file = os.path.join(card_path, "device", "mem_info_vram_total")
|
|
if os.path.exists(vram_file):
|
|
try:
|
|
with open(vram_file, 'r') as f:
|
|
val = int(f.read().strip())
|
|
if val > vram_total:
|
|
vram_total = val
|
|
except (ValueError, OSError):
|
|
pass
|
|
|
|
self._vram_total = vram_total
|
|
|
|
# APU detection: < 512MB VRAM means shared memory
|
|
if 0 < vram_total < 512 * 1024 * 1024:
|
|
self._is_apu = True
|
|
self._vram_total = vram_total # Keep actual value, even if small
|
|
elif vram_total == 0:
|
|
self._is_apu = True # Likely APU with no dedicated VRAM reporting
|
|
|
|
def _detect_hwmon(self):
|
|
"""Find hwmon paths for AMD GPU temperatures."""
|
|
for card_path in self._card_paths:
|
|
hwmon_dir = os.path.join(card_path, "device", "hwmon")
|
|
if os.path.exists(hwmon_dir):
|
|
for hwmon_entry in os.listdir(hwmon_dir):
|
|
hwmon_path = os.path.join(hwmon_dir, hwmon_entry)
|
|
name_file = os.path.join(hwmon_path, "name")
|
|
if os.path.exists(name_file):
|
|
try:
|
|
with open(name_file, 'r') as f:
|
|
name = f.read().strip().lower()
|
|
if 'amdgpu' in name:
|
|
self._hwmon_paths.append(hwmon_path)
|
|
except Exception:
|
|
pass
|
|
|
|
def _detect_system_ram(self):
|
|
"""Read total system RAM for APU shared memory tracking."""
|
|
try:
|
|
with open('/proc/meminfo', 'r') as f:
|
|
for line in f:
|
|
if line.startswith('MemTotal:'):
|
|
parts = line.split()
|
|
if len(parts) >= 2:
|
|
self._system_ram_total = int(parts[1]) * 1024
|
|
break
|
|
except Exception:
|
|
self._system_ram_total = 8 * 1024 * 1024 * 1024 # 8GB fallback
|
|
|
|
def _read_sysfs(self, path: str) -> Optional[str]:
|
|
"""Safely read a sysfs file."""
|
|
try:
|
|
if os.path.exists(path):
|
|
with open(path, 'r') as f:
|
|
return f.read().strip()
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
def get_stats(self, gpu_id: int = 0) -> HardwareStats:
|
|
try:
|
|
if gpu_id >= len(self._card_paths):
|
|
return HardwareStats(
|
|
gpu_id=gpu_id,
|
|
vendor="amd",
|
|
is_available=False,
|
|
error_message=f"GPU {gpu_id} not found",
|
|
)
|
|
|
|
card_path = self._card_paths[gpu_id]
|
|
name = self.gpu_names[gpu_id] if gpu_id < len(self.gpu_names) else f"AMD GPU {gpu_id}"
|
|
|
|
stats = HardwareStats(
|
|
gpu_id=gpu_id,
|
|
gpu_name=name,
|
|
vendor="amd",
|
|
driver="amdgpu-sysfs",
|
|
is_apu=self._is_apu,
|
|
is_available=True,
|
|
)
|
|
|
|
# GPU Utilization (%)
|
|
util_val = self._read_sysfs(os.path.join(card_path, "device", "gpu_busy_percent"))
|
|
if util_val:
|
|
try:
|
|
stats.utilization_gpu = min(100.0, max(0.0, float(util_val)))
|
|
except ValueError:
|
|
pass
|
|
|
|
# VRAM
|
|
vram_used_file = os.path.join(card_path, "device", "mem_info_vram_used")
|
|
vram_used = self._read_sysfs(vram_used_file)
|
|
if vram_used and self._vram_total > 0:
|
|
try:
|
|
used = int(vram_used)
|
|
stats.memory_used = used
|
|
stats.memory_total = self._vram_total
|
|
stats.memory_free = max(0, self._vram_total - used)
|
|
if self._vram_total > 0:
|
|
stats.utilization_memory = (used / self._vram_total) * 100.0
|
|
except ValueError:
|
|
pass
|
|
|
|
# APU shared memory fallback
|
|
if self._is_apu and stats.memory_total == 0:
|
|
try:
|
|
import psutil
|
|
svmem = psutil.virtual_memory()
|
|
stats.memory_shared = self._system_ram_total
|
|
stats.memory_used = int(svmem.used * 0.25) # Estimate GPU portion
|
|
stats.utilization_memory = svmem.percent
|
|
except ImportError:
|
|
stats.memory_shared = self._system_ram_total
|
|
|
|
# Temperature (cached)
|
|
stats.temperature = self._get_temperature()
|
|
|
|
# Core Clock
|
|
clock_val = self._read_sysfs(os.path.join(card_path, "device", "pp_dpm_sclk"))
|
|
if clock_val:
|
|
try:
|
|
# Parse average clock from PP table
|
|
lines = clock_val.strip().split('\n')
|
|
for line in lines:
|
|
if 'Mhz' in line or 'MHz' in line:
|
|
parts = line.split(':')
|
|
if len(parts) >= 2:
|
|
# Extract number before MHz
|
|
import re
|
|
match = re.search(r'(\d+)\s*Mhz', parts[-1], re.IGNORECASE)
|
|
if match:
|
|
stats.core_clock = int(match.group(1))
|
|
break
|
|
except Exception:
|
|
pass
|
|
|
|
# Power draw
|
|
power = self._read_power(card_path)
|
|
if power is not None:
|
|
stats.power_draw = power
|
|
|
|
return stats
|
|
|
|
except Exception as e:
|
|
logger.error(f"Linux AMD get_stats({gpu_id}) error: {e}")
|
|
return HardwareStats(
|
|
gpu_id=gpu_id,
|
|
vendor="amd",
|
|
is_available=False,
|
|
error_message=str(e),
|
|
)
|
|
|
|
def _get_temperature(self) -> float:
|
|
"""Read GPU temperature from hwmon with caching."""
|
|
now = time.time()
|
|
if now - self._temp_cache_timestamp < self._TEMP_CACHE_SECONDS:
|
|
return self._cached_temp
|
|
|
|
temp = 0.0
|
|
for hwmon_path in self._hwmon_paths:
|
|
try:
|
|
for entry in sorted(os.listdir(hwmon_path)):
|
|
if entry.startswith("temp") and entry.endswith("_input"):
|
|
temp_file = os.path.join(hwmon_path, entry)
|
|
try:
|
|
with open(temp_file, 'r') as f:
|
|
raw = int(f.read().strip())
|
|
celsius = raw / 1000.0
|
|
if 20 <= celsius <= 120:
|
|
if celsius > temp:
|
|
temp = celsius
|
|
except (ValueError, OSError):
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
self._cached_temp = temp
|
|
self._temp_cache_timestamp = now
|
|
return temp
|
|
|
|
def _read_power(self, card_path: str) -> Optional[float]:
|
|
"""Read GPU power draw from hwmon."""
|
|
for hwmon_path in self._hwmon_paths:
|
|
try:
|
|
power_file = os.path.join(hwmon_path, "power1_average")
|
|
if os.path.exists(power_file):
|
|
with open(power_file, 'r') as f:
|
|
raw = int(f.read().strip())
|
|
# Convert microwatts to watts
|
|
return raw / 1_000_000.0
|
|
except (ValueError, OSError):
|
|
pass
|
|
return None
|
|
|
|
def close(self):
|
|
pass |