Coverage for orchestrant/monitoring/snapshot.py: 100%

1"""Single-read system snapshot shared by both SystemMonitor variants. 

2 

3Two monitors consume this: :class:`orchestrant.monitoring.system.SystemMonitor` 

4(a time-series logger with start/record/summary) and 

5:class:`orchestrant.pipeline.monitoring.system.SystemMonitor` (a live 

6per-frame snapshotter for pipeline UIs). They shape the data differently but 

7must read it identically — this module owns that read. 

8""" 

9 

10from __future__ import annotations 

11 

12from dataclasses import dataclass 

13from typing import TYPE_CHECKING, Any 

14 

15import psutil 

16 

17 

18if TYPE_CHECKING: 

19 from orchestrant.monitoring.gpu import GPUProbe, GPUSnapshot 

20 

21 

22@dataclass(frozen=True) 

23class RawSystemSnapshot: 

24 """One synchronized read of CPU, RAM, and (optionally) GPU state.""" 

25 

26 cpu_percent: float 

27 ram: Any # psutil svmem namedtuple (total/used/available/percent) 

28 gpu: GPUSnapshot | None 

29 

30 

31def read_raw_snapshot(gpu_probe: GPUProbe) -> RawSystemSnapshot: 

32 """Read CPU utilization, virtual memory, and the GPU probe in one pass. 

33 

34 ``psutil.cpu_percent(interval=None)`` is non-blocking and relative to the 

35 previous call, so callers must have primed it once at construction time. 

36 """ 

37 return RawSystemSnapshot( 

38 cpu_percent=psutil.cpu_percent(interval=None), 

39 ram=psutil.virtual_memory(), 

40 gpu=gpu_probe.read(), 

41 )