Coverage for orchestr_ant_ion / pipeline / ui / wx.py: 15%
175 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-19 08:36 +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 orchestr_ant_ion.pipeline.ui.viewmodel import format_labels
15wx: Any | None
16_WX_IMPORT_ERROR: ImportError | None = None
18try:
19 import wx as _wx
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 orchestr_ant_ion.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 perf_metrics: PerformanceMetrics | None = None,
252 sys_stats: SystemStats | None = None,
253 proc_stats: dict | None = None,
254 camera_info: dict | None = None,
255 detections_count: int | None = None,
256 classification: dict | None = None,
257 log_lines: list[str] | None = None,
258 hardware_info: dict | None = None,
259 power_info: dict | None = None,
260 ) -> None:
261 """Schedule a UI update for the provided frame and stats."""
262 if not self.is_open() or not self._ready.is_set():
263 return
264 with suppress(Exception):
265 self.wx.CallAfter(
266 self._update_ui,
267 frame,
268 perf_metrics,
269 sys_stats,
270 proc_stats,
271 camera_info,
272 detections_count,
273 classification,
274 log_lines,
275 hardware_info,
276 power_info,
277 )
279 def _update_frame(self, frame: np.ndarray) -> tuple[int, int]:
280 frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
281 h, w = frame_rgb.shape[:2]
282 if (w, h) != self._frame_size:
283 self._frame_size = (w, h)
284 self._needs_layout = True
285 self.frame_panel.SetMinSize((w, h))
286 self.frame_panel.SetSize((w, h))
288 bitmap = self.wx.Bitmap.FromBuffer(w, h, frame_rgb)
289 self.frame_panel.set_bitmap(bitmap)
290 return w, h
292 def _update_perf_labels(
293 self,
294 *,
295 perf_metrics: PerformanceMetrics | None,
296 sys_stats: SystemStats | None,
297 proc_stats: dict | None,
298 camera_info: dict | None,
299 detections_count: int | None,
300 hardware_info: dict | None,
301 power_info: dict | None,
302 classification: dict | None,
303 frame_size: tuple[int, int],
304 ) -> None:
305 labels = format_labels(
306 perf_metrics=perf_metrics,
307 sys_stats=sys_stats,
308 proc_stats=proc_stats,
309 camera_info=camera_info,
310 detections_count=detections_count,
311 hardware_info=hardware_info,
312 power_info=power_info,
313 classification=classification,
314 frame_size=frame_size,
315 )
317 label_map = {
318 "resolution": "resolution",
319 "capture": "capture",
320 "backend": "backend",
321 "pipeline": "pipeline",
322 "cpu_model": "cpu_model",
323 "ram_total": "ram_total",
324 "gpu_model": "gpu_model",
325 "vram_total": "vram_total",
326 "detections": "detections",
327 "camera_fps": "camera_fps",
328 "inference_ms": "inference_ms",
329 "budget": "budget",
330 "headroom": "headroom",
331 "sys_cpu": "sys_cpu",
332 "sys_ram": "sys_ram",
333 "gpu": "gpu",
334 "vram": "vram",
335 "power": "power",
336 "energy": "energy",
337 "proc": "proc",
338 "classification": "class",
339 }
340 for view_field, widget_key in label_map.items():
341 if widget_key in self._labels:
342 self._set_label(widget_key, getattr(labels, view_field))
344 def _update_logs(self, log_lines: list[str] | None) -> None:
345 if log_lines is None:
346 return
347 new_log = "\n".join(log_lines)
348 if new_log != self._last_log_text:
349 self.log_ctrl.Freeze()
350 self.log_ctrl.SetValue(new_log)
351 self.log_ctrl.Thaw()
352 self._last_log_text = new_log
354 def _update_ui(
355 self,
356 frame: np.ndarray,
357 perf_metrics: PerformanceMetrics | None,
358 sys_stats: SystemStats | None,
359 proc_stats: dict | None,
360 camera_info: dict | None,
361 detections_count: int | None,
362 classification: dict | None,
363 log_lines: list[str] | None,
364 hardware_info: dict | None,
365 power_info: dict | None,
366 ) -> None:
367 """Update UI widgets with the latest frame and stats."""
368 if not self.is_open():
369 return
370 frame_size = self._update_frame(frame)
371 self._update_perf_labels(
372 perf_metrics=perf_metrics,
373 sys_stats=sys_stats,
374 proc_stats=proc_stats,
375 camera_info=camera_info,
376 detections_count=detections_count,
377 hardware_info=hardware_info,
378 power_info=power_info,
379 classification=classification,
380 frame_size=frame_size,
381 )
382 self._update_logs(log_lines)
384 if self._needs_layout:
385 self.panel.Layout()
386 self._needs_layout = False
387 self.wx.YieldIfNeeded()
389 def close(self) -> None:
390 """Request window close and stop UI callbacks."""
391 if self._open:
392 self._open = False
393 self._closing = True
394 with suppress(Exception):
395 if self._ready.is_set() and hasattr(self, "wx"):
396 self.wx.CallAfter(self._on_close, None)
398 def _set_label(self, key: str, text: str) -> None:
399 """Update a label only when the text changes."""
400 if self._last_labels.get(key) == text:
401 return
402 self._labels[key].SetLabel(text)
403 self._last_labels[key] = text