bartool/duck-preview
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: implement core functionality for camera preview application

Browse Source 
- Add FrameDispatcher for distributing QVideoFrames to subscribers
- Implement TelemetryCollector to measure video pipeline performance metrics
- Create MainWindow as the main application interface with video rendering
- Develop AppMenuBar for camera selection, resolution, and FPS settings
- Establish overlay system for displaying telemetry metrics
- Set up project structure and configuration files
- Add unit tests for FrameDispatcher and TelemetryCollector
This commit is contained in:
bartool
2026-05-12 19:49:53 +02:00
parent 65b98c352d
commit cd7f196b25
22 changed files with 1642 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
app/ui/__init__.py Normal file
View File

170
app/ui/main_window.py Normal file
View File

@@ -0,0 +1,170 @@
"""Main application window."""
from __future__ import annotations
import logging
from PySide6.QtCore import Qt, QTimer
from PySide6.QtMultimedia import QVideoWidget
from PySide6.QtWidgets import QLabel, QMainWindow, QSizePolicy, QStatusBar
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)
# --- Video widget (central widget) ---
self._video_widget = QVideoWidget(self)
self._video_widget.setSizePolicy(
QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding
)
self._video_widget.setAspectRatioMode(Qt.AspectRatioMode.KeepAspectRatio)
self.setCentralWidget(self._video_widget)
# Connect camera session to video widget — this is the zero-copy render path
self._camera_service.capture_session().setVideoOutput(self._video_widget)
# --- Overlay ---
self._overlay = OverlayWidget(parent=self._video_widget)
self._overlay.raise_()
self._overlay.resize(self._video_widget.size())
# --- 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)
# ------------------------------------------------------------------
# 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)
# Keep overlay covering the video widget
if hasattr(self, "_overlay") and hasattr(self, "_video_widget"):
self._overlay.resize(self._video_widget.size())
def closeEvent(self, event) -> None: # noqa: N802
self._camera_service.stop()
super().closeEvent(event)

198
app/ui/menu_bar.py Normal file
View File

