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

1from textual.app import App, ComposeResult 

2from textual.widgets import Static, Header, Footer 

3from textual.containers import Container, Horizontal, Vertical, Grid 

4from typing import Optional 

5from textual import events, work 

6from textual.reactive import reactive 

7from textual.worker import get_current_worker 

8import asyncio 

9import psutil 

10from collections import deque 

11import time 

12 

13# Use textual-plotext for proper Textual integration 

14try: 

15 from textual_plotext import PlotextPlot 

16 PLOTEXT_AVAILABLE = True 

17except ImportError: 

18 PLOTEXT_AVAILABLE = False 

19 

20# Import the SystemMonitor from services 

21from openhcs.textual_tui.services.system_monitor import SystemMonitor 

22 

23try: 

24 import GPUtil 

25 GPU_AVAILABLE = True 

26except ImportError: 

27 GPU_AVAILABLE = False 

28 

29class SystemMonitorTextual(Container): 

30 """A Textual widget for system monitoring using proper Textual components""" 

31 

32 DEFAULT_CSS = """ 

33 SystemMonitorTextual { 

34 background: $background; 

35 color: $text; 

36 } 

37 

38 #top_row { 

39 height: 9; 

40 background: $surface; 

41 } 

42 

43 #header { 

44 width: 1fr; 

45 text-align: left; 

46 color: $accent; 

47 padding: 1; 

48 } 

49 

50 #system_info { 

51 width: 1fr; 

52 color: $text-muted; 

53 text-align: right; 

54 padding: 1; 

55 content-align: center middle; 

56 } 

57 

58 #plots_grid { 

59 height: 1fr; 

60 grid-size: 2 2; 

61 grid-gutter: 1; 

62 } 

63 

64 #fallback_display { 

65 height: 1fr; 

66 text-align: center; 

67 color: $primary; 

68 } 

69 

70 

71 """ 

72 

73 update_interval = reactive(1.0) 

74 is_monitoring = reactive(True) 

75 

76 def __init__(self): 

77 super().__init__() 

78 self.monitor = SystemMonitor() 

79 self.monitor_worker = None 

80 

81 def compose(self) -> ComposeResult: 

82 """Compose the system monitor layout""" 

83 # Top row with title on left and system info on right 

84 with Horizontal(id="top_row"): 

85 yield Static(self._get_ascii_header(), id="header") 

86 yield Static("", id="system_info") 

87 

88 if PLOTEXT_AVAILABLE: 

89 # Use Grid layout for plots 

90 with Grid(id="plots_grid"): 

91 yield PlotextPlot(id="cpu_plot") 

92 yield PlotextPlot(id="ram_plot") 

93 yield PlotextPlot(id="gpu_plot") 

94 yield PlotextPlot(id="vram_plot") 

95 else: 

96 # Fallback to text-based display 

97 yield Static("", id="fallback_display") 

98 

99 async def on_mount(self) -> None: 

100 """Start the monitoring when widget is mounted""" 

101 # Start the worker using the @work decorated method 

102 self.monitor_worker = self._monitor_worker() 

103 self.is_monitoring = True 

104 

105 def on_unmount(self) -> None: 

106 """Stop monitoring and cleanup background thread when widget is unmounted""" 

107 self._stop_monitoring() 

108 

109 def _start_monitoring(self) -> None: 

110 """Start the monitoring Textual worker""" 

111 # Stop any existing monitoring 

112 self._stop_monitoring() 

113 

114 # Start Textual worker using @work decorated method 

115 self.monitor_worker = self._monitor_worker() 

116 self.is_monitoring = True 

117 

118 def _stop_monitoring(self) -> None: 

119 """Stop the monitoring worker""" 

120 if self.monitor_worker and not self.monitor_worker.is_finished: 

121 self.monitor_worker.cancel() 

122 self.monitor_worker = None 

123 self.is_monitoring = False 

124 

125 def toggle_monitoring(self) -> None: 

126 """Toggle monitoring on/off - called from menu""" 

127 if self.is_monitoring: 

128 self._stop_monitoring() 

129 else: 

130 self._start_monitoring() 

