Coverage for openhcs/ui/shared/system_monitor_core.py: 17.1%

1""" 

2System Monitor Core - Framework-agnostic metrics collection. 

3 

4This module provides pure system metrics collection without any visualization dependencies. 

5Can be used by any UI framework (PyQt, Textual, etc.) for system monitoring. 

6""" 

7 

8import platform 

9import psutil 

10import subprocess 

11import time 

12from datetime import datetime 

13from collections import deque 

14from typing import Dict, Any, Optional 

15 

16# Try to import GPU monitoring libraries 

17try: 

18 import GPUtil 

19 GPU_AVAILABLE = True 

20except ImportError: 

21 GPU_AVAILABLE = False 

22 

23 

24def is_wsl(): 

25 """Check if running in Windows Subsystem for Linux.""" 

26 return 'microsoft' in platform.uname().release.lower() 

27 

28 

29def get_cpu_freq_mhz(): 

30 """Get CPU frequency in MHz, with WSL compatibility.""" 

31 if is_wsl(): 

32 try: 

33 output = subprocess.check_output( 

34 ['powershell.exe', '-Command', 

35 'Get-CimInstance -ClassName Win32_Processor | Select-Object -ExpandProperty CurrentClockSpeed'], 

36 stderr=subprocess.DEVNULL 

37 ) 

38 return int(output.strip()) 

39 except Exception: 

40 return 0 

41 try: 

42 freq = psutil.cpu_freq() 

43 return int(freq.current) if freq else 0 

44 except Exception: 

45 return 0 

46 

47 

48class SystemMonitorCore: 

49 """ 

50 Framework-agnostic system monitoring core. 

51  

52 Collects CPU, RAM, GPU, and VRAM metrics without any visualization dependencies. 

53 Maintains historical data in deques for efficient time-series tracking. 

54 """ 

55 

56 def __init__(self, history_length: int = 60): 

57 """ 

58 Initialize the system monitor core. 

59  

60 Args: 

61 history_length: Number of historical data points to keep 

62 """ 

63 self.history_length = history_length 

64 

65 # Initialize data storage 

66 self.cpu_history = deque(maxlen=history_length) 

67 self.ram_history = deque(maxlen=history_length) 

68 self.gpu_history = deque(maxlen=history_length) 

69 self.vram_history = deque(maxlen=history_length) 

70 self.time_stamps = deque(maxlen=history_length) 

71 

72 # Cache current metrics to avoid duplicate system calls 

73 self._current_metrics = {} 

74 

75 # Initialize with zeros 

76 for _ in range(history_length): 

77 self.cpu_history.append(0) 

78 self.ram_history.append(0) 

79 self.gpu_history.append(0) 

80 self.vram_history.append(0) 

81 self.time_stamps.append(0) 

82 

83 def update_metrics(self) -> None: 

84 """ 

85 Update system metrics and cache current values. 

86  

87 Collects CPU, RAM, GPU, and VRAM usage and appends to history. 

88 Updates internal cache for efficient access via get_metrics_dict(). 

89 """ 

90 # CPU usage 

91 cpu_percent = psutil.cpu_percent(interval=None) 

92 self.cpu_history.append(cpu_percent) 

93 

94 # RAM usage 

95 ram = psutil.virtual_memory() 

96 ram_percent = ram.percent 

97 self.ram_history.append(ram_percent) 

98 

99 # Cache current metrics to avoid duplicate calls in get_metrics_dict() 

100 self._current_metrics = { 

101 'cpu_percent': cpu_percent, 

102 'ram_percent': ram_percent, 

103 'ram_used_gb': ram.used / (1024**3), 

104 'ram_total_gb': ram.total / (1024**3), 

105 'cpu_cores': psutil.cpu_count(), 

106 'cpu_freq_mhz': get_cpu_freq_mhz(), 

107 } 

108 

109 # GPU usage (if available) 

110 if GPU_AVAILABLE: 

111 try: 

112 gpus = GPUtil.getGPUs() 

113 if gpus: 

114 gpu = gpus[0] # Use first GPU 

115 gpu_load = gpu.load * 100 

116 vram_util = gpu.memoryUtil * 100 

117 self.gpu_history.append(gpu_load) 

118 self.vram_history.append(vram_util) 

119 

120 # Cache GPU metrics 

121 self._current_metrics.update({ 

122 'gpu_percent': gpu_load, 

123 'vram_percent': vram_util, 

124 'gpu_name': gpu.name, 

125 'gpu_temp': gpu.temperature, 

126 'vram_used_mb': gpu.memoryUsed, 

127 'vram_total_mb': gpu.memoryTotal, 

128 }) 

129 else: 

130 self.gpu_history.append(0) 

131 self.vram_history.append(0) 

132 except: 

133 self.gpu_history.append(0) 

134 self.vram_history.append(0) 

135 else: 

136 self.gpu_history.append(0) 

137 self.vram_history.append(0) 

138 

139 # Update timestamps 

140 self.time_stamps.append(time.time()) 

141 

142 def get_metrics_dict(self) -> Dict[str, Any]: 

143 """ 

144 Get current metrics as a dictionary. 

145  

146 Uses cached data from update_metrics() to avoid duplicate system calls. 

147  

148 Returns: 

149 Dictionary containing current system metrics 

150 """ 

151 # Return cached metrics to avoid duplicate system calls 

152 # If no cached data exists (first call), return defaults 

153 if not self._current_metrics: 

154 return { 

155 'cpu_percent': 0, 

156 'ram_percent': 0, 

157 'ram_used_gb': 0, 

158 'ram_total_gb': 0, 

159 'cpu_cores': 0, 

160 'cpu_freq_mhz': 0, 

161 } 

162 

163 return self._current_metrics.copy() 

164 

165 def get_system_info(self) -> Dict[str, Any]: 

166 """ 

167 Get static system information. 

168  

169 Returns: 

170 Dictionary containing system information (OS, CPU, RAM, GPU) 

171 """ 

172 info = { 

173 'os': platform.system(), 

174 'os_version': platform.version(), 

175 'cpu_cores': psutil.cpu_count(), 

176 'ram_total_gb': psutil.virtual_memory().total / (1024**3), 

177 } 

178 

179 # Add GPU info if available 

180 if GPU_AVAILABLE: 

181 try: 

182 gpus = GPUtil.getGPUs() 

183 if gpus: 

184 gpu = gpus[0] 

185 info['gpu_name'] = gpu.name 

186 info['vram_total_mb'] = gpu.memoryTotal 

187 except: 

188 pass 

189 

190 return info 

191 

192 def reset_history(self) -> None: 

193 """Reset all historical data to zeros.""" 

194 self.cpu_history.clear() 

195 self.ram_history.clear() 

196 self.gpu_history.clear() 

197 self.vram_history.clear() 

198 self.time_stamps.clear() 

199 

200 # Re-initialize with zeros 

201 for _ in range(self.history_length): 

202 self.cpu_history.append(0) 

203 self.ram_history.append(0) 

204 self.gpu_history.append(0) 

205 self.vram_history.append(0) 

206 self.time_stamps.append(0) 

207 

208 self._current_metrics = {} 

209