Coverage for openhcs/textual_tui/services/system_monitor.py: 0.0%

1""" 

2System Monitor Core Module 

3Real-time system monitoring with plotext visualization 

4""" 

5 

6import plotext as plt 

7import platform 

8import psutil 

9import subprocess 

10import time 

11from datetime import datetime 

12from collections import deque 

13import io 

14import sys 

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 SystemMonitor: 

49 """Core system monitoring class with plotext visualization""" 

50 

51 def __init__(self, history_length=60): 

52 self.history_length = history_length 

53 

54 # Initialize data storage 

55 self.cpu_history = deque(maxlen=history_length) 

56 self.ram_history = deque(maxlen=history_length) 

57 self.gpu_history = deque(maxlen=history_length) 

58 self.vram_history = deque(maxlen=history_length) 

59 self.time_stamps = deque(maxlen=history_length) 

60 

61 # Cache current metrics to avoid duplicate system calls 

62 self._current_metrics = {} 

63 

64 # Initialize with zeros 

65 for _ in range(history_length): 

66 self.cpu_history.append(0) 

67 self.ram_history.append(0) 

68 self.gpu_history.append(0) 

69 self.vram_history.append(0) 

70 self.time_stamps.append(0) 

71 

72 def update_metrics(self): 

73 """Update system metrics and cache current values""" 

74 # CPU usage 

75 cpu_percent = psutil.cpu_percent(interval=None) 

76 self.cpu_history.append(cpu_percent) 

77 

78 # RAM usage 

79 ram = psutil.virtual_memory() 

80 ram_percent = ram.percent 

81 self.ram_history.append(ram_percent) 

82 

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

84 self._current_metrics = { 

85 'cpu_percent': cpu_percent, 

86 'ram_percent': ram_percent, 

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

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

89 'cpu_cores': psutil.cpu_count(), 

90 'cpu_freq_mhz': get_cpu_freq_mhz(), 

91 } 

92 

93 # GPU usage (if available) 

94 if GPU_AVAILABLE: 

95 try: 

96 gpus = GPUtil.getGPUs() 

97 if gpus: 

98 gpu = gpus[0] # Use first GPU 

99 gpu_load = gpu.load * 100 

100 vram_util = gpu.memoryUtil * 100 

101 self.gpu_history.append(gpu_load) 

102 self.vram_history.append(vram_util) 

103 

104 # Cache GPU metrics 

105 self._current_metrics.update({ 

106 'gpu_percent': gpu_load, 

107 'vram_percent': vram_util, 

108 'gpu_name': gpu.name, 

109 'gpu_temp': gpu.temperature, 

110 'vram_used_mb': gpu.memoryUsed, 

111 'vram_total_mb': gpu.memoryTotal, 

112 }) 

113 else: 

114 self.gpu_history.append(0) 

115 self.vram_history.append(0) 

116 except: 

117 self.gpu_history.append(0) 

118 self.vram_history.append(0) 

119 else: 

120 self.gpu_history.append(0) 

121 self.vram_history.append(0) 

122 

123 # Update timestamps 

124 self.time_stamps.append(time.time()) 

125 

126 def get_ascii_title(self): 

127 """Return the OpenHCS ASCII art title""" 

128 return """ 

129╔═══════════════════════════════════════════════════════════════════════╗ 

130║ ██████╗ ██████╗ ███████╗███╗ ██╗██╗ ██╗ ██████╗███████╗ ║ 

131║ ██╔═══██╗██╔══██╗██╔════╝████╗ ██║██║ ██║██╔════╝██╔════╝ ║ 

132║ ██║ ██║██████╔╝█████╗ ██╔██╗ ██║███████║██║ ███████╗ ║ 

133║ ██║ ██║██╔═══╝ ██╔══╝ ██║╚██╗██║██╔══██║██║ ╚════██║ ║ 

134║ ╚██████╔╝██║ ███████╗██║ ╚████║██║ ██║╚██████╗███████║ ║ 

135║ ╚═════╝ ╚═╝ ╚══════╝╚═╝ ╚═══╝╚═╝ ╚═╝ ╚═════╝╚══════╝ ║ 

136╚═══════════════════════════════════════════════════════════════════════╝""" 

137 

138 def create_monitor_view(self, width=80, height=30): 

139 """Create the system monitor visualization""" 

140 plt.clear_figure() 

141 

142 # Create subplots 

143 plt.subplots(2, 2) 

144 plt.plot_size(width=width, height=height) 

145 

146 # Get current metrics for display 

147 current_cpu = self.cpu_history[-1] if self.cpu_history else 0 

148 current_ram = self.ram_history[-1] if self.ram_history else 0 

149 current_gpu = self.gpu_history[-1] if self.gpu_history else 0 

150 current_vram = self.vram_history[-1] if self.vram_history else 0 

151 

152 # Get system info 

153 ram_info = psutil.virtual_memory() 

154 ram_used_gb = ram_info.used / (1024**3) 

155 ram_total_gb = ram_info.total / (1024**3) 

156 

