Coverage for orchestrant/monitoring/gpu_amd.py: 53%
380 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-19 11:22 +0000
1"""AMD GPU probing: ADL PMLog on Windows, amdgpu sysfs on Linux."""
3from __future__ import annotations
5import ctypes
6import importlib
7import sys
8from contextlib import suppress
9from dataclasses import dataclass
10from functools import lru_cache
11from pathlib import Path
12from typing import TYPE_CHECKING, Self
14from loguru import logger
16from orchestrant.monitoring.gpu_types import GPUSnapshot
19if TYPE_CHECKING:
20 from types import ModuleType, TracebackType
23# The ADL DLL load is deferred to AmdGpuProbe so importing never touches the driver.
24AMD_AVAILABLE = bool(
25 sys.platform == "win32"
26 or (sys.platform.startswith("linux") and Path("/sys/class/drm").is_dir())
27)
30@lru_cache(maxsize=1)
31def _load_amdsmi() -> ModuleType | None:
32 """Import AMD SMI on first use (without libamd_smi.so it raises and prints), or None."""
33 try:
34 module = importlib.import_module("amdsmi")
35 except Exception: # pragma: no cover - environment-dependent
36 return None
37 return module
40def _read_text(path: Path) -> str | None:
41 """Read a small sysfs file, returning None when it cannot be read."""
42 try:
43 return path.read_text(encoding="utf-8").strip()
44 except (OSError, ValueError) as exc:
45 logger.debug("Cannot read {}: {}", path, exc)
46 return None
49def _read_int(path: Path) -> int | None:
50 """Read an integer from a sysfs file, returning None on any failure."""
51 raw = _read_text(path)
52 if raw is None:
53 return None
54 try:
55 return int(raw)
56 except ValueError:
57 logger.debug("Not an integer in {}: {!r}", path, raw)
58 return None
61def _read_float(path: Path) -> float | None:
62 """Read a float from a sysfs file, returning None on any failure."""
63 raw = _read_text(path)
64 if raw is None:
65 return None
66 try:
67 return float(raw)
68 except ValueError:
69 logger.debug("Not a number in {}: {!r}", path, raw)
70 return None
73def _first_sensor(sensors: dict[int, int], ids: tuple[int, ...]) -> int | None:
74 """First present PMLog sensor among ``ids`` (priority order)."""
75 for sensor_id in ids:
76 if sensor_id in sensors:
77 return sensors[sensor_id]
78 return None
81def _amdsmi_drm_card(smi: ModuleType, handle: object) -> int | None:
82 """DRM card index AMD SMI reports for a processor handle, if any."""
83 try:
84 info = smi.amdsmi_get_gpu_enumeration_info(handle) or {}
85 except Exception as exc:
86 logger.debug("amdsmi enumeration info failed: {}", exc)
87 return None
88 value = info.get("drm_card")
89 return value if isinstance(value, int) else None
92def _amdsmi_product_name(card_index: int) -> str | None:
93 """Marketing name of an amdgpu card via optional AMD SMI; sysfs has only PCI IDs."""
94 smi = _load_amdsmi()
95 if smi is None:
96 return None
97 try:
98 smi.amdsmi_init()
99 except Exception as exc:
100 logger.debug("amdsmi init failed: {}", exc)
101 return None
102 try:
103 handles = smi.amdsmi_get_processor_handles()
104 if not handles:
105 return None
106 handle = None
107 for candidate in handles:
108 if _amdsmi_drm_card(smi, candidate) == card_index:
109 handle = candidate
110 break
111 if handle is None and 0 <= card_index < len(handles):
112 handle = handles[card_index]
113 if handle is None:
114 return None
115 board = smi.amdsmi_get_gpu_board_info(handle) or {}
116 name = board.get("product_name") or board.get("market_name")
117 return str(name).strip() or None
118 except Exception as exc:
119 logger.debug("amdsmi name lookup failed: {}", exc)
120 return None
121 finally:
122 with suppress(Exception):
123 smi.amdsmi_shut_down()
126class _SysfsBackend:
127 """Linux amdgpu backend over ``/sys/class/drm/card*/device``."""
129 backend = "sysfs"
131 def __init__(self, gpu_index: int = 0, root: Path | None = None) -> None:
132 """Select the ``gpu_index``-th AMD card below ``root``, largest VRAM first."""
133 self.available = False
134 self.name = "N/A"
135 self.memory_type = ""
136 self.memory_bandwidth_mbps = 0
137 self.memory_total_bytes = 0
138 self.handle: object | None = None
139 self._device: Path | None = None
140 self._hwmon: Path | None = None
142 cards = self._discover(root or Path("/sys/class/drm"))
143 if not 0 <= gpu_index < len(cards):
144 logger.debug(
145 "AMD GPU index {} out of range ({} card(s) found)",
146 gpu_index,
147 len(cards),
148 )
149 return
151 device = cards[gpu_index]
152 self._device = device
153 self._hwmon = self._find_hwmon(device)
154 self.memory_total_bytes = _read_int(device / "mem_info_vram_total") or 0
155 card_name = device.parent.name
156 card_number = card_name.removeprefix("card")
157 self.name = (
158 (_amdsmi_product_name(int(card_number)) if card_number.isdigit() else None)
159 or _read_text(device / "product_name")
160 or f"AMD GPU ({card_name})"
161 )
162 self.handle = device
163 self.available = True
164 logger.success("AMD GPU monitoring initialized: {} (sysfs)", self.name)
166 @staticmethod
167 def _discover(root: Path) -> list[Path]:
168 """AMD device directories under drm, largest VRAM first."""
169 cards: list[tuple[int, int, Path]] = []
170 for device in root.glob("card[0-9]*/device"):
171 vendor = (_read_text(device / "vendor") or "").lower()
172 if vendor != "0x1002":
173 continue
174 card_name = device.parent.name
175 card_number = card_name.removeprefix("card")
176 if not card_number.isdigit():
177 continue
178 vram = _read_int(device / "mem_info_vram_total") or 0
179 cards.append((-vram, int(card_number), device))
180 cards.sort()
181 return [device for _vram, _number, device in cards]
183 @staticmethod
184 def _find_hwmon(device: Path) -> Path | None:
185 """The ``amdgpu`` hwmon directory, or the first hwmon that has temp."""
186 fallback = None
187 for candidate in sorted((device / "hwmon").glob("hwmon*")):
188 if (_read_text(candidate / "name") or "") == "amdgpu":
189 return candidate
190 if fallback is None and (candidate / "temp1_input").exists():
191 fallback = candidate
192 return fallback
194 def read(self) -> GPUSnapshot | None:
195 """One sysfs read of utilization, VRAM, temperature and power."""
196 if not self.available or self._device is None:
197 return None
198 device = self._device
199 temperature = 0.0
200 power = 0.0
201 if self._hwmon is not None:
202 temp_raw = _read_float(self._hwmon / "temp1_input")
203 temperature = (temp_raw or 0.0) / 1000.0
204 power_raw = _read_float(self._hwmon / "power1_average")
205 if power_raw is None:
206 power_raw = _read_float(self._hwmon / "power1_input")
207 power = (power_raw or 0.0) / 1_000_000.0
208 return GPUSnapshot(
209 utilization=float(_read_int(device / "gpu_busy_percent") or 0),
210 memory_used_bytes=_read_int(device / "mem_info_vram_used") or 0,
211 memory_total_bytes=self.memory_total_bytes,
212 temperature_celsius=temperature,
213 power_watts=power,
214 vendor="amd",
215 )
217 def shutdown(self) -> None:
218 """Nothing to release: sysfs reads hold no resources."""
219 self.available = False
222# Windows: AMD Display Library (ADL)
224_ADL_MAX_PATH = 256
225_ADL_OK = 0
226# ADL reports the PCI vendor ID in decimal (1002), not as 0x1002.
227_ADL_VENDOR_AMD = 1002
228_PMLOG_MAX_SENSORS = 256
230# ADL_PMLOG_SENSORS IDs in preference order: the first supported sensor wins.
231_PMLOG_TEMP_IDS = (8, 27, 28, 29, 9) # EDGE, HOTSPOT, GFX, SOC, MEM
232_PMLOG_POWER_IDS = (23, 73, 30, 17) # ASIC, BOARD, GFX, SOC
233_PMLOG_ACTIVITY_GFX = 19
236class _AdlUnavailableError(RuntimeError):
237 """Raised when the ADL library or its context cannot be brought up."""
240class _AdlAdapterInfo(ctypes.Structure):
241 """Modern ``AdapterInfo`` layout (1572 bytes on x64); pyadl's smaller one fails with -3."""
243 _fields_ = [
244 ("iSize", ctypes.c_int),
245 ("iAdapterIndex", ctypes.c_int),
246 ("strUDID", ctypes.c_char * _ADL_MAX_PATH),
247 ("iBusNumber", ctypes.c_int),
248 ("iDeviceNumber", ctypes.c_int),
249 ("iFunctionNumber", ctypes.c_int),
250 ("iVendorID", ctypes.c_int),
251 ("strAdapterName", ctypes.c_char * _ADL_MAX_PATH),
252 ("strDisplayName", ctypes.c_char * _ADL_MAX_PATH),
253 ("iPresent", ctypes.c_int),
254 ("iExist", ctypes.c_int),
255 ("strDriverPath", ctypes.c_char * _ADL_MAX_PATH),
256 ("strDriverPathExt", ctypes.c_char * _ADL_MAX_PATH),
257 ("strPNPString", ctypes.c_char * _ADL_MAX_PATH),
258 ("iOSDisplayIndex", ctypes.c_int),
259 ]
262class _AdlMemoryInfo(ctypes.Structure):
263 """``ADL2_Adapter_MemoryInfo_Get`` payload; padding guards newer fields."""
265 _fields_ = [
266 ("iMemorySize", ctypes.c_longlong),
267 ("strMemoryType", ctypes.c_char * _ADL_MAX_PATH),
268 ("iMemoryBandwidth", ctypes.c_longlong),
269 ("reserved", ctypes.c_char * 512),
270 ]
273class _AdlSingleSensor(ctypes.Structure):
274 """One PMLog sensor: a supported flag plus its current value."""
276 _fields_ = [("supported", ctypes.c_int), ("value", ctypes.c_int)]
279class _AdlPMLogData(ctypes.Structure):
280 """Output of ``ADL2_New_QueryPMLogData_Get`` (256 sensor slots)."""
282 _fields_ = [
283 ("iSize", ctypes.c_int),
284 ("sensors", _AdlSingleSensor * _PMLOG_MAX_SENSORS),
285 ]
288@dataclass(frozen=True)
289class _AdlAdapter:
290 """One deduplicated AMD display adapter as ADL sees it."""
292 index: int
293 name: str
294 pnp_id: str
295 memory_total_bytes: int
296 memory_type: str
297 memory_bandwidth_mbps: int
300def _decode(raw: bytes) -> str:
301 """Decode a fixed-size ADL C string, trimming at the first NUL."""
302 return raw.split(b"\x00", 1)[0].decode("utf-8", errors="replace").strip()
305class _AdlApi:
306 """ctypes binding over the ADL DLL; tests substitute its three public methods."""
308 _MALLOC_CB = ctypes.CFUNCTYPE(ctypes.c_void_p, ctypes.c_int)
310 # Annotation only: a linux-targeting checker sees __init__'s binding as unreachable.
311 _lib: ctypes.CDLL
312 _msvcrt: ctypes.CDLL
314 def __init__(self) -> None:
315 """Load ``atiadlxx.dll`` and open an ADL2 context."""
316 if sys.platform != "win32":
317 msg = "ADL is only available on Windows"
318 raise _AdlUnavailableError(msg)
319 self._lib = self._load_dll()
320 self._msvcrt = ctypes.CDLL("msvcrt.dll")
321 self._malloc = self._MALLOC_CB(self._alloc)
322 self._ctx: ctypes.c_void_p | None = ctypes.c_void_p()
323 self._bind_required()
324 self._bind_optional()
325 if self._create(self._malloc, 1, ctypes.byref(self._ctx)) != _ADL_OK:
326 msg = "ADL2_Main_Control_Create failed"
327 raise _AdlUnavailableError(msg)
329 def _alloc(self, size: int) -> int | None:
330 """Allocator callback ADL calls for its internal buffers."""
331 return self._msvcrt.malloc(size)
333 @staticmethod
334 def _load_dll() -> ctypes.CDLL:
335 """Load the 64-bit ADL DLL, falling back to the 32-bit one."""
336 last_error: OSError | None = None
337 for dll in ("atiadlxx.dll", "atiadlxy.dll"):
338 try:
339 return ctypes.CDLL(dll)
340 except OSError as exc:
341 last_error = exc
342 msg = f"ADL library not loadable: {last_error}"
343 raise _AdlUnavailableError(msg)
345 def _bind_required(self) -> None:
346 """Bind the functions every ADL version exposes."""
347 lib = self._lib
348 self._create = lib.ADL2_Main_Control_Create
349 self._create.restype = ctypes.c_int
350 self._create.argtypes = [
351 self._MALLOC_CB,
352 ctypes.c_int,
353 ctypes.POINTER(ctypes.c_void_p),
354 ]
355 self._destroy = lib.ADL2_Main_Control_Destroy
356 self._destroy.restype = ctypes.c_int
357 self._destroy.argtypes = [ctypes.c_void_p]
358 self._number = lib.ADL2_Adapter_NumberOfAdapters_Get
359 self._number.restype = ctypes.c_int
360 self._number.argtypes = [ctypes.c_void_p, ctypes.POINTER(ctypes.c_int)]
361 self._adapter_info = lib.ADL2_Adapter_AdapterInfo_Get
362 self._adapter_info.restype = ctypes.c_int
363 self._adapter_info.argtypes = [
364 ctypes.c_void_p,
365 ctypes.POINTER(_AdlAdapterInfo),
366 ctypes.c_int,
367 ]
368 self._adapter_id = lib.ADL2_Adapter_ID_Get
369 self._adapter_id.restype = ctypes.c_int
370 self._adapter_id.argtypes = [
371 ctypes.c_void_p,
372 ctypes.c_int,
373 ctypes.POINTER(ctypes.c_int),
374 ]
375 self._query_pmlog = lib.ADL2_New_QueryPMLogData_Get
376 self._query_pmlog.restype = ctypes.c_int
377 self._query_pmlog.argtypes = [
378 ctypes.c_void_p,
379 ctypes.c_int,
380 ctypes.POINTER(_AdlPMLogData),
381 ]
383 def _bind_optional(self) -> None:
384 """Bind functions older drivers may not export."""
385 try:
386 self._memory_info = self._lib.ADL2_Adapter_MemoryInfo_Get
387 self._memory_info.restype = ctypes.c_int
388 self._memory_info.argtypes = [
389 ctypes.c_void_p,
390 ctypes.c_int,
391 ctypes.POINTER(_AdlMemoryInfo),
392 ]
393 except AttributeError:
394 self._memory_info = None
395 try:
396 self._vram_usage = self._lib.ADL2_Adapter_DedicatedVRAMUsage_Get
397 self._vram_usage.restype = ctypes.c_int
398 self._vram_usage.argtypes = [
399 ctypes.c_void_p,
400 ctypes.c_int,
401 ctypes.POINTER(ctypes.c_int),
402 ]
403 except AttributeError:
404 self._vram_usage = None
406 def list_adapters(self) -> list[_AdlAdapter]:
407 """AMD adapters, deduplicated by adapter ID and with VRAM size."""
408 context = self._ctx
409 if context is None:
410 return []
411 number = ctypes.c_int(0)
412 if self._number(context, ctypes.byref(number)) != _ADL_OK or number.value <= 0:
413 return []
414 entries = (_AdlAdapterInfo * number.value)()
415 if (
416 self._adapter_info(
417 context,
418 ctypes.cast(entries, ctypes.POINTER(_AdlAdapterInfo)),
419 ctypes.sizeof(entries),
420 )
421 != _ADL_OK
422 ):
423 return []
424 adapters: list[_AdlAdapter] = []
425 seen: set[int] = set()
426 for entry in entries:
427 if entry.iVendorID != _ADL_VENDOR_AMD:
428 continue
429 adapter_id = ctypes.c_int(0)
430 if (
431 self._adapter_id(context, entry.iAdapterIndex, ctypes.byref(adapter_id))
432 != _ADL_OK
433 ):
434 continue
435 if adapter_id.value in seen:
436 continue
437 seen.add(adapter_id.value)
438 memory_total = 0
439 memory_type = ""
440 bandwidth = 0
441 memory = _AdlMemoryInfo()
442 if (
443 self._memory_info is not None
444 and self._memory_info(
445 context, entry.iAdapterIndex, ctypes.byref(memory)
446 )
447 == _ADL_OK
448 ):
449 memory_total = int(memory.iMemorySize)
450 memory_type = _decode(memory.strMemoryType)
451 bandwidth = int(memory.iMemoryBandwidth)
452 adapters.append(
453 _AdlAdapter(
454 index=entry.iAdapterIndex,
455 name=_decode(entry.strAdapterName)
456 or f"AMD GPU {entry.iAdapterIndex}",
457 pnp_id=_decode(entry.strPNPString),
458 memory_total_bytes=memory_total,
459 memory_type=memory_type,
460 memory_bandwidth_mbps=bandwidth,
461 )
462 )
463 return adapters
465 def read_sensors(self, adapter_index: int) -> dict[int, int]:
466 """Current PMLog sensor values, keyed by ``ADL_PMLOG_*`` ID."""
467 context = self._ctx
468 if context is None:
469 return {}
470 data = _AdlPMLogData()
471 data.iSize = ctypes.sizeof(data)
472 if self._query_pmlog(context, adapter_index, ctypes.byref(data)) != _ADL_OK:
473 return {}
474 return {i: s.value for i, s in enumerate(data.sensors) if s.supported}
476 def vram_used_mb(self, adapter_index: int) -> int | None:
477 """Dedicated VRAM in use, in MiB, when the driver reports it."""
478 context = self._ctx
479 if context is None or self._vram_usage is None:
480 return None
481 usage = ctypes.c_int(-1)
482 if (
483 self._vram_usage(context, adapter_index, ctypes.byref(usage)) != _ADL_OK
484 or usage.value < 0
485 ):
486 return None
487 return int(usage.value)
489 def close(self) -> None:
490 """Destroy the ADL context. Safe to call more than once."""
491 if self._ctx is None:
492 return
493 with suppress(Exception):
494 self._destroy(self._ctx)
495 self._ctx = None
498class _AdlBackend:
499 """Windows AMD backend over the ADL PMLog sensors."""
501 backend = "adl"
503 def __init__(self, gpu_index: int = 0, api: _AdlApi | None = None) -> None:
504 """Open ADL and select the AMD adapter with the most VRAM."""
505 self.available = False
506 self.name = "N/A"
507 self.memory_type = ""
508 self.memory_bandwidth_mbps = 0
509 self.memory_total_bytes = 0
510 self.handle: object | None = None
511 self._api: _AdlApi | None = None
512 self._index = -1
514 try:
515 adl = api if api is not None else _AdlApi()
516 except Exception as exc:
517 logger.debug("AMD ADL backend unavailable: {}", exc)
518 return
519 try:
520 adapters = sorted(
521 adl.list_adapters(),
522 key=lambda adapter: (-adapter.memory_total_bytes, adapter.index),
523 )
524 except Exception as exc:
525 logger.debug("AMD adapter enumeration failed: {}", exc)
526 adl.close()
527 return
528 if not adapters:
529 adl.close()
530 logger.debug("No AMD display adapter reported by ADL")
531 return
532 if not 0 <= gpu_index < len(adapters):
533 adl.close()
534 logger.debug(
535 "AMD GPU index {} out of range ({} adapter(s) found)",
536 gpu_index,
537 len(adapters),
538 )
539 return
540 chosen = adapters[gpu_index]
541 self._api = adl
542 self._index = chosen.index
543 self.name = chosen.name
544 self.memory_type = chosen.memory_type
545 self.memory_bandwidth_mbps = chosen.memory_bandwidth_mbps
546 self.memory_total_bytes = chosen.memory_total_bytes
547 self.handle = chosen.index
548 self.available = True
549 logger.success("AMD GPU monitoring initialized: {} (ADL)", self.name)
551 def read(self) -> GPUSnapshot | None:
552 """One ADL read of utilization, VRAM, temperature and power."""
553 if not self.available or self._api is None:
554 return None
555 sensors = self._api.read_sensors(self._index)
556 if not sensors:
557 return None
558 used_mb = self._api.vram_used_mb(self._index)
559 return GPUSnapshot(
560 utilization=float(sensors.get(_PMLOG_ACTIVITY_GFX, 0)),
561 memory_used_bytes=(used_mb or 0) * 1024**2,
562 memory_total_bytes=self.memory_total_bytes,
563 temperature_celsius=float(_first_sensor(sensors, _PMLOG_TEMP_IDS) or 0),
564 power_watts=float(_first_sensor(sensors, _PMLOG_POWER_IDS) or 0),
565 vendor="amd",
566 )
568 def shutdown(self) -> None:
569 """Destroy the ADL context and mark the backend unavailable."""
570 if self._api is not None:
571 self._api.close()
572 self._api = None
573 self.available = False
576class AmdGpuProbe:
577 """Vendor-side probe picking the AMD backend for this platform."""
579 def __init__(self, gpu_index: int = 0) -> None:
580 """Construct the platform backend and adopt it when it answers."""
581 self.available = False
582 self.name = "N/A"
583 self.backend = "none"
584 self.memory_type = ""
585 self.memory_bandwidth_mbps = 0
586 self.handle: object | None = None
587 self._backend: _AdlBackend | _SysfsBackend | None = None
589 backend: _AdlBackend | _SysfsBackend | None = None
590 if sys.platform == "win32":
591 backend = _AdlBackend(gpu_index)
592 elif sys.platform.startswith("linux"):
593 backend = _SysfsBackend(gpu_index)
594 if backend is None or not backend.available:
595 return
596 self._backend = backend
597 self.available = True
598 self.name = backend.name
599 self.backend = backend.backend
600 self.memory_type = backend.memory_type
601 self.memory_bandwidth_mbps = backend.memory_bandwidth_mbps
602 self.handle = backend.handle
604 def read(self) -> GPUSnapshot | None:
605 """Read the current GPU metrics, or None when unavailable."""
606 if self._backend is None:
607 return None
608 return self._backend.read()
610 def shutdown(self) -> None:
611 """Release backend resources. Safe to call multiple times."""
612 if self._backend is not None:
613 self._backend.shutdown()
614 self._backend = None
615 self.available = False
617 def __enter__(self) -> Self:
618 """Enter context manager."""
619 return self
621 def __exit__(
622 self,
623 _exc_type: type[BaseException] | None,
624 _exc_val: BaseException | None,
625 _exc_tb: TracebackType | None,
626 ) -> None:
627 """Exit context manager and ensure cleanup."""
628 self.shutdown()