"""Main application window.""" from __future__ import annotations import logging from PySide6.QtCore import Qt, QTimer from PySide6.QtMultimediaWidgets import QVideoWidget from PySide6.QtWidgets import QLabel, QMainWindow, QSizePolicy, QStatusBar, QVBoxLayout, QWidget from app.camera.camera_enumerator import CameraEnumerator, CameraInfo from app.camera.camera_service import CameraService from app.config import APP_NAME, APP_VERSION from app.overlay.overlay_widget import OverlayWidget from app.pipeline.frame_dispatcher import FrameDispatcher from app.telemetry.telemetry_collector import TelemetryCollector from app.ui.menu_bar import AppMenuBar logger = logging.getLogger(__name__) class MainWindow(QMainWindow): """ Top-level application window. Wires together: CameraService → FrameDispatcher → TelemetryCollector → OverlayWidget (via metrics_updated) CameraService.capture_session → QVideoWidget (direct, zero-copy path) """ def __init__(self) -> None: super().__init__() self.setWindowTitle(f"{APP_NAME} v{APP_VERSION}") self.setMinimumSize(640, 480) self.resize(1280, 720) # --- Core components --- self._camera_service = CameraService(self) self._dispatcher = FrameDispatcher(self) self._telemetry = TelemetryCollector(parent=self) # --- Central container --- # We need an extra QWidget layer between QMainWindow and QVideoWidget so # that OverlayWidget can be a sibling of QVideoWidget (not its child). # QVideoWidget renders via a native D3D/GL surface that occludes any # QWidget children painted on top of it. self._container = QWidget(self) self._container.setStyleSheet("background: black;") container_layout = QVBoxLayout(self._container) container_layout.setContentsMargins(0, 0, 0, 0) container_layout.setSpacing(0) # --- Video widget --- self._video_widget = QVideoWidget(self._container) self._video_widget.setSizePolicy( QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding ) self._video_widget.setAspectRatioMode(Qt.AspectRatioMode.KeepAspectRatio) container_layout.addWidget(self._video_widget) self.setCentralWidget(self._container) # Connect camera session to video widget — this is the zero-copy render path self._camera_service.capture_session().setVideoOutput(self._video_widget) # --- Overlay --- # Sibling of QVideoWidget inside _container; positioned manually so it # floats above the video without being occluded by the native GL surface. self._overlay = OverlayWidget(parent=self._container) self._overlay.raise_() # --- Menu bar --- self._menu = AppMenuBar(self) self.setMenuBar(self._menu) # --- Status bar --- self._status_bar = QStatusBar(self) self.setStatusBar(self._status_bar) self._status_label = QLabel("Initialising…") self._status_bar.addWidget(self._status_label) # --- Wire signals --- self._wire_signals() # --- Enumerate cameras and start --- QTimer.singleShot(0, self._initialise_cameras) # Reposition overlay after the event loop starts (layout is finalised) QTimer.singleShot(0, self._reposition_overlay) # ------------------------------------------------------------------ # Initialisation # ------------------------------------------------------------------ def _initialise_cameras(self) -> None: cameras = CameraEnumerator.list_cameras() if not cameras: self._status_label.setText("No cameras found") logger.warning("No cameras detected") return self._menu.populate_cameras(cameras) default = CameraEnumerator.default_camera() start_cam = default if default is not None else cameras[0] self._menu.populate_formats(start_cam) self._start_camera(start_cam) def _start_camera(self, cam: CameraInfo) -> None: self._telemetry.reset_counters() self._camera_service.start(cam) self._menu.set_active_camera(cam) self._status_label.setText(f"Opening: {cam.name}") # ------------------------------------------------------------------ # Signal wiring # ------------------------------------------------------------------ def _wire_signals(self) -> None: # CameraService → FrameDispatcher self._camera_service.frame_ready.connect(self._dispatcher.dispatch) # FrameDispatcher → TelemetryCollector (never drop for telemetry) self._dispatcher.subscribe(self._telemetry.on_frame, drop_if_busy=False) # TelemetryCollector → OverlayWidget self._telemetry.metrics_updated.connect(self._overlay.update_metrics) # CameraService status self._camera_service.camera_started.connect(self._on_camera_started) self._camera_service.camera_stopped.connect(self._on_camera_stopped) self._camera_service.camera_error.connect(self._on_camera_error) # Menu signals self._menu.camera_selected.connect(self._on_camera_selected) self._menu.resolution_selected.connect(self._on_resolution_selected) self._menu.fps_selected.connect(self._on_fps_selected) self._menu.reconnect_requested.connect(self._camera_service.reconnect) self._menu.overlay_toggled.connect(self._overlay.set_overlay_visible) # ------------------------------------------------------------------ # Camera status slots # ------------------------------------------------------------------ def _on_camera_started(self) -> None: cam = self._camera_service.current_camera name = cam.name if cam else "Unknown" self._status_label.setText(f"Streaming: {name}") logger.info("Camera streaming: %s", name) def _on_camera_stopped(self) -> None: self._status_label.setText("Camera stopped") def _on_camera_error(self, message: str) -> None: self._status_label.setText(f"Error: {message}") logger.error("Camera error: %s", message) # ------------------------------------------------------------------ # Menu action slots # ------------------------------------------------------------------ def _on_camera_selected(self, cam: CameraInfo) -> None: self._start_camera(cam) def _on_resolution_selected(self, width: int, height: int) -> None: self._camera_service.set_resolution(width, height) def _on_fps_selected(self, fps: float) -> None: self._camera_service.set_fps(fps) # ------------------------------------------------------------------ # Qt overrides # ------------------------------------------------------------------ def resizeEvent(self, event) -> None: # noqa: N802 super().resizeEvent(event) self._reposition_overlay() def _reposition_overlay(self) -> None: """Keep the overlay covering the video widget exactly.""" if not (hasattr(self, "_overlay") and hasattr(self, "_video_widget")): return # _overlay and _video_widget share the same parent (_container), # so video_widget.geometry() is already in the right coordinate space. self._overlay.setGeometry(self._video_widget.geometry()) self._overlay.raise_() def closeEvent(self, event) -> None: # noqa: N802 self._camera_service.stop() super().closeEvent(event)