@@ -0,0 +1,198 @@
"""Menu bar — camera, video format, FPS and debug controls."""
from __future__ import annotations
import logging
from PySide6.QtCore import Signal
from PySide6.QtGui import QAction, QActionGroup
from PySide6.QtWidgets import QMenuBar, QWidget
from app.camera.camera_enumerator import CameraInfo
logger = logging.getLogger(__name__)
class AppMenuBar(QMenuBar):
"""
Application menu bar.
Signals:
camera_selected(CameraInfo) — user picked a camera
resolution_selected(int, int) — user picked (width, height)
fps_selected(float) — user picked a target FPS
reconnect_requested() — user hit Reconnect
overlay_toggled(bool) — overlay show/hide
log_toggled(bool) — console logging on/off
"""
camera_selected = Signal(object) # CameraInfo
resolution_selected = Signal(int, int)
fps_selected = Signal(float)
reconnect_requested = Signal()
overlay_toggled = Signal(bool)
log_toggled = Signal(bool)
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent)
self._camera_group: QActionGroup | None = None
self._resolution_group: QActionGroup | None = None
self._fps_group: QActionGroup | None = None
self._cameras: list[CameraInfo] = []
self._build_menus()
# ------------------------------------------------------------------
# Public API — called after camera enumeration
# ------------------------------------------------------------------
def populate_cameras(self, cameras: list[CameraInfo]) -> None:
"""Populate the Camera menu with discovered devices."""
self._cameras = cameras
menu = self._camera_menu
# Remove existing camera actions (keep Reconnect + separator)
for action in list(menu.actions()):
if action not in (self._reconnect_action, self._cam_separator):
menu.removeAction(action)
self._camera_group = QActionGroup(self)
self._camera_group.setExclusive(True)
for cam in cameras:
action = QAction(cam.name, self)
action.setCheckable(True)
action.setData(cam)
self._camera_group.addAction(action)
menu.insertAction(self._cam_separator, action)
action.triggered.connect(self._on_camera_action)
if cameras:
first = self._camera_group.actions()[0]
first.setChecked(True)
def populate_formats(self, camera_info: CameraInfo) -> None:
"""Populate Resolution and FPS menus based on a camera's supported formats."""
self._populate_resolutions(camera_info)
self._populate_fps(camera_info)
def set_active_camera(self, camera_info: CameraInfo) -> None:
"""Check the menu item matching camera_info."""
if self._camera_group is None:
return
for action in self._camera_group.actions():
if action.data() is camera_info:
action.setChecked(True)
return
# ------------------------------------------------------------------
# Menu construction
# ------------------------------------------------------------------
def _build_menus(self) -> None:
# --- Camera menu ---
self._camera_menu = self.addMenu("Camera")
self._cam_separator = self._camera_menu.addSeparator()
self._reconnect_action = QAction("Reconnect", self)
self._reconnect_action.triggered.connect(self.reconnect_requested)
self._camera_menu.addAction(self._reconnect_action)
# --- Video menu ---
self._video_menu = self.addMenu("Video")
self._res_menu = self._video_menu.addMenu("Resolution")
self._fps_menu = self._video_menu.addMenu("FPS")
# --- Debug menu ---
debug_menu = self.addMenu("Debug")
self._overlay_action = QAction("Show Overlay", self)
self._overlay_action.setCheckable(True)
self._overlay_action.setChecked(True)
self._overlay_action.toggled.connect(self.overlay_toggled)
debug_menu.addAction(self._overlay_action)
self._log_action = QAction("Console Logging", self)
self._log_action.setCheckable(True)
self._log_action.setChecked(False)
self._log_action.toggled.connect(self._on_log_toggled)
debug_menu.addAction(self._log_action)
def _populate_resolutions(self, camera_info: CameraInfo) -> None:
self._res_menu.clear()
self._resolution_group = QActionGroup(self)
self._resolution_group.setExclusive(True)
seen: set[tuple[int, int]] = set()
for w, h, _ in camera_info.formats:
key = (w, h)
if key in seen:
continue
seen.add(key)
action = QAction(f"{w} × {h}", self)
action.setCheckable(True)
action.setData((w, h))
self._resolution_group.addAction(action)
self._res_menu.addAction(action)
action.triggered.connect(self._on_resolution_action)
actions = self._resolution_group.actions()
if actions:
actions[0].setChecked(True)
def _populate_fps(self, camera_info: CameraInfo) -> None:
self._fps_menu.clear()
self._fps_group = QActionGroup(self)
self._fps_group.setExclusive(True)
seen: set[int] = set()
for _, _, fps in camera_info.formats:
key = round(fps)
if key in seen:
continue
seen.add(key)
action = QAction(f"{key} fps", self)
action.setCheckable(True)
action.setData(float(fps))
self._fps_group.addAction(action)
self._fps_menu.addAction(action)
action.triggered.connect(self._on_fps_action)
actions = self._fps_group.actions()
if actions:
actions[0].setChecked(True)
# ------------------------------------------------------------------
# Slots
# ------------------------------------------------------------------
def _on_camera_action(self) -> None:
action = self.sender()
if action is None:
return
cam: CameraInfo = action.data()
logger.debug("Camera selected: %s", cam.name)
self.camera_selected.emit(cam)
self._populate_resolutions(cam)
self._populate_fps(cam)
def _on_resolution_action(self) -> None:
action = self.sender()
if action is None:
return
w, h = action.data()
logger.debug("Resolution selected: %dx%d", w, h)
self.resolution_selected.emit(w, h)
def _on_fps_action(self) -> None:
action = self.sender()
if action is None:
return
fps: float = action.data()
logger.debug("FPS selected: %.1f", fps)
self.fps_selected.emit(fps)
def _on_log_toggled(self, enabled: bool) -> None:
level = logging.DEBUG if enabled else logging.WARNING
logging.getLogger().setLevel(level)
self.log_toggled.emit(enabled)