131 

132 async def manual_refresh(self) -> None: 

133 """Manual refresh - called from menu""" 

134 await self.update_display() 

135 

136 @work(thread=True, exclusive=True) 

137 def _monitor_worker(self) -> None: 

138 """Textual worker that collects system metrics and updates UI.""" 

139 worker = get_current_worker() 

140 import logging 

141 logger = logging.getLogger(__name__) 

142 logger.info("SystemMonitor worker started") 

143 

144 while not worker.is_cancelled: 

145 try: 

146 # Collect system metrics 

147 self.monitor.update_metrics() 

148 

149 # Update UI via call_from_thread every cycle (1 second updates) 

150 if not worker.is_cancelled: 

151 self.app.call_from_thread(self._update_ui_from_thread) 

152 

153 # Sleep for the update interval (default 1 second) 

154 time.sleep(self.update_interval) 

155 

156 except Exception as e: 

157 import logging 

158 logger = logging.getLogger(__name__) 

159 logger.debug(f"Error in system monitor worker: {e}") 

160 time.sleep(2.0) # Longer sleep on error 

161 

162 def _update_ui_from_thread(self) -> None: 

163 """Update UI components - called from background thread via call_from_thread.""" 

164 if PLOTEXT_AVAILABLE: 

165 self._update_plots() 

166 else: 

167 self._update_fallback_display() 

168 

169 self._update_system_info() 

170 

171 def _get_ascii_header(self) -> str: 

172 """Get the OpenHCS ASCII art header""" 

173 return """ 

174╔══════════════════════════════════════════════════════════════╗ 

175║ ██████╗ ██████╗ ███████╗███╗ ██╗██╗ ██╗ ██████╗███████╗ ║ 

176║ ██╔═══██╗██╔══██╗██╔════╝████╗ ██║██║ ██║██╔════╝██╔════╝ ║ 

177║ ██║ ██║██████╔╝█████╗ ██╔██╗ ██║███████║██║ ███████╗ ║ 

178║ ██║ ██║██╔═══╝ ██╔══╝ ██║╚██╗██║██╔══██║██║ ╚════██║ ║ 

179║ ╚██████╔╝██║ ███████╗██║ ╚████║██║ ██║╚██████╗███████║ ║ 

180║ ╚═════╝ ╚═╝ ╚══════╝╚═╝ ╚═══╝╚═╝ ╚═╝ ╚═════╝╚══════╝ ║ 

181╚══════════════════════════════════════════════════════════════╝""" 

182 

183 async def update_display(self) -> None: 

184 """Update the display with new system stats - for manual refresh only""" 

185 # Update metrics using the SystemMonitor from services - wrap in executor to avoid blocking 

186 import asyncio 

187 await asyncio.get_event_loop().run_in_executor(None, self.monitor.update_metrics) 

188 

189 # Update UI directly (this is for manual refresh) 

190 self._update_ui_from_thread() 

191 

192 def _update_plots(self) -> None: 

193 """Update the plotext plots""" 

194 try: 

195 # Get current metrics 

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

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

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

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

200 

201 x_range = list(range(len(self.monitor.cpu_history))) 

202 

203 # Update CPU plot with Braille patterns for higher resolution 

204 cpu_plot = self.query_one("#cpu_plot", PlotextPlot) 

205 cpu_plot.plt.clear_data() 

206 cpu_plot.plt.plot(x_range, list(self.monitor.cpu_history), marker="braille") 

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

208 cpu_plot.plt.ylim(0, 100) 

209 cpu_plot.refresh() # Force visual refresh 

210 

211 # Update RAM plot with Braille patterns for higher resolution 

212 ram_plot = self.query_one("#ram_plot", PlotextPlot) 

213 ram_plot.plt.clear_data() 

214 ram_plot.plt.plot(x_range, list(self.monitor.ram_history), marker="braille") 

215 ram_plot.plt.title(f"RAM Usage: {current_ram:.1f}%") 

216 ram_plot.plt.ylim(0, 100) 

217 ram_plot.refresh() # Force visual refresh 

218 

219 # Update GPU plot with Braille patterns for higher resolution 

