""" 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 - Product name: ``lspci -mm`` (sysfs does NOT expose a human-readable name for most consumer AMD GPUs — this is the only reliable source without requiring a userspace tool like rocminfo.) 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 re import subprocess import time from typing import Optional, List from .base import MonitorBackend, HardwareStats logger = logging.getLogger(__name__) def _read_product_name_from_lspci(pci_slot: str) -> Optional[str]: """Look up the human-readable product name for a PCI slot via ``lspci -mm``. sysfs exposes ``vendor`` / ``device`` IDs (e.g. ``1002:747e``) but no product string for most consumer AMD GPUs. ``lspci -mm`` gives us a machine-readable ``"VGA compatible controller" "Vendor" "Product Name"`` tuple that's easy to parse. Falls back to ``None`` when lspci is unavailable or doesn't know the device. """ try: result = subprocess.run( ["lspci", "-mm", "-s", pci_slot], capture_output=True, text=True, timeout=3, ) except (FileNotFoundError, subprocess.TimeoutExpired): return None except Exception as e: logger.debug(f"lspci lookup failed for {pci_slot}: {e}") return None if result.returncode != 0 or not result.stdout.strip(): return None # Example output: # 03:00.0 "VGA" "Advanced Micro Devices, Inc. [AMD/ATI]" "Navi 32 [Radeon RX 7700 XT / 7800 XT]" ... # Quoted fields; we want field index 2 (the product name). parts = re.findall(r'"([^"]*)"', result.stdout) if len(parts) >= 3: product = parts[2].strip() # Strip "AMD/ATI" / "[AMD/ATI]" style vendor tags that lspci sometimes # leaves in the product string on older versions. product = re.sub(r"\[?AMD/ATI\]?\s*", "", product).strip() # Strip "Navi N" / "Navi NN" silicon code prefix when a marketing # name is present — the marketing name is more user-friendly. m = re.search(r"\[(Radeon[^\]]+)\]", product) if m: return m.group(1).strip() return product or None return None 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._pci_slots: list[str] = [] # PCI BDF per card (e.g. "0000:03:00.0") 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._cached_fan: Optional[int] = None # None => not yet sampled self._fan_cache_timestamp: float = 0 self._TEMP_CACHE_SECONDS = 2.0 self._FAN_CACHE_SECONDS = 1.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._resolve_product_names() self.gpu_count = len(self._card_paths) self.gpu_names = [ n if n else f"AMD GPU {i}" for i, n in enumerate(self._product_names) ] or [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"names={self.gpu_names}, " 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 # Match a PCI BDF anywhere in the path, e.g. "0000:03:00.0". # The card's device symlink points into # /sys/devices/pci.../BBBB:BB:BB.B/drm/cardN, so the *last* BDF in # the path is the leaf GPU device — root bridges upstream appear # earlier and would confuse lspci's -s match. bdf_re = re.compile(r"([0-9a-fA-F]{4}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}\.\d+)") 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) # The device symlink resolves to something like # /sys/devices/pci0000:00/.../0000:03:00.0/drm/cardN # — pull the leaf PCI BDF out of it. real = os.path.realpath(card_path) matches = bdf_re.findall(real) self._pci_slots.append(matches[-1] if matches else "") except (OSError, ValueError) as e: logger.debug(f"Linux AMD: BDF lookup failed: {e}") # 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) real = os.path.realpath(card_path) matches = bdf_re.findall(real) self._pci_slots.append(matches[-1] if matches else "") def _resolve_product_names(self): """Populate ``self._product_names`` via ``lspci -mm``. sysfs exposes only PCI vendor/device IDs (e.g. ``1002:747e``) but no human-readable product string for consumer AMD GPUs. We fall back to ``lspci`` to recover something like ``"Radeon RX 7800 XT"``. Emits a single WARNING if lspci returns nothing for any GPU — the overlay will fall back to "AMD GPU N" placeholders, and the user should know why instead of staring at a generic widget. """ self._product_names = [] lspci_missing_count = 0 for slot in self._pci_slots: name = _read_product_name_from_lspci(slot) if slot else None # Normalize the lspci BDF "0000:03:00.0" into the "-s" argument # form "03:00.0" if needed. lspci accepts both. if not name and slot: short = re.sub(r"^[0-9a-fA-F]{4}:", "", slot) name = _read_product_name_from_lspci(short) if not name: lspci_missing_count += 1 self._product_names.append(name) if lspci_missing_count == len(self._pci_slots) and self._pci_slots: # ALL GPUs missing — likely lspci is not installed. logger.warning( "Linux AMD: lspci is unavailable; install 'pciutils' to " "display real GPU product names instead of 'AMD GPU N'." ) elif lspci_missing_count > 0: logger.warning( f"Linux AMD: lspci did not resolve names for " f"{lspci_missing_count}/{len(self._pci_slots)} GPU(s)" ) else: logger.debug(f"Linux AMD: resolved product names: {self._product_names}") 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. Logs a one-shot WARNING if no amdgpu hwmon device is found — without it the overlay will show ``N/A`` for temperature and the user is left guessing why (most common cause: missing ``amdgpu`` kernel module or insufficient permissions on ``/sys/class/drm/card*/device/hwmon``). """ 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 (OSError, ValueError) as e: logger.debug(f"Linux AMD: hwmon driver name read failed: {e}") if self._card_paths and not self._hwmon_paths: logger.warning( "Linux AMD: no amdgpu hwmon device found — temperature " "and power readings will be N/A (check that the amdgpu " "kernel module is loaded and hwmon is readable)." ) 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 (OSError, ValueError) as e: logger.debug(f"Linux AMD: system RAM read failed: {e}") 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 (OSError, ValueError) as e: logger.debug(f"Linux AMD: sysfs read failed for {path}: {e}") 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() # Fan speed (RPM -> duty %) stats.fan_speed = self._get_fan_speed() # 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 (ValueError, AttributeError) as e: logger.debug(f"Linux AMD: pp_dpm_sclk parse failed: {e}") # 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. Sensor priority (highest wins): 1. ``temp*_label == 'junction'`` — GPU die temp, the value AMD Radeon Software and HWiNFO report as "GPU Temperature". Best signal for thermal throttling on RDNA3. 2. ``temp*_label == 'edge'`` — case edge temp, the conservative number most users see when they touch the backplate. 3. Fallback: max of all ``temp*_input`` files within 20–120°C. """ now = time.time() if now - self._temp_cache_timestamp < self._TEMP_CACHE_SECONDS: return self._cached_temp junction = None edge = None fallback_max = 0.0 for hwmon_path in self._hwmon_paths: try: for entry in sorted(os.listdir(hwmon_path)): if not (entry.startswith("temp") and entry.endswith("_input")): continue temp_file = os.path.join(hwmon_path, entry) label_file = temp_file.replace("_input", "_label") try: with open(temp_file, 'r') as f: raw = int(f.read().strip()) except (ValueError, OSError): continue celsius = raw / 1000.0 if not (20 <= celsius <= 120): continue if celsius > fallback_max: fallback_max = celsius label = "" if os.path.exists(label_file): try: with open(label_file, 'r') as f: label = f.read().strip().lower() except OSError: pass if label == "junction" and (junction is None or celsius > junction): junction = celsius elif label == "edge" and (edge is None or celsius > edge): edge = celsius except OSError as e: logger.debug(f"Linux AMD: hwmon listdir failed for {hwmon_path}: {e}") temp = junction if junction is not None else (edge if edge is not None else fallback_max) self._cached_temp = temp self._temp_cache_timestamp = now return temp def _get_fan_speed(self) -> int: """Read fan duty cycle (%) from amdgpu hwmon. Strategy chain — first hit wins, each card path is tried in order: 1. ``pwm1`` / ``pwm1_max`` — raw PWM duty (legacy pre-6.x driver, also some aftermarket drivers). 2. ``fan1_target`` / ``fan1_max`` — **firmware-intended RPM** / max RPM. This is the most useful value on RDNA3 (Navi 32, RX 7700/7800 XT) when the kernel driver has *stopped* the fan at idle: ``fan1_input`` reports 0 RPM (the fan really is stopped) but ``fan1_target`` shows what the firmware would command the moment temperature rises. Reporting the target avoids the "stuck at 0%" false reading. 3. ``fan1_input`` / ``fan1_max`` — current RPM scaled to %. Pre-RDNA3 fallback; on modern driver the input is 0 in zero-RPM idle and this returns 0. 4. ``fan1_input`` permille (raw // 10) — older drivers that report duty as 0–1000 instead of RPM. Result is clamped to [0, 100] and cached for 5s like temperature to keep sysfs reads light. Card without fan control (``pwm1_enable == "0"`` AND no ``fan1_*`` files) returns 0 — passive / fanless designs shouldn't pretend to have a fan. """ now = time.time() # Independent cache window — fan can change much faster than # temp during a workload ramp, so sharing temp's 5s window made # the overlay look "stuck" at the boot-time reading. cached = getattr(self, "_cached_fan", None) if ( cached is not None and now - self._fan_cache_timestamp < self._FAN_CACHE_SECONDS ): return cached fan_pct = 0 for hwmon_path in self._hwmon_paths: # If the driver exposes pwm1_enable == 0 (manual off) AND no # fan1_* files, the card is fanless — skip the whole path. enable_file = os.path.join(hwmon_path, "pwm1_enable") has_pwm1 = os.path.exists(os.path.join(hwmon_path, "pwm1")) has_fan1 = os.path.exists(os.path.join(hwmon_path, "fan1_input")) if not (has_pwm1 or has_fan1): continue pwm_disabled = False if os.path.exists(enable_file): try: with open(enable_file, 'r') as f: if f.read().strip() == "0": pwm_disabled = True except OSError: pass # === Strategy 1: pwm1 / pwm1_max (raw duty, legacy) === logger.debug(f"FAN: hwmon={hwmon_path} has_pwm1={has_pwm1} pwm_disabled={pwm_disabled} pwm1_enable_file={os.path.exists(enable_file)}") if not pwm_disabled and has_pwm1: pwm_file = os.path.join(hwmon_path, "pwm1") pwm_max_file = os.path.join(hwmon_path, "pwm1_max") logger.debug(f"FAN: Strategy1 checking {pwm_file} exists={os.path.exists(pwm_file)} max_exists={os.path.exists(pwm_max_file)}") if os.path.exists(pwm_max_file): try: with open(pwm_file, 'r') as f: pwm = int(f.read().strip()) with open(pwm_max_file, 'r') as f: pwm_max = int(f.read().strip()) logger.debug(f"FAN: Strategy1 pwm={pwm} pwm_max={pwm_max}") if pwm_max > 0: pct = int(round(pwm * 100.0 / pwm_max)) logger.debug(f"FAN: Strategy1 pct={pct}") fan_pct = max(fan_pct, pct) if pct > 0: # Got a real reading — stop walking backends. break except (ValueError, OSError) as e: logger.debug(f"FAN: Strategy1 exception: {e}") # === Strategy 2: fan1_target / fan1_max (firmware intent) === # On modern RDNA3 drivers, this is the *only* way to see fan # activity before the fan actually spins up — fan1_input stays # at 0 RPM in zero-RPM idle, so a naive implementation reports # "0%" the whole time. fan1_target reflects what the firmware # would command once temperature crosses the ramp threshold. target_file = os.path.join(hwmon_path, "fan1_target") max_file = os.path.join(hwmon_path, "fan1_max") if os.path.exists(target_file) and os.path.exists(max_file): try: with open(target_file, 'r') as f: target = int(f.read().strip()) with open(max_file, 'r') as f: fan_max = int(f.read().strip()) if fan_max > 0 and target > 0: pct = int(round(target * 100.0 / fan_max)) fan_pct = max(fan_pct, pct) # Keep walking — Strategy 3 may still report a # higher RPM-derived value while fan is spinning. except (ValueError, OSError): pass # === Strategy 3: fan1_input / fan1_max (current RPM scaled) === # Modern RDNA3 only — older drivers used a different file. if has_fan1 and os.path.exists(max_file): try: with open(os.path.join(hwmon_path, "fan1_input"), 'r') as f: rpm = int(f.read().strip()) if rpm > 0: with open(max_file, 'r') as f: fan_max = int(f.read().strip()) if fan_max > 0: fan_pct = max(fan_pct, int(round(rpm * 100.0 / fan_max))) except (ValueError, OSError): pass # === Strategy 4: fan1_input permille (legacy 0–1000 driver) === # Older out-of-tree drivers reported fan duty as 0–1000 # permille instead of RPM. Modern drivers report RPM (often # > 1000 even at low speeds) and were already handled by # Strategy 3; here we only act on values that *can't* be # RPM. Upper bound 1000 means "if RPM is reported, it would # be at least 1001" — false on most cards but cheap to guard. if has_fan1: try: with open(os.path.join(hwmon_path, "fan1_input"), 'r') as f: raw = int(f.read().strip()) # permille on legacy drivers: 778 = 77.8% duty. if 11 < raw <= 1000 and raw != int(open(os.path.join(hwmon_path, "fan1_max")).read().strip()): # Last guard: skip if this value happens to equal # fan1_max (some quirky firmware reports max as # input when at full tilt). fan_pct = max(fan_pct, raw // 10) except (ValueError, OSError): pass if fan_pct > 0: break fan_pct = max(0, min(100, fan_pct)) self._cached_fan = fan_pct self._fan_cache_timestamp = time.time() return fan_pct 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