Coverage for orchestrant/pipeline/ui/flet_viewer.py: 79%
236 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-21 01:08 +0000
1"""Flet viewer for monitoring pipelines: a desktop window, or with ``web`` a page browsers open.
3The detection thread calls :meth:`FletViewer.render`; each window or browser tab has a
4pump thread that pushes the newest frame and drops the ones it could not keep up with.
5Controls change only on Flet's event loop, which every session shares.
6"""
8from __future__ import annotations
10import asyncio
11import threading
12from collections import deque
13from contextlib import suppress
14from dataclasses import dataclass
15from typing import TYPE_CHECKING, Any
17import cv2
18from loguru import logger
20from orchestrant.pipeline.ui.viewmodel import format_labels
23ft: Any
24cv: Any
25uvicorn: Any
26_FLET_IMPORT_ERROR: ImportError | None = None
27try:
28 import flet as _ft # ty: ignore[unresolved-import]
29 import flet.canvas as _cv # ty: ignore[unresolved-import]
30except ImportError as exc: # pragma: no cover - optional dependency
31 ft = cv = None
32 _FLET_IMPORT_ERROR = exc
33else:
34 ft, cv = _ft, _cv
35try:
36 import uvicorn as _uvicorn # ty: ignore[unresolved-import]
37except (
38 ImportError
39): # pragma: no cover - flet-web brings it, and the extra leaves it out on Windows
40 uvicorn = None
41else:
42 uvicorn = _uvicorn
44if TYPE_CHECKING:
45 from collections.abc import Callable
47 import numpy as np
49 from orchestrant.pipeline.types import (
50 PerformanceMetrics,
51 SystemStats,
52 )
53 from orchestrant.pipeline.ui.viewmodel import ViewLabels
56FLET_AVAILABLE = ft is not None
57HISTORY = 120
58SIDE_WIDTH = 380
59CHART_SIZE = (340.0, 110.0)
60FRAME_TIMEOUT_S = 2.0
61JPEG_QUALITY = 70
62# The frame area is at most ~1220 px wide; a 1080p frame as RGBA is 250 MB/s at 30 fps, 720p 110.
63MAX_EDGE = 1280
65# The panels and the ViewLabels fields each shows, as the wxPython and DearPyGui viewers had them.
66SECTIONS = (
67 (
68 "Overview",
69 (
70 "resolution",
71 "capture",
72 "backend",
73 "pipeline",
74 "cpu_model",
75 "ram_total",
76 "gpu_model",
77 "vram_total",
78 "detections",
79 ),
80 ),
81 ("Performance", ("camera_fps", "inference_ms", "budget", "headroom")),
82 ("System", ("sys_cpu", "sys_ram", "gpu", "vram", "power", "energy")),
83 ("Process", ("proc", "classification")),
84)
86# (title, series, legend, fixed top or None to fit the data)
87TRENDS = (
88 (
89 "Performance",
90 ("camera_fps", "inference_ms"),
91 ("Camera FPS", "Inference ms"),
92 None,
93 ),
94 ("System", ("cpu", "ram", "gpu"), ("CPU %", "RAM %", "GPU %"), 100.0),
95)
96SERIES_COLORS = ("PRIMARY", "TERTIARY", "ORANGE")
99def require_flet() -> Any: # noqa: ANN401 - the flet module, or ImportError
100 """The flet module; ImportError naming the extra when it is not installed."""
101 if ft is None:
102 message = "the Flet viewer needs flet (the frontend extra)"
103 raise ImportError(message) from _FLET_IMPORT_ERROR
104 return ft
107def web_server(main: Callable[[Any], None], host: str, port: int) -> Any: # noqa: ANN401
108 """A uvicorn server for `main` as a Flet web app; ``should_exit = True`` stops it from any thread."""
109 flet = require_flet()
110 if uvicorn is None:
111 message = (
112 "serving to a browser needs flet-web (the frontend extra, off Windows)"
113 )
114 raise ImportError(message)
115 app = flet.run(main, export_asgi_app=True)
116 return uvicorn.Server(
117 uvicorn.Config(app, host=host, port=port, log_level="warning")
118 )
121def polyline(values: list[float], top: float) -> list[tuple[float, float]]:
122 """Canvas points for one series, oldest at the left and `top` at the top edge."""
123 width, height = CHART_SIZE
124 top = top or 1.0
125 step = width / max(len(values) - 1, 1)
126 return [
127 (i * step, height - min(max(v / top, 0.0), 1.0) * height)
128 for i, v in enumerate(values)
129 ]
132def fitted(frame: np.ndarray, max_edge: int = MAX_EDGE) -> np.ndarray:
133 """The frame scaled down to a long edge of at most `max_edge`; a smaller one as it is."""
134 height, width = frame.shape[:2]
135 scale = max_edge / max(height, width)
136 if scale >= 1:
137 return frame
138 size = (round(width * scale), round(height * scale))
139 return cv2.resize(frame, size, interpolation=cv2.INTER_AREA)
142def frame_payload(
143 frame: np.ndarray, *, web: bool
144) -> tuple[int, int, bytes] | bytes | None:
145 """What a session's RawImage takes, at most MAX_EDGE wide: JPEG for a browser, RGBA for a window."""
146 frame = fitted(frame)
147 if web:
148 ok, buffer = cv2.imencode(
149 ".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, JPEG_QUALITY]
150 )
151 return buffer.tobytes() if ok else None
152 rgba = cv2.cvtColor(frame, cv2.COLOR_BGR2RGBA)
153 height, width = rgba.shape[:2]
154 return width, height, rgba.tobytes()
157class TrendHistory:
158 """The last HISTORY samples per series."""
160 def __init__(self, maxlen: int = HISTORY) -> None:
161 """Start every series empty."""
162 names = [name for _, series, _, _ in TRENDS for name in series]
163 self.series: dict[str, deque[float]] = {n: deque(maxlen=maxlen) for n in names}
165 def add(self, perf: PerformanceMetrics | None, stats: SystemStats | None) -> None:
166 """Append one sample per series; a frame without both readings adds nothing."""
167 if perf is None or stats is None:
168 return
169 self.series["camera_fps"].append(float(perf.camera_fps))
170 self.series["inference_ms"].append(float(perf.inference_ms))
171 self.series["cpu"].append(float(stats.cpu_percent))
172 self.series["ram"].append(float(stats.ram_percent))
173 gpu = float(stats.gpu_percent) if stats.gpu_name != "N/A" else 0.0
174 self.series["gpu"].append(gpu)
176 def copy(self) -> dict[str, list[float]]:
177 """Each series as a list, safe to hand to another thread."""
178 return {name: list(values) for name, values in self.series.items()}
181@dataclass(slots=True)
182class Snapshot:
183 """One render() call, formatted once for every session to draw."""
185 frame: np.ndarray
186 labels: ViewLabels
187 log: str
188 trends: dict[str, list[float]]
191class _Session:
192 """One window or browser tab: its controls, drawn on the event loop."""
194 def __init__(self, page: Any) -> None: # noqa: ANN401 - a flet.Page
195 flet = require_flet()
196 self.page = page
197 self.web = bool(page.web)
198 self.closed = threading.Event()
199 self.image = flet.RawImage(expand=True, fit=flet.BoxFit.CONTAIN)
200 self.labels = {
201 key: flet.Text("", size=12, selectable=True, visible=False)
202 for _, keys in SECTIONS
203 for key in keys
204 }
205 self.log = flet.Text("", size=11, font_family="monospace", selectable=True)
206 self.lines = {
207 name: cv.Points(
208 point_mode=cv.PointMode.POLYGON,
209 paint=flet.Paint(
210 color=getattr(flet.Colors, SERIES_COLORS[i]),
211 stroke_width=2,
212 style=flet.PaintingStyle.STROKE,
213 ),
214 )
215 for _, series, _, _ in TRENDS
216 for i, name in enumerate(series)
217 }
219 def layout(self, title: str) -> Any: # noqa: ANN401 - a flet control
220 """The frame on the left, the panels, trends and log on the right."""
221 flet = require_flet()
222 side = [flet.Text(title, size=18, weight=flet.FontWeight.BOLD)]
223 for heading, keys in SECTIONS:
224 side.append(_panel(heading, [self.labels[k] for k in keys]))
225 side += [_trend_panel(t, self.lines) for t in TRENDS]
226 side.append(_panel("Logs", [self.log]))
227 return flet.Row(
228 [
229 flet.Container(content=self.image, expand=True),
230 flet.Container(
231 content=flet.Column(
232 side,
233 scroll=flet.ScrollMode.AUTO,
234 spacing=8,
235 horizontal_alignment=flet.CrossAxisAlignment.STRETCH,
236 ),
237 width=SIDE_WIDTH,
238 ),
239 ],
240 expand=True,
241 vertical_alignment=flet.CrossAxisAlignment.START,
242 )
244 def show(self, snapshot: Snapshot) -> None:
245 """Set every label and trend line from one snapshot; runs on the event loop."""
246 try:
247 for key, text in self.labels.items():
248 text.value = getattr(snapshot.labels, key)
249 text.visible = bool(text.value)
250 self.log.value = snapshot.log
251 for _, series, _, top in TRENDS:
252 fitted = top or max(
253 (max(snapshot.trends[n], default=0.0) for n in series), default=1.0
254 )
255 for name in series:
256 self.lines[name].points = polyline(snapshot.trends[name], fitted)
257 self.page.update()
258 except (
259 Exception
260 ) as exc: # a window closing under an update ends this session only
261 logger.debug("Flet session update failed: {}", exc)
262 self.closed.set()
265def _panel(heading: str, rows: list[Any]) -> Any: # noqa: ANN401 - a flet control
266 flet = require_flet()
267 return flet.Container(
268 content=flet.Column(
269 [flet.Text(heading, size=13, color=flet.Colors.ON_SURFACE_VARIANT), *rows],
270 spacing=2,
271 ),
272 padding=flet.Padding.all(10),
273 border=flet.Border.all(1, flet.Colors.OUTLINE_VARIANT),
274 border_radius=flet.BorderRadius.all(8),
275 )
278def _trend_panel(trend: tuple, lines: dict[str, Any]) -> Any: # noqa: ANN401 - a flet control
279 flet = require_flet()
280 heading, series, legend, _ = trend
281 width, height = CHART_SIZE
282 keys = flet.Row(
283 [
284 flet.Text(label, size=11, color=getattr(flet.Colors, SERIES_COLORS[i]))
285 for i, label in enumerate(legend)
286 ],
287 spacing=12,
288 )
289 canvas = cv.Canvas([lines[name] for name in series], width=width, height=height)
290 return _panel(f"Trends: {heading}", [keys, canvas])
293class FletViewer:
294 """The monitor's viewer, in a Flet window or, with `web`, served to browsers."""
296 def __init__(
297 self,
298 width: int,
299 height: int,
300 title: str = "YOLO Monitor",
301 *,
302 web: bool = False,
303 host: str = "127.0.0.1",
304 port: int = 8550,
305 ) -> None:
306 """Prepare the viewer; :meth:`run` opens it."""
307 require_flet()
308 self.width = int(width)
309 self.height = int(height)
310 self.title = title
311 self.web = web
312 self.host = host
313 self.port = port
314 self._changed = threading.Condition()
315 self._open = True
316 self._seq = 0
317 self._snapshot: Snapshot | None = None
318 self._history = TrendHistory()
319 self._sessions: list[_Session] = []
320 self._server: Any = None
322 def is_open(self) -> bool:
323 """False once the window closed, the server stopped or close() ran."""
324 with self._changed:
325 return self._open
327 def render(
328 self,
329 frame: np.ndarray,
330 *,
331 perf_metrics: PerformanceMetrics | None = None,
332 sys_stats: SystemStats | None = None,
333 proc_stats: dict | None = None,
334 camera_info: dict | None = None,
335 detections_count: int | None = None,
336 classification: dict | None = None,
337 log_lines: list[str] | None = None,
338 hardware_info: dict | None = None,
339 power_info: dict | None = None,
340 ) -> None:
341 """Hand the newest frame and telemetry to every session; never blocks on drawing."""
342 height, width = frame.shape[:2]
343 labels = format_labels(
344 perf_metrics=perf_metrics,
345 sys_stats=sys_stats,
346 proc_stats=proc_stats,
347 camera_info=camera_info,
348 detections_count=detections_count,
349 hardware_info=hardware_info,
350 power_info=power_info,
351 classification=classification,
352 frame_size=(width, height),
353 )
354 with self._changed:
355 if not self._open:
356 return
357 self._history.add(perf_metrics, sys_stats)
358 log = "\n".join(log_lines) if log_lines is not None else ""
359 self._snapshot = Snapshot(frame, labels, log, self._history.copy())
360 self._seq += 1
361 self._changed.notify_all()
363 def run(self) -> None:
364 """Serve until the window closes, Ctrl+C, or close() from the detection thread."""
365 flet = require_flet()
366 try:
367 if self.web:
368 server = web_server(self._main, self.host, self.port)
369 with self._changed:
370 self._server = server
371 serve = self._open
372 if serve:
373 logger.info("Monitor at http://{}:{}", self.host, self.port)
374 server.run()
375 else:
376 flet.run(self._main, view=flet.AppView.FLET_APP)
377 with self._changed:
378 opened = bool(self._sessions)
379 if not opened:
380 logger.error(
381 "The Flet window never opened: its client exited first (its "
382 "error is above; on Linux it needs libsecret-1-0). "
383 "--ui opencv uses OpenCV's window instead."
384 )
385 finally:
386 self._mark_closed()
388 def close(self) -> None:
389 """Stop drawing, close every window and stop the server."""
390 sessions = self._mark_closed()
391 if self._server is not None:
392 self._server.should_exit = True
393 for session in sessions:
394 if not session.web:
395 with suppress(RuntimeError):
396 asyncio.run_coroutine_threadsafe(
397 session.page.window.close(), session.page.loop
398 )
400 def _mark_closed(self) -> list[_Session]:
401 with self._changed:
402 self._open = False
403 self._changed.notify_all()
404 return list(self._sessions)
406 def _main(self, page: Any) -> None: # noqa: ANN401 - a flet.Page
407 flet = require_flet()
408 page.title = self.title
409 page.theme = flet.Theme(color_scheme_seed=flet.Colors.GREEN_ACCENT)
410 page.dark_theme = flet.Theme(color_scheme_seed=flet.Colors.GREEN_ACCENT)
411 page.theme_mode = flet.ThemeMode.DARK
412 page.padding = 12
413 session = _Session(page)
414 if not session.web:
415 page.window.width = min(self.width + SIDE_WIDTH + 48, 1600)
416 # Tall enough for the side panels, whatever the camera's height.
417 page.window.height = min(max(self.height + 48, 900), 1000)
418 page.on_close = page.on_disconnect = lambda _event: session.closed.set()
419 with self._changed:
420 self._sessions.append(session)
421 page.add(session.layout(self.title))
422 page.run_thread(self.pump, session)
424 def next_snapshot(
425 self, session: _Session, seen: int
426 ) -> tuple[int, Snapshot] | None:
427 """Wait for a snapshot newer than `seen`; None once the viewer or the session closed."""
428 with self._changed:
429 while self._open and not session.closed.is_set() and self._seq == seen:
430 self._changed.wait(timeout=0.5)
431 if not self._open or session.closed.is_set() or self._snapshot is None:
432 return None
433 return self._seq, self._snapshot
435 def pump(self, session: _Session) -> None:
436 """Feed one session the newest frames until it or the viewer closes."""
437 seen = 0
438 while (latest := self.next_snapshot(session, seen)) is not None:
439 seen, snapshot = latest
440 try:
441 self._push(session, snapshot)
442 except TimeoutError:
443 logger.debug("Flet session fell behind; dropping a frame")
444 except (
445 Exception
446 ) as exc: # a closed window or tab ends its pump, not the monitor
447 logger.debug("Flet session ended: {}", exc)
448 session.closed.set()
450 @staticmethod
451 def _push(session: _Session, snapshot: Snapshot) -> None:
452 payload = frame_payload(snapshot.frame, web=session.web)
453 if payload is None:
454 logger.warning("Frame encoding failed; skipping frame")
455 return
456 if isinstance(payload, bytes):
457 drawn = session.image.render_encoded(payload)
458 else:
459 drawn = session.image.render_rgba(*payload)
460 future = asyncio.run_coroutine_threadsafe(drawn, session.page.loop)
461 try:
462 future.result(timeout=FRAME_TIMEOUT_S)
463 except TimeoutError:
464 future.cancel()
465 raise
466 session.page.loop.call_soon_threadsafe(session.show, snapshot)