220 gpu_plot = self.query_one("#gpu_plot", PlotextPlot) 

221 gpu_plot.plt.clear_data() 

222 if GPU_AVAILABLE and any(self.monitor.gpu_history): 

223 gpu_plot.plt.plot(x_range, list(self.monitor.gpu_history), marker="braille") 

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

225 else: 

226 gpu_plot.plt.title("GPU: Not Available") 

227 gpu_plot.plt.ylim(0, 100) 

228 gpu_plot.refresh() # Force visual refresh 

229 

230 # Update VRAM plot with Braille patterns for higher resolution 

231 vram_plot = self.query_one("#vram_plot", PlotextPlot) 

232 vram_plot.plt.clear_data() 

233 if GPU_AVAILABLE and any(self.monitor.vram_history): 

234 vram_plot.plt.plot(x_range, list(self.monitor.vram_history), marker="braille") 

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

236 else: 

237 vram_plot.plt.title("VRAM: Not Available") 

238 vram_plot.plt.ylim(0, 100) 

239 vram_plot.refresh() # Force visual refresh 

240 

241 except Exception as e: 

242 # If plots fail, fall back to text display 

243 self._update_fallback_display() 

244 

245 def _update_fallback_display(self) -> None: 

246 """Update with text-based display if plots aren't available""" 

247 try: 

248 metrics = self.monitor.get_metrics_dict() 

249 

250 display_text = f""" 

251┌─────────────────────────────────────────────────────────────────┐ 

252│ CPU: {self._create_bar(metrics.get('cpu_percent', 0))} {metrics.get('cpu_percent', 0):5.1f}% 

253│ RAM: {self._create_bar(metrics.get('ram_percent', 0))} {metrics.get('ram_percent', 0):5.1f}% ({metrics.get('ram_used_gb', 0):.1f}/{metrics.get('ram_total_gb', 0):.1f}GB) 

254│ GPU: {self._create_bar(metrics.get('gpu_percent', 0))} {metrics.get('gpu_percent', 0):5.1f}% 

255│ VRAM: {self._create_bar(metrics.get('vram_percent', 0))} {metrics.get('vram_percent', 0):5.1f}% 

256└─────────────────────────────────────────────────────────────────┘ 

257""" 

258 

259 fallback_widget = self.query_one("#fallback_display", Static) 

260 fallback_widget.update(display_text) 

261 except Exception: 

262 pass 

263 

264 def _update_system_info(self) -> None: 

265 """Update system information in top-right""" 

266 try: 

267 from datetime import datetime 

268 # Now get_metrics_dict() just returns cached data - no blocking calls! 

269 metrics = self.monitor.get_metrics_dict() 

270 

271 info_text = f""" 

272═══════════════════════════════════════════════════════════════════════ 

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

274═══════════════════════════════════════════════════════════════════════ 

275CPU Cores: {metrics.get('cpu_cores', 'N/A')} | CPU Frequency: {metrics.get('cpu_freq_mhz', 0):.0f} MHz 

276Total RAM: {metrics.get('ram_total_gb', 0):.1f} GB | Used RAM: {metrics.get('ram_used_gb', 0):.1f} GB""" 

277 

278 if GPU_AVAILABLE and 'gpu_name' in metrics: 

279 info_text += f"\nGPU: {metrics.get('gpu_name', 'N/A')} | Temperature: {metrics.get('gpu_temp', 'N/A')}°C" 

280 info_text += f"\nVRAM: {metrics.get('vram_used_mb', 0):.0f}/{metrics.get('vram_total_mb', 0):.0f} MB" 

281 

282 info_text += "\n═══════════════════════════════════════════════════════════════════════" 

283 

284 info_widget = self.query_one("#system_info", Static) 

285 info_widget.update(info_text) 

286 except Exception: 

287 pass 

288 

289 def _create_bar(self, percent): 

290 """Create a text-based progress bar""" 

291 bar_length = 20 

292 filled = int(bar_length * percent / 100) 

293 bar = '█' * filled + '░' * (bar_length - filled) 

294 return f"[{bar}]" 

295