157 x_range = list(range(len(self.cpu_history))) 

158 

159 # CPU Usage Plot (Top Left) 

160 plt.subplot(1, 1) 

161 plt.theme('dark') 

162 plt.plot(x_range, list(self.cpu_history), color='cyan') 

163 plt.ylim(0, 100) 

164 plt.title(f"CPU Usage: {current_cpu:.1f}%") 

165 plt.xlabel("Time (seconds)") 

166 plt.ylabel("Usage %") 

167 

168 # RAM Usage Plot (Top Right) 

169 plt.subplot(1, 2) 

170 plt.theme('dark') 

171 plt.plot(x_range, list(self.ram_history), color='green') 

172 plt.ylim(0, 100) 

173 plt.title(f"RAM: {ram_used_gb:.1f}/{ram_total_gb:.1f} GB ({current_ram:.1f}%)") 

174 plt.xlabel("Time (seconds)") 

175 plt.ylabel("Usage %") 

176 

177 # GPU Usage Plot (Bottom Left) 

178 plt.subplot(2, 1) 

179 plt.theme('dark') 

180 if GPU_AVAILABLE and any(self.gpu_history): 

181 plt.plot(x_range, list(self.gpu_history), color='yellow') 

182 plt.ylim(0, 100) 

183 plt.title(f"GPU Usage: {current_gpu:.1f}%") 

184 else: 

185 plt.plot([0, 1], [0, 0], color='red') 

186 plt.title("GPU: Not Available") 

187 plt.xlabel("Time (seconds)") 

188 plt.ylabel("Usage %") 

189 

190 # VRAM Usage Plot (Bottom Right) 

191 plt.subplot(2, 2) 

192 plt.theme('dark') 

193 if GPU_AVAILABLE and any(self.vram_history): 

194 plt.plot(x_range, list(self.vram_history), color='magenta') 

195 plt.ylim(0, 100) 

196 plt.title(f"VRAM Usage: {current_vram:.1f}%") 

197 else: 

198 plt.plot([0, 1], [0, 0], color='red') 

199 plt.title("VRAM: Not Available") 

200 plt.xlabel("Time (seconds)") 

201 plt.ylabel("Usage %") 

202 

203 # Capture the plot output 

204 old_stdout = sys.stdout 

205 sys.stdout = buffer = io.StringIO() 

206 

207 try: 

208 # Print title 

209 print(self.get_ascii_title()) 

210 

211 # Show plots 

212 plt.show() 

213 

214 # Additional system info 

215 print("\n" + "═" * 75) 

216 print(f"System Information | Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") 

217 print("═" * 75) 

218 print(f"CPU Cores: {psutil.cpu_count()} | CPU Frequency: {get_cpu_freq_mhz()} MHz") 

219 print(f"Total RAM: {ram_total_gb:.1f} GB | Available RAM: {ram_info.available/(1024**3):.1f} GB") 

220 

221 if GPU_AVAILABLE: 

222 try: 

223 gpus = GPUtil.getGPUs() 

224 if gpus: 

225 gpu = gpus[0] 

226 print(f"GPU: {gpu.name} | VRAM: {gpu.memoryUsed:.0f}/{gpu.memoryTotal:.0f} MB") 

227 print(f"GPU Temperature: {gpu.temperature}°C") 

228 except: 

229 pass 

230 

231 # Disk usage 

232 disk = psutil.disk_usage('/') 

233 print(f"Disk Usage: {disk.used/(1024**3):.1f}/{disk.total/(1024**3):.1f} GB ({disk.percent}%)") 

234 

235 # Network info 

236 net = psutil.net_io_counters() 

237 print(f"Network - Sent: {net.bytes_sent/(1024**2):.1f} MB | Recv: {net.bytes_recv/(1024**2):.1f} MB") 

238 print("═" * 75) 

239 

240 output = buffer.getvalue() 

241 finally: 

242 sys.stdout = old_stdout 

243 

244 return output 

245 

246 def get_metrics_dict(self): 

247 """Get current metrics as a dictionary - uses cached data from update_metrics()""" 

248 # Return cached metrics to avoid duplicate system calls 

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

250 if not self._current_metrics: 

251 return { 

252 'cpu_percent': 0, 

253 'ram_percent': 0, 

254 'ram_used_gb': 0, 

255 'ram_total_gb': 0, 

256 'cpu_cores': 0, 

257 'cpu_freq_mhz': 0, 

258 } 

259 

260 return self._current_metrics.copy() 

261 

262 

263# Standalone CLI usage 

264def main(): 

265 """Run system monitor in standalone mode""" 

266 monitor = SystemMonitor() 

267 

268 print("Starting System Monitor...") 

269 print("Press Ctrl+C to exit") 

270 

271 try: 

272 while True: 

273 monitor.update_metrics() 

274 print("\033[2J\033[H") # Clear screen 

275 print(monitor.create_monitor_view()) 

276 time.sleep(1) 

277 except KeyboardInterrupt: 

278 print("\nExiting System Monitor...") 

279 

280 

281if __name__ == "__main__": 

282 main()