Coverage for src/lilbee/providers/fleet/gpu_backends/apple.py: 100%

1"""Apple Metal GPU utilization via ioreg PerformanceStatistics.""" 

2 

3from __future__ import annotations 

4 

5import re 

6 

7from lilbee.providers.fleet.gpu_backends.base import UtilSample, run_smi 

8 

9# llama-server --list-devices emits "MTL0: Apple M1 Pro (21845 MiB, ...)" -> prefix 

10# "MTL". The build-dependent "Metal" variant is also registered in __init__.py. 

11BACKEND_KEY = "MTL" 

12 

13_TOOL = "ioreg" 

14# Root at the GPU accelerator (AGXAccelerator conforms to IOAccelerator on every 

15# Apple Silicon generation) and read its properties. A plain "-d 1 -k 

16# PerformanceStatistics" never reaches the GPU node -- it caps traversal at depth 

17# 1 and matches a shallow always-zero entry, so the bar read 0% even under load. 

18_ARGS = ("-r", "-c", "IOAccelerator", "-d", "1") 

19_TIMEOUT_S = 5.0 

20 

21# Apple Silicon exposes one integrated GPU; its load is "Device Utilization %". 

22_UTIL_RE = re.compile(r'"Device Utilization %"=(\d+)') 

23 

24 

25class AppleBackend: 

26 """Apple Metal util via ioreg; VRAM stays structural via the orchestrator.""" 

27 

28 def sample(self, indices: frozenset[int]) -> dict[int, UtilSample]: 

29 return _ioreg_samples(indices) 

30 

31 

32def _ioreg_output() -> str: 

33 """ioreg PerformanceStatistics stdout, or "" when it can't run.""" 

34 return run_smi(_TOOL, list(_ARGS), _TIMEOUT_S) 

35 

36 

37def _parse_ioreg(raw: str, indices: frozenset[int]) -> dict[int, UtilSample]: 

38 """Apply the single Apple GPU's utilization to every requested index.""" 

39 match = _UTIL_RE.search(raw) 

40 if match is None: 

41 return {} 

42 util = int(match.group(1)) 

43 # Apple has no per-card temperature without sudo, and unified-memory totals 

44 # come from the structural probe, so leave temp/VRAM as the 0 sentinel. 

45 return { 

46 index: UtilSample( 

47 index=index, 

48 utilization_pct=util, 

49 temperature_c=None, 

50 free_bytes=0, 

51 total_bytes=0, 

52 ) 

53 for index in indices 

54 } 

55 

56 

57def _ioreg_samples(indices: frozenset[int]) -> dict[int, UtilSample]: 

58 return _parse_ioreg(_ioreg_output(), indices)