Coverage for orchestrant/pipeline/ui/wx.py: 14%
175 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-18 00:48 +0000
1"""wxPython viewer for monitoring pipelines."""
3from __future__ import annotations
5import threading
6from contextlib import suppress
7from typing import TYPE_CHECKING, Any, cast
9import cv2
10from loguru import logger
12from orchestrant.pipeline.ui.viewmodel import format_labels
15wx: Any | None
16_WX_IMPORT_ERROR: ImportError | None = None
18try:
19 import wx as _wx # ty: ignore[unresolved-import]
20except ImportError as exc: # pragma: no cover - optional dependency
21 wx = None
22 _WX_IMPORT_ERROR = exc
23else:
24 wx = _wx
25 _WX_IMPORT_ERROR = None
27if TYPE_CHECKING:
28 import numpy as np
30 from orchestrant.pipeline.types import (
31 PerformanceMetrics,
32 SystemStats,
33 )
36class WxPythonViewer:
37 """wxPython viewer for rendering frames with a side info panel."""
39 def __init__(self, width: int, height: int, title: str = "YOLO Monitor") -> None:
40 """Initialize the wxPython UI in a background thread."""
41 if wx is None:
42 message = "wxPython is required for WxPythonViewer"
43 raise ImportError(message) from _WX_IMPORT_ERROR
44 self.width = int(width)
45 self.height = int(height)
46 self._frame_size = (self.width, self.height)
47 self._open = True
48 self._closing = False
49 self._ready = threading.Event()
50 self._colors = {
51 "bg": wx.Colour(17, 24, 39),
52 "panel": wx.Colour(31, 41, 55),
53 "panel_alt": wx.Colour(24, 32, 45),
54 "border": wx.Colour(55, 65, 81),
55 "text": wx.Colour(229, 231, 235),
56 "muted": wx.Colour(156, 163, 175),
57 "accent": wx.Colour(34, 211, 238),
58 }
59 self._run_ui(title)
61 self._last_labels: dict[str, str] = {}
62 self._last_log_text = ""
63 self._needs_layout = True
65 def _run_ui(self, title: str) -> None:
66 """Build and show the wxPython UI."""
67 if wx is None:
68 message = "wxPython is required for WxPythonViewer"
69 raise ImportError(message) from _WX_IMPORT_ERROR
70 wx_lib = cast("Any", wx)
71 self.wx = wx_lib
72 self.app = wx_lib.GetApp() or wx_lib.App(redirect=False)
73 self.frame = wx_lib.Frame(
74 None,
75 title=title,
76 size=wx_lib.Size(self.width + 360, self.height + 120),
77 )
78 self.panel = wx_lib.Panel(self.frame)
79 self.frame.SetBackgroundColour(self._colors["bg"])
80 self.panel.SetBackgroundColour(self._colors["panel"])
81 self.frame.SetDoubleBuffered(on=True)
82 self.panel.SetDoubleBuffered(on=True)
84 class _FramePanel(wx_lib.Panel):
85 def __init__(self, parent: object) -> None:
86 super().__init__(parent)
87 self._bitmap: object | None = None
88 self.SetBackgroundStyle(wx_lib.BG_STYLE_PAINT)
89 self.SetBackgroundColour(cast("Any", parent).GetBackgroundColour())
90 self.Bind(wx_lib.EVT_PAINT, self._on_paint)
92 def set_bitmap(self, bmp: object) -> None:
93 self._bitmap = bmp
94 self.Refresh(eraseBackground=False)
96 def _on_paint(self, _event: object) -> None:
97 dc = wx_lib.BufferedPaintDC(self)
98 dc.SetBackground(wx_lib.Brush(self.GetBackgroundColour()))
99 dc.Clear()
100 if self._bitmap is not None:
101 dc.DrawBitmap(self._bitmap, 0, 0)
103 self.frame_panel = _FramePanel(self.panel)
104 self.frame_panel.SetMinSize((self.width, self.height))
105 self.frame_panel.SetSize((self.width, self.height))
107 self._labels = {
108 "resolution": wx.StaticText(self.panel, label=""),
109 "capture": wx.StaticText(self.panel, label=""),
110 "backend": wx.StaticText(self.panel, label=""),
111 "pipeline": wx.StaticText(self.panel, label=""),
112 "cpu_model": wx.StaticText(self.panel, label=""),
113 "ram_total": wx.StaticText(self.panel, label=""),
114 "gpu_model": wx.StaticText(self.panel, label=""),
115 "vram_total": wx.StaticText(self.panel, label=""),
116 "detections": wx.StaticText(self.panel, label=""),
117 "camera_fps": wx.StaticText(self.panel, label=""),
118 "inference_ms": wx.StaticText(self.panel, label=""),
119 "budget": wx.StaticText(self.panel, label=""),
120 "headroom": wx.StaticText(self.panel, label=""),
121 "sys_cpu": wx.StaticText(self.panel, label=""),
122 "sys_ram": wx.StaticText(self.panel, label=""),
123 "gpu": wx.StaticText(self.panel, label=""),
124 "vram": wx.StaticText(self.panel, label=""),
125 "power": wx.StaticText(self.panel, label=""),
126 "energy": wx.StaticText(self.panel, label=""),
127 "proc": wx.StaticText(self.panel, label=""),
128 "class": wx.StaticText(self.panel, label=""),
129 }
131 for label in self._labels.values():
132 label.SetForegroundColour(self._colors["text"])
134 self.log_ctrl = wx_lib.TextCtrl(
135 self.panel,
136 style=wx_lib.TE_MULTILINE | wx_lib.TE_READONLY,
137 size=wx_lib.Size(320, 200),
138 )
139 self.log_ctrl.SetBackgroundColour(self._colors["panel_alt"])
140 self.log_ctrl.SetForegroundColour(self._colors["text"])
142 title_label = wx_lib.StaticText(self.panel, label="YOLO Monitor")
143 title_label.SetForegroundColour(self._colors["accent"])
144 title_font = title_label.GetFont()
145 title_font.SetPointSize(title_font.GetPointSize() + 2)
146 title_font.SetWeight(wx_lib.FONTWEIGHT_BOLD)
147 title_label.SetFont(title_font)
149 def _make_section(title: str, keys: list[str]) -> object:
150 box = wx_lib.StaticBox(self.panel, label=title)
151 box.SetForegroundColour(self._colors["muted"])
152 sizer = wx_lib.StaticBoxSizer(box, wx_lib.VERTICAL)
153 for key in keys:
154 sizer.Add(self._labels[key], 0, wx_lib.ALL, 2)
155 return sizer
157 right_sizer = wx_lib.BoxSizer(wx_lib.VERTICAL)
158 right_sizer.Add(title_label, 0, wx_lib.ALL, 6)
159 right_sizer.Add(
160 _make_section(
161 "Overview",
162 [
163 "resolution",
164 "capture",
165 "backend",
166 "pipeline",
167 "cpu_model",
168 "ram_total",
169 "gpu_model",
170 "vram_total",
171 "detections",
172 ],
173 ),
174 0,
175 wx_lib.EXPAND | wx_lib.ALL,
176 4,
177 )
178 right_sizer.Add(
179 _make_section(
180 "Performance",
181 ["camera_fps", "inference_ms", "budget", "headroom"],
182 ),
183 0,
184 wx_lib.EXPAND | wx_lib.ALL,
185 4,
186 )
187 right_sizer.Add(
188 _make_section(
189 "System",
190 ["sys_cpu", "sys_ram", "gpu", "vram", "power", "energy"],
191 ),
192 0,
193 wx_lib.EXPAND | wx_lib.ALL,
194 4,
195 )
196 right_sizer.Add(
197 _make_section("Process", ["proc", "class"]),
198 0,
199 wx_lib.EXPAND | wx_lib.ALL,
200 4,
201 )
202 right_sizer.Add(self.log_ctrl, 0, wx_lib.ALL, 6)
204 main_sizer = wx_lib.BoxSizer(wx_lib.HORIZONTAL)
205 main_sizer.Add(self.frame_panel, 0, wx_lib.ALL, 6)
206 main_sizer.Add(right_sizer, 0, wx_lib.ALL, 6)
208 self.panel.SetSizer(main_sizer)
209 self.frame.Bind(wx_lib.EVT_CLOSE, self._on_close)
210 self.frame.Show()
212 self._ready.set()
214 def run(self) -> None:
215 """Run the wxPython main loop if not already running."""
216 if not self._ready.is_set():
217 return
218 try:
219 if not self.app.IsMainLoopRunning():
220 self.app.MainLoop()
221 except Exception as exc:
222 logger.debug("wxPython main loop failed: {}", exc)
224 def _on_close(self, event: object | None) -> None:
225 """Handle window close event and shutdown the UI."""
226 if self._closing:
227 return
228 self._closing = True
229 self._open = False
230 with suppress(Exception):
231 if hasattr(self, "frame") and self.frame:
232 self.frame.Destroy()
233 with suppress(Exception):
234 if hasattr(self, "app") and self.app:
235 self.app.ExitMainLoop()
236 if event is not None:
237 with suppress(Exception):
238 event_skip = getattr(event, "Skip", None)
239 if callable(event_skip):
240 event_skip()
242 def is_open(self) -> bool:
243 """Return True when the viewer is running and not closing."""
244 if not self._ready.is_set():
245 return False
246 return self._open and not self._closing
248 def render(
249 self,
250 frame: np.ndarray,
251 *,
252 perf_metrics: PerformanceMetrics | None = None,
253 sys_stats: SystemStats | None = None,
254 proc_stats: dict | None = None,
255 camera_info: dict | None = None,
256 detections_count: int | None = None,
257 classification: dict | None = None,
258 log_lines: list[str] | None = None,
259 hardware_info: dict | None = None,
260 power_info: dict | None = None,
261 ) -> None:
262 """Schedule a UI update for the provided frame and stats."""
263 if not self.is_open() or not self._ready.is_set():
264 return
265 with suppress(Exception):
266 self.wx.CallAfter(
267 self._update_ui,
268 frame,
269 perf_metrics=perf_metrics,
270 sys_stats=sys_stats,
271 proc_stats=proc_stats,
272 camera_info=camera_info,
273 detections_count=detections_count,
274 classification=classification,
275 log_lines=log_lines,
276 hardware_info=hardware_info,
277 power_info=power_info,
278 )
280 def _update_frame(self, frame: np.ndarray) -> tuple[int, int]:
281 frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
282 h, w = frame_rgb.shape[:2]
283 if (w, h) != self._frame_size:
284 self._frame_size = (w, h)
285 self._needs_layout = True
286 self.frame_panel.SetMinSize((w, h))
287 self.frame_panel.SetSize((w, h))
289 bitmap = self.wx.Bitmap.FromBuffer(w, h, frame_rgb)
290 self.frame_panel.set_bitmap(bitmap)
291 return w, h
293 def _update_perf_labels(
294 self,
295 *,
296 perf_metrics: PerformanceMetrics | None,
297 sys_stats: SystemStats | None,
298 proc_stats: dict | None,
299 camera_info: dict | None,
300 detections_count: int | None,
301 hardware_info: dict | None,
302 power_info: dict | None,
303 classification: dict | None,
304 frame_size: tuple[int, int],
305 ) -> None:
306 labels = format_labels(
307 perf_metrics=perf_metrics,
308 sys_stats=sys_stats,
309 proc_stats=proc_stats,
310 camera_info=camera_info,
311 detections_count=detections_count,
312 hardware_info=hardware_info,
313 power_info=power_info,
314 classification=classification,
315 frame_size=frame_size,
316 )
318 label_map = {
319 "resolution": "resolution",
320 "capture": "capture",
321 "backend": "backend",
322 "pipeline": "pipeline",
323 "cpu_model": "cpu_model",
324 "ram_total": "ram_total",
325 "gpu_model": "gpu_model",
326 "vram_total": "vram_total",
327 "detections": "detections",
328 "camera_fps": "camera_fps",
329 "inference_ms": "inference_ms",
330 "budget": "budget",
331 "headroom": "headroom",
332 "sys_cpu": "sys_cpu",
333 "sys_ram": "sys_ram",
334 "gpu": "gpu",
335 "vram": "vram",
336 "power": "power",
337 "energy": "energy",
338 "proc": "proc",
339 "classification": "class",
340 }
341 for view_field, widget_key in label_map.items():
342 if widget_key in self._labels:
343 self._set_label(widget_key, getattr(labels, view_field))
345 def _update_logs(self, log_lines: list[str] | None) -> None:
346 if log_lines is None:
347 return
348 new_log = "\n".join(log_lines)
349 if new_log != self._last_log_text:
350 self.log_ctrl.Freeze()
351 self.log_ctrl.SetValue(new_log)
352 self.log_ctrl.Thaw()
353 self._last_log_text = new_log
355 def _update_ui(
356 self,
357 frame: np.ndarray,
358 *,
359 perf_metrics: PerformanceMetrics | None,
360 sys_stats: SystemStats | None,
361 proc_stats: dict | None,
362 camera_info: dict | None,
363 detections_count: int | None,
364 classification: dict | None,
365 log_lines: list[str] | None,
366 hardware_info: dict | None,
367 power_info: dict | None,
368 ) -> None:
369 """Update UI widgets with the latest frame and stats."""
370 if not self.is_open():
371 return
372 frame_size = self._update_frame(frame)
373 self._update_perf_labels(
374 perf_metrics=perf_metrics,
375 sys_stats=sys_stats,
376 proc_stats=proc_stats,
377 camera_info=camera_info,
378 detections_count=detections_count,
379 hardware_info=hardware_info,
380 power_info=power_info,
381 classification=classification,
382 frame_size=frame_size,
383 )
384 self._update_logs(log_lines)
386 if self._needs_layout:
387 self.panel.Layout()
388 self._needs_layout = False
389 self.wx.YieldIfNeeded()
391 def close(self) -> None:
392 """Request window close and stop UI callbacks."""
393 if self._open:
394 self._open = False
395 self._closing = True
396 with suppress(Exception):
397 if self._ready.is_set() and hasattr(self, "wx"):
398 self.wx.CallAfter(self._on_close, None)
400 def _set_label(self, key: str, text: str) -> None:
401 """Update a label only when the text changes."""
402 if self._last_labels.get(key) == text:
403 return
404 self._labels[key].SetLabel(text)
405 self._last_labels[key] = text