Coverage for openhcs/textual_tui/widgets/system_monitor.py: 0.0%
148 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-04 02:09 +0000
1from textual.app import ComposeResult
2from textual.widgets import Static
3from textual.containers import Container, Horizontal, Grid
4from textual import work
5from textual.reactive import reactive
6from textual.worker import get_current_worker
7import asyncio
8import time
10# Use textual-plotext for proper Textual integration
11try:
12 from textual_plotext import PlotextPlot
13 PLOTEXT_AVAILABLE = True
14except ImportError:
15 PLOTEXT_AVAILABLE = False
17# Import the SystemMonitor from services
18from openhcs.textual_tui.services.system_monitor import SystemMonitor
20try:
21 import GPUtil
22 GPU_AVAILABLE = True
23except ImportError:
24 GPU_AVAILABLE = False
26class SystemMonitorTextual(Container):
27 """A Textual widget for system monitoring using proper Textual components"""
29 DEFAULT_CSS = """
30 SystemMonitorTextual {
31 background: $background;
32 color: $text;
33 }
35 #top_row {
36 height: 9;
37 background: $surface;
38 }
40 #header {
41 width: 1fr;
42 text-align: left;
43 color: $accent;
44 padding: 1;
45 }
47 #system_info {
48 width: 1fr;
49 color: $text-muted;
50 text-align: right;
51 padding: 1;
52 content-align: center middle;
53 }
55 #plots_grid {
56 height: 1fr;
57 grid-size: 2 2;
58 grid-gutter: 1;
59 }
61 #fallback_display {
62 height: 1fr;
63 text-align: center;
64 color: $primary;
65 }
68 """
70 update_interval = reactive(1.0)
71 is_monitoring = reactive(True)
73 def __init__(self):
74 super().__init__()
75 self.monitor = SystemMonitor()
76 self.monitor_worker = None
78 def compose(self) -> ComposeResult:
79 """Compose the system monitor layout"""
80 # Top row with title on left and system info on right
81 with Horizontal(id="top_row"):
82 yield Static(self._get_ascii_header(), id="header")
83 yield Static("", id="system_info")
85 if PLOTEXT_AVAILABLE:
86 # Use Grid layout for plots
87 with Grid(id="plots_grid"):
88 yield PlotextPlot(id="cpu_plot")
89 yield PlotextPlot(id="ram_plot")
90 yield PlotextPlot(id="gpu_plot")
91 yield PlotextPlot(id="vram_plot")
92 else:
93 # Fallback to text-based display
94 yield Static("", id="fallback_display")
96 async def on_mount(self) -> None:
97 """Start the monitoring when widget is mounted"""
98 # Start the worker using the @work decorated method
99 self.monitor_worker = self._monitor_worker()
100 self.is_monitoring = True
102 def on_unmount(self) -> None:
103 """Stop monitoring and cleanup background thread when widget is unmounted"""
104 self._stop_monitoring()
106 def _start_monitoring(self) -> None:
107 """Start the monitoring Textual worker"""
108 # Stop any existing monitoring
109 self._stop_monitoring()
111 # Start Textual worker using @work decorated method
112 self.monitor_worker = self._monitor_worker()
113 self.is_monitoring = True
115 def _stop_monitoring(self) -> None:
116 """Stop the monitoring worker"""
117 if self.monitor_worker and not self.monitor_worker.is_finished:
118 self.monitor_worker.cancel()
119 self.monitor_worker = None
120 self.is_monitoring = False
122 def toggle_monitoring(self) -> None:
123 """Toggle monitoring on/off - called from menu"""
124 if self.is_monitoring:
125 self._stop_monitoring()
126 else:
127 self._start_monitoring()
129 async def manual_refresh(self) -> None:
130 """Manual refresh - called from menu"""
131 await self.update_display()
133 @work(thread=True, exclusive=True)
134 def _monitor_worker(self) -> None:
135 """Textual worker that collects system metrics and updates UI."""
136 worker = get_current_worker()
137 import logging
138 logger = logging.getLogger(__name__)
139 logger.info("SystemMonitor worker started")
141 while not worker.is_cancelled:
142 try:
143 # Collect system metrics
144 self.monitor.update_metrics()
146 # Update UI via call_from_thread every cycle (1 second updates)
147 if not worker.is_cancelled:
148 self.app.call_from_thread(self._update_ui_from_thread)
150 # Sleep for the update interval (default 1 second)
151 time.sleep(self.update_interval)
153 except Exception as e:
154 import logging
155 logger = logging.getLogger(__name__)
156 logger.debug(f"Error in system monitor worker: {e}")
157 time.sleep(2.0) # Longer sleep on error
159 def _update_ui_from_thread(self) -> None:
160 """Update UI components - called from background thread via call_from_thread."""
161 if PLOTEXT_AVAILABLE:
162 self._update_plots()
163 else:
164 self._update_fallback_display()
166 self._update_system_info()
168 def _get_ascii_header(self) -> str:
169 """Get the OpenHCS ASCII art header"""
170 return """
171╔══════════════════════════════════════════════════════════════╗
172║ ██████╗ ██████╗ ███████╗███╗ ██╗██╗ ██╗ ██████╗███████╗ ║
173║ ██╔═══██╗██╔══██╗██╔════╝████╗ ██║██║ ██║██╔════╝██╔════╝ ║
174║ ██║ ██║██████╔╝█████╗ ██╔██╗ ██║███████║██║ ███████╗ ║
175║ ██║ ██║██╔═══╝ ██╔══╝ ██║╚██╗██║██╔══██║██║ ╚════██║ ║
176║ ╚██████╔╝██║ ███████╗██║ ╚████║██║ ██║╚██████╗███████║ ║
177║ ╚═════╝ ╚═╝ ╚══════╝╚═╝ ╚═══╝╚═╝ ╚═╝ ╚═════╝╚══════╝ ║
178╚══════════════════════════════════════════════════════════════╝"""
180 async def update_display(self) -> None:
181 """Update the display with new system stats - for manual refresh only"""
182 # Update metrics using the SystemMonitor from services - wrap in executor to avoid blocking
183 await asyncio.get_event_loop().run_in_executor(None, self.monitor.update_metrics)
185 # Update UI directly (this is for manual refresh)
186 self._update_ui_from_thread()
188 def _update_plots(self) -> None:
189 """Update the plotext plots"""
190 try:
191 # Get current metrics
192 current_cpu = self.monitor.cpu_history[-1] if self.monitor.cpu_history else 0
193 current_ram = self.monitor.ram_history[-1] if self.monitor.ram_history else 0
194 current_gpu = self.monitor.gpu_history[-1] if self.monitor.gpu_history else 0
195 current_vram = self.monitor.vram_history[-1] if self.monitor.vram_history else 0
197 x_range = list(range(len(self.monitor.cpu_history)))
199 # Update CPU plot with Braille patterns for higher resolution
200 cpu_plot = self.query_one("#cpu_plot", PlotextPlot)
201 cpu_plot.plt.clear_data()
202 cpu_plot.plt.plot(x_range, list(self.monitor.cpu_history), marker="braille")
203 cpu_plot.plt.title(f"CPU Usage: {current_cpu:.1f}%")
204 cpu_plot.plt.ylim(0, 100)
205 cpu_plot.refresh() # Force visual refresh
207 # Update RAM plot with Braille patterns for higher resolution
208 ram_plot = self.query_one("#ram_plot", PlotextPlot)
209 ram_plot.plt.clear_data()
210 ram_plot.plt.plot(x_range, list(self.monitor.ram_history), marker="braille")
211 ram_plot.plt.title(f"RAM Usage: {current_ram:.1f}%")
212 ram_plot.plt.ylim(0, 100)
213 ram_plot.refresh() # Force visual refresh
215 # Update GPU plot with Braille patterns for higher resolution
216 gpu_plot = self.query_one("#gpu_plot", PlotextPlot)
217 gpu_plot.plt.clear_data()
218 if GPU_AVAILABLE and any(self.monitor.gpu_history):
219 gpu_plot.plt.plot(x_range, list(self.monitor.gpu_history), marker="braille")
220 gpu_plot.plt.title(f"GPU Usage: {current_gpu:.1f}%")
221 else:
222 gpu_plot.plt.title("GPU: Not Available")
223 gpu_plot.plt.ylim(0, 100)
224 gpu_plot.refresh() # Force visual refresh
226 # Update VRAM plot with Braille patterns for higher resolution
227 vram_plot = self.query_one("#vram_plot", PlotextPlot)
228 vram_plot.plt.clear_data()
229 if GPU_AVAILABLE and any(self.monitor.vram_history):
230 vram_plot.plt.plot(x_range, list(self.monitor.vram_history), marker="braille")
231 vram_plot.plt.title(f"VRAM Usage: {current_vram:.1f}%")
232 else:
233 vram_plot.plt.title("VRAM: Not Available")
234 vram_plot.plt.ylim(0, 100)
235 vram_plot.refresh() # Force visual refresh
237 except Exception as e:
238 # If plots fail, fall back to text display
239 self._update_fallback_display()
241 def _update_fallback_display(self) -> None:
242 """Update with text-based display if plots aren't available"""
243 try:
244 metrics = self.monitor.get_metrics_dict()
246 display_text = f"""
247┌─────────────────────────────────────────────────────────────────┐
248│ CPU: {self._create_bar(metrics.get('cpu_percent', 0))} {metrics.get('cpu_percent', 0):5.1f}%
249│ 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)
250│ GPU: {self._create_bar(metrics.get('gpu_percent', 0))} {metrics.get('gpu_percent', 0):5.1f}%
251│ VRAM: {self._create_bar(metrics.get('vram_percent', 0))} {metrics.get('vram_percent', 0):5.1f}%
252└─────────────────────────────────────────────────────────────────┘
253"""
255 fallback_widget = self.query_one("#fallback_display", Static)
256 fallback_widget.update(display_text)
257 except Exception:
258 pass
260 def _update_system_info(self) -> None:
261 """Update system information in top-right"""
262 try:
263 from datetime import datetime
264 # Now get_metrics_dict() just returns cached data - no blocking calls!
265 metrics = self.monitor.get_metrics_dict()
267 info_text = f"""
268═══════════════════════════════════════════════════════════════════════
269System Information | Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
270═══════════════════════════════════════════════════════════════════════
271CPU Cores: {metrics.get('cpu_cores', 'N/A')} | CPU Frequency: {metrics.get('cpu_freq_mhz', 0):.0f} MHz
272Total RAM: {metrics.get('ram_total_gb', 0):.1f} GB | Used RAM: {metrics.get('ram_used_gb', 0):.1f} GB"""
274 if GPU_AVAILABLE and 'gpu_name' in metrics:
275 info_text += f"\nGPU: {metrics.get('gpu_name', 'N/A')} | Temperature: {metrics.get('gpu_temp', 'N/A')}°C"
276 info_text += f"\nVRAM: {metrics.get('vram_used_mb', 0):.0f}/{metrics.get('vram_total_mb', 0):.0f} MB"
278 info_text += "\n═══════════════════════════════════════════════════════════════════════"
280 info_widget = self.query_one("#system_info", Static)
281 info_widget.update(info_text)
282 except Exception:
283 pass
285 def _create_bar(self, percent):
286 """Create a text-based progress bar"""
287 bar_length = 20
288 filled = int(bar_length * percent / 100)
289 bar = '█' * filled + '░' * (bar_length - filled)
290 return f"[{bar}]"