From e6ab0a411a66783783e7d812da8077cf5e4824c9 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Tue, 30 Jun 2026 16:40:43 +0200 Subject: [PATCH 01/21] Lock processor settings during DLC inference Disable all DLC and processor configuration widgets consistently while inference is active, including the processor-control checkbox. Refactor processor discovery into shared helpers that detect direct and indirect `dlclive.Processor` subclasses, standardize metadata extraction, and reuse the same fallback logic for package scans and file-based loading. --- dlclivegui/processors/processor_utils.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dlclivegui/processors/processor_utils.py b/dlclivegui/processors/processor_utils.py index e47dbe2f8..e5b62ed97 100644 --- a/dlclivegui/processors/processor_utils.py +++ b/dlclivegui/processors/processor_utils.py @@ -18,7 +18,7 @@ def default_processors_dir() -> str: def _processor_base_class(): - from dlclive.processor import Processor + from dlclive import Processor return Processor From 3b3d5245cbc1f6807a43c17f54279d5b4ed4b778 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Wed, 1 Jul 2026 09:50:45 +0200 Subject: [PATCH 02/21] Improve processor discovery and logging Expand processor class discovery to include re-exported classes by disabling module-only filtering in package/file scans. Also broaden subclass-check error handling to catch unexpected exceptions and log full context when discovery encounters problematic objects. --- dlclivegui/processors/processor_utils.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/dlclivegui/processors/processor_utils.py b/dlclivegui/processors/processor_utils.py index e5b62ed97..d93093b20 100644 --- a/dlclivegui/processors/processor_utils.py +++ b/dlclivegui/processors/processor_utils.py @@ -194,7 +194,8 @@ def load_processors_from_file(file_path: str | Path) -> dict[str, dict]: spec.loader.exec_module(module) # Fallback path: discover subclasses of dlclive.Processor - return discover_processor_classes(module) + # here module only is disabled to allow classes re-exported in other modules to be discovered + return discover_processor_classes(module, only_defined_in_module=False) except Exception: # Full traceback helps a ton when a plugin fails to import From 112c2b7b4c9b8f05d7504bae2efc326f982930e7 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Wed, 1 Jul 2026 10:06:38 +0200 Subject: [PATCH 03/21] Add processors package exports Create `dlclivegui/processors/__init__.py` to re-export `register_processor`, `BaseProcessorSocket`, and `PROCESSOR_REGISTRY` from `dlc_processor_socket`, making these APIs available via package-level imports. --- dlclivegui/processors/__init__.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/dlclivegui/processors/__init__.py b/dlclivegui/processors/__init__.py index 8e7717155..ee94194dd 100644 --- a/dlclivegui/processors/__init__.py +++ b/dlclivegui/processors/__init__.py @@ -1,3 +1,3 @@ -from .registry import PROCESSOR_REGISTRY, register_processor +from .dlc_processor_socket import PROCESSOR_REGISTRY, BaseProcessorSocket, register_processor -__all__ = ["register_processor", "PROCESSOR_REGISTRY"] +__all__ = ["register_processor", "BaseProcessorSocket", "PROCESSOR_REGISTRY"] From b1b176f6f3e310b7aa0ba0b3fec6c27d4aad0d42 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Wed, 1 Jul 2026 10:08:11 +0200 Subject: [PATCH 04/21] Move example socket processors to examples module Refactors `dlc_processor_socket.py` by removing the in-file example processors and `OneEuroFilter`, and adds them to a new `dlclivegui/processors/examples.py` module. This separates demonstration/experiment-specific logic from the core socket processor implementation, improving maintainability while preserving existing example processor behavior. --- dlclivegui/processors/dlc_processor_socket.py | 49 +++++++++++++++++++ dlclivegui/processors/examples.py | 3 +- 2 files changed, 50 insertions(+), 2 deletions(-) diff --git a/dlclivegui/processors/dlc_processor_socket.py b/dlclivegui/processors/dlc_processor_socket.py index 6a91ef1a3..2c120c35c 100644 --- a/dlclivegui/processors/dlc_processor_socket.py +++ b/dlclivegui/processors/dlc_processor_socket.py @@ -23,6 +23,20 @@ _handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")) logger.addHandler(_handler) +# Registry for GUI discovery +PROCESSOR_REGISTRY = {} + + +def register_processor(cls): + registry_key = getattr(cls, "PROCESSOR_ID", cls.__name__) + if registry_key in PROCESSOR_REGISTRY: + raise ValueError( + f"Duplicate processor registration key '{registry_key}': " + f"{PROCESSOR_REGISTRY[registry_key].__name__} vs {cls.__name__}" + ) + PROCESSOR_REGISTRY[registry_key] = cls + return cls + # pragma: cover class BaseProcessorSocket(Processor): @@ -421,3 +435,38 @@ def get_data(self): if self.dlc_cfg is not None: save_dict["dlc_cfg"] = self.dlc_cfg return save_dict + + +def get_available_processors(): + """ + Get list of available processor classes. + + Returns: + dict: Dictionary mapping registry keys to processor info. + """ + return { + name: { + "class": cls, + "name": getattr(cls, "PROCESSOR_NAME", name), + "description": getattr(cls, "PROCESSOR_DESCRIPTION", ""), + "params": getattr(cls, "PROCESSOR_PARAMS", {}), + } + for name, cls in PROCESSOR_REGISTRY.items() + } + + +def instantiate_processor(class_name, **kwargs): + """ + Instantiate a processor by class name with given parameters. + + Args: + class_name: Registry key (e.g., "MyProcessorSocket") + **kwargs: Constructor kwargs + + Raises: + ValueError: If class_name is not in registry + """ + if class_name not in PROCESSOR_REGISTRY: + available = ", ".join(PROCESSOR_REGISTRY.keys()) + raise ValueError(f"Unknown processor '{class_name}'. Available: {available}") + return PROCESSOR_REGISTRY[class_name](**kwargs) diff --git a/dlclivegui/processors/examples.py b/dlclivegui/processors/examples.py index 7e96fc068..60c5f8421 100644 --- a/dlclivegui/processors/examples.py +++ b/dlclivegui/processors/examples.py @@ -6,8 +6,7 @@ import numpy as np -from dlclivegui.processors import register_processor -from dlclivegui.processors.dlc_processor_socket import BaseProcessorSocket +from dlclivegui.processors import BaseProcessorSocket, register_processor logger = logging.getLogger(__name__) From fa84f8ce223584de15080fae6d709797119c9b4f Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Wed, 1 Jul 2026 10:12:22 +0200 Subject: [PATCH 05/21] Skip socket base module in processor scan Update processor package discovery to ignore `dlc_processor_socket` during namespace scanning, since it only provides the base class/registry and should not be listed as an available processor source. The package fallback scan now uses default class discovery behavior, and related outdated comments/docstring lines were cleaned up. --- dlclivegui/processors/processor_utils.py | 1 - 1 file changed, 1 deletion(-) diff --git a/dlclivegui/processors/processor_utils.py b/dlclivegui/processors/processor_utils.py index d93093b20..e631a44a0 100644 --- a/dlclivegui/processors/processor_utils.py +++ b/dlclivegui/processors/processor_utils.py @@ -194,7 +194,6 @@ def load_processors_from_file(file_path: str | Path) -> dict[str, dict]: spec.loader.exec_module(module) # Fallback path: discover subclasses of dlclive.Processor - # here module only is disabled to allow classes re-exported in other modules to be discovered return discover_processor_classes(module, only_defined_in_module=False) except Exception: From 1b147d64ae9a9c825cafd44ca7fa15c0dd61654e Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Wed, 1 Jul 2026 10:13:13 +0200 Subject: [PATCH 06/21] Update processor_utils.py --- dlclivegui/processors/processor_utils.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dlclivegui/processors/processor_utils.py b/dlclivegui/processors/processor_utils.py index e631a44a0..e5b62ed97 100644 --- a/dlclivegui/processors/processor_utils.py +++ b/dlclivegui/processors/processor_utils.py @@ -194,7 +194,7 @@ def load_processors_from_file(file_path: str | Path) -> dict[str, dict]: spec.loader.exec_module(module) # Fallback path: discover subclasses of dlclive.Processor - return discover_processor_classes(module, only_defined_in_module=False) + return discover_processor_classes(module) except Exception: # Full traceback helps a ton when a plugin fails to import From 5c196ef8eda5b0462b48b0ce2370fee915adbd5c Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Wed, 1 Jul 2026 10:21:40 +0200 Subject: [PATCH 07/21] Warn on duplicate processor registration Change `register_processor` to log a warning instead of raising on duplicate `PROCESSOR_ID` keys, allowing later registrations to override earlier ones without import-time failures. Update subclass save tests to load processor classes from `dlclivegui.processors.examples` via a dedicated fixture, so the parametrized tests validate the concrete example processors against the correct module data path. --- dlclivegui/processors/dlc_processor_socket.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/dlclivegui/processors/dlc_processor_socket.py b/dlclivegui/processors/dlc_processor_socket.py index 2c120c35c..80fa72291 100644 --- a/dlclivegui/processors/dlc_processor_socket.py +++ b/dlclivegui/processors/dlc_processor_socket.py @@ -30,10 +30,11 @@ def register_processor(cls): registry_key = getattr(cls, "PROCESSOR_ID", cls.__name__) if registry_key in PROCESSOR_REGISTRY: - raise ValueError( + msg = ( f"Duplicate processor registration key '{registry_key}': " f"{PROCESSOR_REGISTRY[registry_key].__name__} vs {cls.__name__}" ) + logger.warning(msg) PROCESSOR_REGISTRY[registry_key] = cls return cls From 403127a992fdfb1a34e47e1c95b656791ad969a8 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Wed, 1 Jul 2026 10:40:46 +0200 Subject: [PATCH 08/21] Fix dlclive Processor import paths Update processor imports to use `from dlclive.processor import Processor` in runtime code to avoid torch import side effects --- dlclivegui/processors/processor_utils.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dlclivegui/processors/processor_utils.py b/dlclivegui/processors/processor_utils.py index e5b62ed97..e47dbe2f8 100644 --- a/dlclivegui/processors/processor_utils.py +++ b/dlclivegui/processors/processor_utils.py @@ -18,7 +18,7 @@ def default_processors_dir() -> str: def _processor_base_class(): - from dlclive import Processor + from dlclive.processor import Processor return Processor From 57bf911618a34172ac27fbba23b3a3fd79555893 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Wed, 1 Jul 2026 10:49:10 +0200 Subject: [PATCH 09/21] Extract processor registry into new module Moves processor registration and discovery helpers out of `dlc_processor_socket.py` into a new `registry.py` module so registry access no longer depends on importing socket logic. `dlc_processor_socket.py` now imports the shared registry helpers and adds a safe fallback when `dlclive` is unavailable, reducing import-time failures in environments without that dependency. Package exports were updated to expose registry APIs from the new module. --- dlclivegui/processors/__init__.py | 4 +- dlclivegui/processors/dlc_processor_socket.py | 50 ------------------- dlclivegui/processors/examples.py | 3 +- 3 files changed, 4 insertions(+), 53 deletions(-) diff --git a/dlclivegui/processors/__init__.py b/dlclivegui/processors/__init__.py index ee94194dd..8e7717155 100644 --- a/dlclivegui/processors/__init__.py +++ b/dlclivegui/processors/__init__.py @@ -1,3 +1,3 @@ -from .dlc_processor_socket import PROCESSOR_REGISTRY, BaseProcessorSocket, register_processor +from .registry import PROCESSOR_REGISTRY, register_processor -__all__ = ["register_processor", "BaseProcessorSocket", "PROCESSOR_REGISTRY"] +__all__ = ["register_processor", "PROCESSOR_REGISTRY"] diff --git a/dlclivegui/processors/dlc_processor_socket.py b/dlclivegui/processors/dlc_processor_socket.py index 80fa72291..6a91ef1a3 100644 --- a/dlclivegui/processors/dlc_processor_socket.py +++ b/dlclivegui/processors/dlc_processor_socket.py @@ -23,21 +23,6 @@ _handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")) logger.addHandler(_handler) -# Registry for GUI discovery -PROCESSOR_REGISTRY = {} - - -def register_processor(cls): - registry_key = getattr(cls, "PROCESSOR_ID", cls.__name__) - if registry_key in PROCESSOR_REGISTRY: - msg = ( - f"Duplicate processor registration key '{registry_key}': " - f"{PROCESSOR_REGISTRY[registry_key].__name__} vs {cls.__name__}" - ) - logger.warning(msg) - PROCESSOR_REGISTRY[registry_key] = cls - return cls - # pragma: cover class BaseProcessorSocket(Processor): @@ -436,38 +421,3 @@ def get_data(self): if self.dlc_cfg is not None: save_dict["dlc_cfg"] = self.dlc_cfg return save_dict - - -def get_available_processors(): - """ - Get list of available processor classes. - - Returns: - dict: Dictionary mapping registry keys to processor info. - """ - return { - name: { - "class": cls, - "name": getattr(cls, "PROCESSOR_NAME", name), - "description": getattr(cls, "PROCESSOR_DESCRIPTION", ""), - "params": getattr(cls, "PROCESSOR_PARAMS", {}), - } - for name, cls in PROCESSOR_REGISTRY.items() - } - - -def instantiate_processor(class_name, **kwargs): - """ - Instantiate a processor by class name with given parameters. - - Args: - class_name: Registry key (e.g., "MyProcessorSocket") - **kwargs: Constructor kwargs - - Raises: - ValueError: If class_name is not in registry - """ - if class_name not in PROCESSOR_REGISTRY: - available = ", ".join(PROCESSOR_REGISTRY.keys()) - raise ValueError(f"Unknown processor '{class_name}'. Available: {available}") - return PROCESSOR_REGISTRY[class_name](**kwargs) diff --git a/dlclivegui/processors/examples.py b/dlclivegui/processors/examples.py index 60c5f8421..7e96fc068 100644 --- a/dlclivegui/processors/examples.py +++ b/dlclivegui/processors/examples.py @@ -6,7 +6,8 @@ import numpy as np -from dlclivegui.processors import BaseProcessorSocket, register_processor +from dlclivegui.processors import register_processor +from dlclivegui.processors.dlc_processor_socket import BaseProcessorSocket logger = logging.getLogger(__name__) From dd18dce8e3e82bce65d877cf268ce11a59ecf65e Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 2 Jul 2026 16:28:04 +0200 Subject: [PATCH 10/21] Refactor camera worker and recording pipeline Extracts `SingleCameraWorker` into a new `services/camera_controller.py` module and moves per-frame rotation/cropping plus recording-sink writes into the worker thread. `MultiCameraController` now injects and updates a shared recording sink on workers, and recording enable/disable is propagated directly to active workers. The previous recording-frame emission path and transform handling in the multi-camera slot are removed/commented out to avoid duplicate processing and keep the controller focused on frame aggregation. Also relocates `recording_manager.py` from `gui` to `services` with a file rename. --- dlclivegui/services/camera_controller.py | 255 ++++++++++++++++++ .../services/multi_camera_controller.py | 248 ++--------------- .../{gui => services}/recording_manager.py | 0 3 files changed, 281 insertions(+), 222 deletions(-) create mode 100644 dlclivegui/services/camera_controller.py rename dlclivegui/{gui => services}/recording_manager.py (100%) diff --git a/dlclivegui/services/camera_controller.py b/dlclivegui/services/camera_controller.py new file mode 100644 index 000000000..428503cfb --- /dev/null +++ b/dlclivegui/services/camera_controller.py @@ -0,0 +1,255 @@ +from __future__ import annotations + +import copy +import logging +import time +from threading import Event, Lock + +import cv2 +import numpy as np +from PySide6.QtCore import QObject, Signal, Slot + +from dlclivegui.cameras import CameraFactory +from dlclivegui.cameras.base import CameraBackend + +# from dlclivegui.config import CameraSettings +from dlclivegui.config import ( + SINGLE_CAMERA_WORKER_DO_LOG_TIMING, + CameraSettings, +) +from dlclivegui.utils.stats import WorkerTimingStats + +logger = logging.getLogger(__name__) + + +class SingleCameraWorker(QObject): + """Worker for a single camera in multi-camera mode.""" + + frame_captured = Signal(str, object, float) # camera_id, frame, timestamp + error_occurred = Signal(str, str) # camera_id, error_message + runtime_info = Signal(str, object) # camera_id, dict of runtime info + started = Signal(str) # camera_id + stopped = Signal(str) # camera_id + + def __init__(self, camera_id: str, settings: CameraSettings): + super().__init__() + self._camera_id = camera_id + self._settings = copy.deepcopy(settings) + self._stop_event = Event() + self._backend: CameraBackend | None = None + self._max_consecutive_errors = 5 + self._retry_delay = 0.1 + self._trigger_timeout_delay = 0.05 + self._trigger_wait_log_interval = 2.0 + self._last_trigger_wait_log = 0.0 + self._trigger_wait_suppressed_count = 0 + + self._recording_sink = None + self._recording_enabled = False + self._recording_sink_lock = Lock() + + # Performance logs + self._timing = WorkerTimingStats( + camera_id, logger=logger, log_interval=1.0, enabled=SINGLE_CAMERA_WORKER_DO_LOG_TIMING + ) + + def set_recording_sink(self, sink) -> None: + with self._recording_sink_lock: + self._recording_sink = sink + + def set_recording_enabled(self, enabled: bool) -> None: + with self._recording_sink_lock: + self._recording_enabled = bool(enabled) + + @Slot() + def run(self) -> None: + self._stop_event.clear() + + try: + logger.debug( + "[Worker %s] before create: backend=%s index=%s properties=%s", + self._camera_id, + self._settings.backend, + self._settings.index, + self._settings.properties, + ) + + self._backend = CameraFactory.create(self._settings) + + logger.debug( + "[Worker %s] after create: backend=%s index=%s properties=%s", + self._camera_id, + self._backend.settings.backend, + self._backend.settings.index, + self._backend.settings.properties, + ) + + self._backend.open() + self.runtime_info.emit( + self._camera_id, + { + "actual_fps": getattr(self._backend, "actual_fps", None), + "actual_resolution": getattr(self._backend, "actual_resolution", None), + "actual_pixel_format": getattr(self._backend, "actual_pixel_format", None), + "actual_output_format": getattr(self._backend, "actual_output_format", None), + }, + ) + except Exception as exc: + logger.exception(f"Failed to initialize camera {self._camera_id}", exc_info=exc) + self.error_occurred.emit(self._camera_id, f"Failed to initialize camera: {exc}") + self.stopped.emit(self._camera_id) + return + + self.started.emit(self._camera_id) + consecutive_errors = 0 + + while not self._stop_event.is_set(): + try: + with self._timing.measure("Single.read"): + frame, timestamp = self._backend.read() + if frame is None or frame.size == 0: + consecutive_errors += 1 + if consecutive_errors >= self._max_consecutive_errors: + self.error_occurred.emit( + self._camera_id, "Too many empty frames.\nWas the device disconnected ?" + ) + break + if self._stop_event.wait(self._retry_delay): + break + continue + + consecutive_errors = 0 + with self._timing.measure("Single.transforms"): + frame = self._apply_worker_transforms(frame) + + with self._recording_sink_lock: + recording_enabled = self._recording_enabled + recording_sink = self._recording_sink + + if recording_enabled and recording_sink is not None: + try: + with self._timing.measure("Single.recording_sink"): + recording_sink(self._camera_id, frame, timestamp) + except Exception as exc: + logger.exception(f"Failed to write frame for camera {self._camera_id}: {exc}") + + with self._timing.measure("Single.emit"): + self.frame_captured.emit(self._camera_id, frame, timestamp) + + self._timing.note_frame() + self._timing.maybe_log() + + except TimeoutError as exc: + self._timing.note_timeout() + self._timing.maybe_log() + if self._stop_event.is_set(): + break + + # In hardware-trigger mode, a timeout usually means: + # "no trigger pulse arrived during this poll interval". + # This is expected and should not count as a camera failure. + if bool(getattr(self._backend, "waits_for_hardware_trigger", False)): + self._log_trigger_wait_throttled(exc) + consecutive_errors = 0 + + if self._stop_event.wait(self._trigger_timeout_delay): + break # Stop event set during wait + continue + + consecutive_errors += 1 + if consecutive_errors >= self._max_consecutive_errors: + self.error_occurred.emit(self._camera_id, f"Camera read timeout: {exc}") + break + if self._stop_event.wait(self._retry_delay): + break + continue + + except Exception as exc: + self._timing.note_error() + self._timing.maybe_log() + consecutive_errors += 1 + if self._stop_event.is_set(): + break + if consecutive_errors >= self._max_consecutive_errors: + self.error_occurred.emit(self._camera_id, f"Camera read error: {exc}") + break + if self._stop_event.wait(self._retry_delay): + break + continue + + # Cleanup + if self._backend is not None: + try: + self._backend.close() + except Exception: + pass + self.stopped.emit(self._camera_id) + + def stop(self) -> None: + self._stop_event.set() + + @staticmethod + def apply_rotation(frame: np.ndarray, degrees: int) -> np.ndarray: + """Apply rotation to frame.""" + if degrees == 90: + return cv2.rotate(frame, cv2.ROTATE_90_CLOCKWISE) + elif degrees == 180: + return cv2.rotate(frame, cv2.ROTATE_180) + elif degrees == 270: + return cv2.rotate(frame, cv2.ROTATE_90_COUNTERCLOCKWISE) + return frame + + @staticmethod + def apply_crop(frame: np.ndarray, crop_region: tuple[int, int, int, int]) -> np.ndarray: + """Apply crop to frame.""" + x0, y0, x1, y1 = crop_region + height, width = frame.shape[:2] + + x0 = max(0, min(x0, width)) + y0 = max(0, min(y0, height)) + x1 = max(x0, min(x1, width)) if x1 > 0 else width + y1 = max(y0, min(y1, height)) if y1 > 0 else height + + if x0 < x1 and y0 < y1: + return frame[y0:y1, x0:x1] + return frame + + def _apply_worker_transforms(self, frame: np.ndarray) -> np.ndarray: + if self._settings.rotation: + frame = self.apply_rotation(frame, self._settings.rotation) + + crop_region = self._settings.get_crop_region() + if crop_region: + frame = self.apply_crop(frame, crop_region) + + return frame + + def _log_trigger_wait_throttled(self, exc: BaseException) -> None: + """Log hardware-trigger wait timeouts at a controlled rate. + + In trigger-waiting modes, read timeouts are expected polling misses. + Without throttling, the log can be flooded at ~10-20 messages/sec/camera. + """ + now = time.monotonic() + + if now - self._last_trigger_wait_log < self._trigger_wait_log_interval: + self._trigger_wait_suppressed_count += 1 + return + + suppressed = self._trigger_wait_suppressed_count + self._trigger_wait_suppressed_count = 0 + self._last_trigger_wait_log = now + + if suppressed: + logger.debug( + "[Worker %s] waiting for hardware trigger: %s (suppressed %d repeated timeout logs)", + self._camera_id, + exc, + suppressed, + ) + else: + logger.debug( + "[Worker %s] waiting for hardware trigger: %s", + self._camera_id, + exc, + ) diff --git a/dlclivegui/services/multi_camera_controller.py b/dlclivegui/services/multi_camera_controller.py index 997d0d947..84c374ab8 100644 --- a/dlclivegui/services/multi_camera_controller.py +++ b/dlclivegui/services/multi_camera_controller.py @@ -7,27 +7,26 @@ import time from dataclasses import dataclass from functools import partial -from threading import Event, Lock +from threading import Lock import cv2 import numpy as np -from PySide6.QtCore import QObject, QThread, Signal, Slot +from PySide6.QtCore import QObject, QThread, Signal from PySide6.QtGui import QImage, QPixmap -from dlclivegui.cameras import CameraFactory -from dlclivegui.cameras.base import CameraBackend from dlclivegui.cameras.factory import camera_identity_key # from dlclivegui.config import CameraSettings from dlclivegui.config import ( GUI_MAX_DISPLAY_FPS, MULTI_CAMERA_WORKER_DO_LOG_TIMING, - SINGLE_CAMERA_WORKER_DO_LOG_TIMING, CameraSettings, CameraTriggerSettings, ) from dlclivegui.utils.stats import WorkerTimingStats +from .camera_controller import SingleCameraWorker + LOGGER = logging.getLogger(__name__) QUIT_WAIT_MS = 5000 # wait for cooperative quit (5s) @@ -45,181 +44,6 @@ class MultiFrameData: display_ids: dict[str, str] = None # camera_id -> display_id (for labeling) -class SingleCameraWorker(QObject): - """Worker for a single camera in multi-camera mode.""" - - frame_captured = Signal(str, object, float, object) # camera_id, frame, timestamp, timestamp_metadata - error_occurred = Signal(str, str) # camera_id, error_message - runtime_info = Signal(str, object) # camera_id, dict of runtime info - started = Signal(str) # camera_id - stopped = Signal(str) # camera_id - - def __init__(self, camera_id: str, settings: CameraSettings): - super().__init__() - self._camera_id = camera_id - self._settings = copy.deepcopy(settings) - self._stop_event = Event() - self._backend: CameraBackend | None = None - self._max_consecutive_errors = 5 - self._retry_delay = 0.1 - self._trigger_timeout_delay = 0.05 - - self._trigger_wait_log_interval = 2.0 - self._last_trigger_wait_log = 0.0 - self._trigger_wait_suppressed_count = 0 - - # Performance logs - self._timing = WorkerTimingStats( - camera_id, logger=LOGGER, log_interval=1.0, enabled=SINGLE_CAMERA_WORKER_DO_LOG_TIMING - ) - - @Slot() - def run(self) -> None: - self._stop_event.clear() - - try: - LOGGER.debug( - "[Worker %s] before create: backend=%s index=%s properties=%s", - self._camera_id, - self._settings.backend, - self._settings.index, - self._settings.properties, - ) - - self._backend = CameraFactory.create(self._settings) - - LOGGER.debug( - "[Worker %s] after create: backend=%s index=%s properties=%s", - self._camera_id, - self._backend.settings.backend, - self._backend.settings.index, - self._backend.settings.properties, - ) - - self._backend.open() - self.runtime_info.emit( - self._camera_id, - { - "actual_fps": getattr(self._backend, "actual_fps", None), - "actual_resolution": getattr(self._backend, "actual_resolution", None), - "actual_pixel_format": getattr(self._backend, "actual_pixel_format", None), - "actual_output_format": getattr(self._backend, "actual_output_format", None), - }, - ) - except Exception as exc: - LOGGER.exception(f"Failed to initialize camera {self._camera_id}", exc_info=exc) - self.error_occurred.emit(self._camera_id, f"Failed to initialize camera: {exc}") - self.stopped.emit(self._camera_id) - return - - self.started.emit(self._camera_id) - consecutive_errors = 0 - - while not self._stop_event.is_set(): - try: - with self._timing.measure("Single.read"): - captured = self._backend.read() - frame = captured.frame - timestamp = captured.software_timestamp - timestamp_metadata = captured.timestamp_metadata - if frame is None or frame.size == 0: - consecutive_errors += 1 - if consecutive_errors >= self._max_consecutive_errors: - self.error_occurred.emit( - self._camera_id, "Too many empty frames.\nWas the device disconnected ?" - ) - break - if self._stop_event.wait(self._retry_delay): - break - continue - - consecutive_errors = 0 - with self._timing.measure("Single.emit.frame_captured"): - self.frame_captured.emit(self._camera_id, frame, timestamp, timestamp_metadata) - - self._timing.note_frame() - self._timing.maybe_log() - - except TimeoutError as exc: - self._timing.note_timeout() - self._timing.maybe_log() - if self._stop_event.is_set(): - break - - # In hardware-trigger mode, a timeout usually means: - # "no trigger pulse arrived during this poll interval". - # This is expected and should not count as a camera failure. - if bool(getattr(self._backend, "waits_for_hardware_trigger", False)): - self._log_trigger_wait_throttled(exc) - consecutive_errors = 0 - - if self._stop_event.wait(self._trigger_timeout_delay): - break # Stop event set during wait - continue - - consecutive_errors += 1 - if consecutive_errors >= self._max_consecutive_errors: - self.error_occurred.emit(self._camera_id, f"Camera read timeout: {exc}") - break - if self._stop_event.wait(self._retry_delay): - break - continue - - except Exception as exc: - self._timing.note_error() - self._timing.maybe_log() - consecutive_errors += 1 - if self._stop_event.is_set(): - break - if consecutive_errors >= self._max_consecutive_errors: - self.error_occurred.emit(self._camera_id, f"Camera read error: {exc}") - break - if self._stop_event.wait(self._retry_delay): - break - continue - - # Cleanup - if self._backend is not None: - try: - self._backend.close() - except Exception: - pass - self.stopped.emit(self._camera_id) - - def stop(self) -> None: - self._stop_event.set() - - def _log_trigger_wait_throttled(self, exc: BaseException) -> None: - """Log hardware-trigger wait timeouts at a controlled rate. - - In trigger-waiting modes, read timeouts are expected polling misses. - Without throttling, the log can be flooded at ~10-20 messages/sec/camera. - """ - now = time.monotonic() - - if now - self._last_trigger_wait_log < self._trigger_wait_log_interval: - self._trigger_wait_suppressed_count += 1 - return - - suppressed = self._trigger_wait_suppressed_count - self._trigger_wait_suppressed_count = 0 - self._last_trigger_wait_log = now - - if suppressed: - LOGGER.debug( - "[Worker %s] waiting for hardware trigger: %s (suppressed %d repeated timeout logs)", - self._camera_id, - exc, - suppressed, - ) - else: - LOGGER.debug( - "[Worker %s] waiting for hardware trigger: %s", - self._camera_id, - exc, - ) - - def get_display_id(settings: CameraSettings) -> str: """Return the human-friendly camera label used for GUI display. Intentionally different from get_camera_id(), which should return a stable @@ -312,6 +136,7 @@ def __init__(self): self._stopping = False self._all_stopped_emitted = False self._recording_frame_emission_enabled: bool = False + self._recording_sink = None self._started_cameras: set = set() self._display_ids: dict[str, str] = {} # camera_id -> display_id (for labeling) self._camera_display_order: list[str] = [] @@ -345,12 +170,9 @@ def _timing_for_camera(self, camera_id: str) -> WorkerTimingStats: return timing def set_recording_frame_do_emit(self, enabled: bool) -> None: - """Enable/disable the lightweight per-camera recording frame signal. - - This avoids sending recording-only traffic when the user is only previewing - or running DLC. - """ self._recording_frame_emission_enabled = bool(enabled) + for worker in list(self._workers.values()): + worker.set_recording_enabled(enabled) def _should_emit_display_ready(self) -> bool: """Return True when the UI/display path should be updated. @@ -456,6 +278,8 @@ def _start_camera(self, settings: CameraSettings) -> None: self._display_ids[cam_id] = display_id dc = self._settings[cam_id] worker = SingleCameraWorker(cam_id, dc) + worker.set_recording_sink(self._recording_sink) + worker.set_recording_enabled(self._recording_frame_emission_enabled) thread = QThread() worker.moveToThread(thread) @@ -473,7 +297,13 @@ def _start_camera(self, settings: CameraSettings) -> None: worker.stopped.connect(thread.quit) thread.start() + def set_recording_sink(self, sink) -> None: + self._recording_sink = sink + for worker in list(self._workers.values()): + worker.set_recording_sink(sink) + def _cleanup_camera(self, camera_id: str, *, finalize: bool = True) -> None: + # remove stored frame data with self._frame_lock: self._frames.pop(camera_id, None) self._timestamps.pop(camera_id, None) @@ -608,20 +438,20 @@ def _on_frame_captured( frame_data: MultiFrameData | None = None with timing.measure("Multi.slot.total"): - settings = self._settings.get(camera_id) + self._settings.get(camera_id) - with timing.measure("Multi.apply_transforms"): - if settings and settings.rotation: - frame = MultiCameraController.apply_rotation(frame, settings.rotation) + # with timing.measure("Multi.apply_transforms"): + # if settings and settings.rotation: + # frame = MultiCameraController.apply_rotation(frame, settings.rotation) - if settings: - crop_region = settings.get_crop_region() - if crop_region: - frame = MultiCameraController.apply_crop(frame, crop_region) + # if settings: + # crop_region = settings.get_crop_region() + # if crop_region: + # frame = MultiCameraController.apply_crop(frame, crop_region) - if self._recording_frame_emission_enabled: - with timing.measure("Multi.emit.recording_frame_ready"): - self.recording_frame_ready.emit(camera_id, frame, timestamp, timestamp_metadata) + # if self._recording_frame_emission_enabled: + # with timing.measure("Multi.emit.recording_frame_ready"): + # self.recording_frame_ready.emit(camera_id, frame, timestamp) with self._frame_lock: with timing.measure("Multi.store_latest"): @@ -697,32 +527,6 @@ def actual_fps_by_camera_id(self) -> dict[str, float]: return out - @staticmethod - def apply_rotation(frame: np.ndarray, degrees: int) -> np.ndarray: - """Apply rotation to frame.""" - if degrees == 90: - return cv2.rotate(frame, cv2.ROTATE_90_CLOCKWISE) - elif degrees == 180: - return cv2.rotate(frame, cv2.ROTATE_180) - elif degrees == 270: - return cv2.rotate(frame, cv2.ROTATE_90_COUNTERCLOCKWISE) - return frame - - @staticmethod - def apply_crop(frame: np.ndarray, crop_region: tuple[int, int, int, int]) -> np.ndarray: - """Apply crop to frame.""" - x0, y0, x1, y1 = crop_region - height, width = frame.shape[:2] - - x0 = max(0, min(x0, width)) - y0 = max(0, min(y0, height)) - x1 = max(x0, min(x1, width)) if x1 > 0 else width - y1 = max(y0, min(y1, height)) if y1 > 0 else height - - if x0 < x1 and y0 < y1: - return frame[y0:y1, x0:x1] - return frame - @staticmethod def apply_resize(frame: np.ndarray, max_w: int, max_h: int, allow_upscale: bool = False) -> np.ndarray: """Resize frame to fit within max dimensions while maintaining aspect ratio.""" diff --git a/dlclivegui/gui/recording_manager.py b/dlclivegui/services/recording_manager.py similarity index 100% rename from dlclivegui/gui/recording_manager.py rename to dlclivegui/services/recording_manager.py From 4715134088c43f22f9d438309c915a38a7e9c652 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 2 Jul 2026 16:28:23 +0200 Subject: [PATCH 11/21] Route recording frames through recording sink Switch `RecordingManager` import to the new `services` package path and update recording flow to use an explicit recording sink callback. Recording now sets `multi_camera_controller.set_recording_sink(self._rec_manager.write_frame)` when starting and clears it on stop, replacing the previous direct `recording_frame_ready` signal hookup. --- dlclivegui/gui/main_window.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/dlclivegui/gui/main_window.py b/dlclivegui/gui/main_window.py index ed6e3ee7c..4af17d600 100644 --- a/dlclivegui/gui/main_window.py +++ b/dlclivegui/gui/main_window.py @@ -71,6 +71,7 @@ ) from ..services.dlc_processor import DLCLiveProcessor, PoseResult from ..services.multi_camera_controller import MultiCameraController, MultiFrameData, get_camera_id, get_display_id +from ..services.recording_manager import RecordingManager from ..utils.display import BBoxColors, compute_tile_info, create_tiled_frame, draw_bbox, draw_pose from ..utils.settings_store import DLCLiveGUISettingsStore, ModelPathStore from ..utils.stats import WorkerTimingStats, format_dlc_stats @@ -80,7 +81,6 @@ from .misc import layouts as lyts from .misc.drag_spinbox import ScrubSpinBox from .misc.eliding_label import ElidingPathLabel -from .recording_manager import RecordingManager from .theme import LOGO, LOGO_ALPHA, AppStyle, apply_theme logger = logging.getLogger("DLCLiveGUI") @@ -812,7 +812,7 @@ def _connect_signals(self) -> None: # Multi-camera controller signals (used for both single and multi-camera modes) self.multi_camera_controller.frame_ready.connect(self._on_multi_frame_processing_ready) self.multi_camera_controller.display_ready.connect(self._on_multi_frame_display_ready) - self.multi_camera_controller.recording_frame_ready.connect(self._on_recording_frame_ready) + # self.multi_camera_controller.recording_frame_ready.connect(self._on_recording_frame_ready) self.multi_camera_controller.all_started.connect(self._on_multi_camera_started) self.multi_camera_controller.all_stopped.connect(self._on_multi_camera_stopped) self.multi_camera_controller.camera_error.connect(self._on_multi_camera_error) @@ -1621,6 +1621,7 @@ def _start_multi_camera_recording(self) -> None: if run_dir is None: self._show_error("Failed to start recording.") return + self.multi_camera_controller.set_recording_sink(self._rec_manager.write_frame) self.multi_camera_controller.set_recording_frame_do_emit(True) self._settings_store.set_session_name(session_name) @@ -1645,6 +1646,7 @@ def _stop_multi_camera_recording(self) -> None: # Stop frame emission immediately so no new frames enter recording pipeline. try: self.multi_camera_controller.set_recording_frame_do_emit(False) + self.multi_camera_controller.set_recording_sink(None) except Exception: logger.exception("Failed to disable recording frame emission") From 1ebd56d8774f5301cd2d7a62c3c88f9ef83fc15b Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 2 Jul 2026 16:28:39 +0200 Subject: [PATCH 12/21] Update recording manager test imports Adjust test fixtures and GUI recording manager tests to import `RecordingManager` and related module symbols from `dlclivegui.services.recording_manager` instead of the old `dlclivegui.gui.recording_manager` path. --- tests/conftest.py | 6 +++--- tests/gui/test_rec_manager.py | 4 ++-- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 25c5567e2..20752cd8a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -325,7 +325,7 @@ def _fake_start_all(self, recording, active_cams, current_frames, **kwargs): run_dir.mkdir(parents=True, exist_ok=True) return run_dir - from dlclivegui.gui import recording_manager as rm_mod + from dlclivegui.services import recording_manager as rm_mod monkeypatch.setattr(rm_mod.RecordingManager, "start_all", _fake_start_all) return calls @@ -409,7 +409,7 @@ def recording_settings(app_config_two_cams): @pytest.fixture def patch_video_recorder(monkeypatch): - import dlclivegui.gui.recording_manager as rm_mod + import dlclivegui.services.recording_manager as rm_mod monkeypatch.setattr(rm_mod, "VideoRecorder", FakeVideoRecorder) return FakeVideoRecorder @@ -428,7 +428,7 @@ def _fake_write_frame(cam_id, frame, timestamp=None, timestamp_metadata=None): @pytest.fixture def patch_build_run_dir(monkeypatch, tmp_path): - import dlclivegui.gui.recording_manager as rm_mod + import dlclivegui.services.recording_manager as rm_mod spy = {"session_dir": None, "use_timestamp": None} run_dir = tmp_path / "videos" / "Sess_SANITIZED" / "run_TEST" diff --git a/tests/gui/test_rec_manager.py b/tests/gui/test_rec_manager.py index 45576716c..914132d35 100644 --- a/tests/gui/test_rec_manager.py +++ b/tests/gui/test_rec_manager.py @@ -4,8 +4,8 @@ import pytest from dlclivegui.config import CameraSettings -from dlclivegui.gui.recording_manager import RecordingManager from dlclivegui.services.multi_camera_controller import get_camera_id, get_display_id +from dlclivegui.services.recording_manager import RecordingManager from dlclivegui.utils.stats import RecorderStats from dlclivegui.utils.timestamps import FrameTimestampMetadata @@ -214,7 +214,7 @@ def test_write_frame_uses_time_when_timestamp_missing( mgr = RecordingManager() mgr.start_all(recording_settings, _active_cams_two, current_frames, session_name="Sess") - import dlclivegui.gui.recording_manager as rm_mod # noqa: E402 + import dlclivegui.services.recording_manager as rm_mod # noqa: E402 monkeypatch.setattr(rm_mod.time, "time", lambda: 999.0) From c6f1c4f1cb30d6070d1b00d10f0bddcf15071405 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 2 Jul 2026 17:31:45 +0200 Subject: [PATCH 13/21] Propagate capture metadata in single-camera flow Update the single-camera pipeline to use structured capture results from backend reads, extracting `frame`, `software_timestamp`, and `timestamp_metadata` and forwarding metadata through frame emission and recording writes. Recording queue handling now expects the metadata field as well. Preview transform helpers were also switched to reuse `SingleCameraWorker` crop/rotation methods instead of `MultiCameraController`. --- dlclivegui/gui/camera_config/preview.py | 5 +++-- dlclivegui/services/camera_controller.py | 9 ++++++--- dlclivegui/services/recording_manager.py | 4 ++-- 3 files changed, 11 insertions(+), 7 deletions(-) diff --git a/dlclivegui/gui/camera_config/preview.py b/dlclivegui/gui/camera_config/preview.py index bbd1aef0d..ef2c7570b 100644 --- a/dlclivegui/gui/camera_config/preview.py +++ b/dlclivegui/gui/camera_config/preview.py @@ -7,6 +7,7 @@ from PySide6.QtCore import QTimer +from ...services.camera_controller import SingleCameraWorker from ...services.multi_camera_controller import MultiCameraController if TYPE_CHECKING: @@ -56,7 +57,7 @@ class PreviewSession: def apply_rotation(frame, rotation): - return MultiCameraController.apply_rotation(frame, rotation) + return SingleCameraWorker.apply_rotation(frame, rotation) def apply_crop(frame, x0, y0, x1, y1): @@ -66,7 +67,7 @@ def apply_crop(frame, x0, y0, x1, y1): x1 = max(x0, min(x1, w)) y1 = max(y0, min(y1, h)) - return MultiCameraController.apply_crop(frame, (x0, y0, x1, y1)) + return SingleCameraWorker.apply_crop(frame, (x0, y0, x1, y1)) def resize_to_fit(frame, max_w=400, max_h=300): diff --git a/dlclivegui/services/camera_controller.py b/dlclivegui/services/camera_controller.py index 428503cfb..6aebda975 100644 --- a/dlclivegui/services/camera_controller.py +++ b/dlclivegui/services/camera_controller.py @@ -106,7 +106,10 @@ def run(self) -> None: while not self._stop_event.is_set(): try: with self._timing.measure("Single.read"): - frame, timestamp = self._backend.read() + captured = self._backend.read() + frame = captured.frame + timestamp = captured.software_timestamp + timestamp_metadata = captured.timestamp_metadata if frame is None or frame.size == 0: consecutive_errors += 1 if consecutive_errors >= self._max_consecutive_errors: @@ -129,12 +132,12 @@ def run(self) -> None: if recording_enabled and recording_sink is not None: try: with self._timing.measure("Single.recording_sink"): - recording_sink(self._camera_id, frame, timestamp) + recording_sink(self._camera_id, frame, timestamp, timestamp_metadata) except Exception as exc: logger.exception(f"Failed to write frame for camera {self._camera_id}: {exc}") with self._timing.measure("Single.emit"): - self.frame_captured.emit(self._camera_id, frame, timestamp) + self.frame_captured.emit(self._camera_id, frame, timestamp, timestamp_metadata) self._timing.note_frame() self._timing.maybe_log() diff --git a/dlclivegui/services/recording_manager.py b/dlclivegui/services/recording_manager.py index 37e008fdb..7a6e6043d 100644 --- a/dlclivegui/services/recording_manager.py +++ b/dlclivegui/services/recording_manager.py @@ -159,8 +159,8 @@ def _dispatch_loop(self) -> None: if item is _FRAME_SENTINEL: break - cam_id, frame, timestamp = item - self._write_frame_now(cam_id, frame, timestamp) + cam_id, frame, timestamp, timestamp_metadata = item + self._write_frame_now(cam_id, frame, timestamp, timestamp_metadata) finally: try: From 259cad2250cfb7959391f3b20738fd28cd1106bd Mon Sep 17 00:00:00 2001 From: C-Achard Date: Fri, 3 Jul 2026 16:12:48 +0200 Subject: [PATCH 14/21] Add timestamp metadata to frame signal Update `SingleCameraWorker.frame_captured` to emit a fourth argument for timestamp metadata alongside camera ID, frame, and timestamp. This extends the signal contract so downstream multi-camera consumers can receive richer timing context per frame. --- dlclivegui/services/camera_controller.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dlclivegui/services/camera_controller.py b/dlclivegui/services/camera_controller.py index 6aebda975..57173530f 100644 --- a/dlclivegui/services/camera_controller.py +++ b/dlclivegui/services/camera_controller.py @@ -25,7 +25,7 @@ class SingleCameraWorker(QObject): """Worker for a single camera in multi-camera mode.""" - frame_captured = Signal(str, object, float) # camera_id, frame, timestamp + frame_captured = Signal(str, object, float, object) # camera_id, frame, timestamp, timestamp_metadata error_occurred = Signal(str, str) # camera_id, error_message runtime_info = Signal(str, object) # camera_id, dict of runtime info started = Signal(str) # camera_id From 67f49c3197f45c124b694b984e19a9176afc6fa7 Mon Sep 17 00:00:00 2001 From: C-Achard Date: Fri, 3 Jul 2026 16:13:11 +0200 Subject: [PATCH 15/21] Comment previous signals --- dlclivegui/services/multi_camera_controller.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/dlclivegui/services/multi_camera_controller.py b/dlclivegui/services/multi_camera_controller.py index 84c374ab8..deeafdd6a 100644 --- a/dlclivegui/services/multi_camera_controller.py +++ b/dlclivegui/services/multi_camera_controller.py @@ -110,9 +110,9 @@ class MultiCameraController(QObject): # Signals frame_ready = Signal(object) # MultiFrameData (full cam FPS; inference only) - recording_frame_ready = Signal( - str, object, float, object - ) # camera_id, frame, timestamp, timestamp_metadata (full cam FPS; for recording) + # recording_frame_ready = Signal( + # str, object, float, object + # ) # camera_id, frame, timestamp, timestamp_metadata (full cam FPS; for recording) display_ready = Signal(object) # MultiFrameData for GUI display (throttled to GUI_MAX_DISPLAY_FPS) camera_started = Signal(str, object) # camera_id, settings camera_stopped = Signal(str) # camera_id @@ -438,7 +438,7 @@ def _on_frame_captured( frame_data: MultiFrameData | None = None with timing.measure("Multi.slot.total"): - self._settings.get(camera_id) + # self._settings.get(camera_id) # with timing.measure("Multi.apply_transforms"): # if settings and settings.rotation: From 374125735cc9c45215faa4ba9fbd4dfd4c7143fd Mon Sep 17 00:00:00 2001 From: C-Achard Date: Fri, 3 Jul 2026 16:13:43 +0200 Subject: [PATCH 16/21] Fix dispatcher lifecycle and add flush API Refactors recording frame dispatching to start lazily in `write_frame`, simplifies the dispatch loop to block on queue reads, and makes dispatcher shutdown more reliable by waiting to enqueue the sentinel and only clearing thread/queue state when stopping the current thread. Adds `flush(timeout)` to wait for queued frames to be fully dispatched, improving control around recording stop/teardown behavior. --- dlclivegui/services/recording_manager.py | 77 ++++++++++++++++-------- 1 file changed, 51 insertions(+), 26 deletions(-) diff --git a/dlclivegui/services/recording_manager.py b/dlclivegui/services/recording_manager.py index 7a6e6043d..1cfa28632 100644 --- a/dlclivegui/services/recording_manager.py +++ b/dlclivegui/services/recording_manager.py @@ -106,31 +106,28 @@ def pop(self, cam_id: str, default=None) -> VideoRecorder | None: return self._recorders.pop(cam_id, default) def _start_dispatcher(self) -> None: - with self._lock: - if self._dispatch_thread is not None and self._dispatch_thread.is_alive(): - return + if self._dispatch_thread is not None and self._dispatch_thread.is_alive(): + return - self._dispatch_stop.clear() - self._frame_queue = queue.Queue(maxsize=4096) - self._dispatch_thread = threading.Thread( - target=self._dispatch_loop, - name="RecordingManagerDispatcher", - daemon=True, - ) - self._dispatch_thread.start() + self._dispatch_stop.clear() + self._frame_queue = queue.Queue(maxsize=4096) + self._dispatch_thread = threading.Thread( + target=self._dispatch_loop, + name="RecordingManagerDispatcher", + daemon=True, + ) + self._dispatch_thread.start() def _stop_dispatcher(self, timeout: float = 2.0) -> None: - self._dispatch_stop.set() - with self._lock: q = self._frame_queue t = self._dispatch_thread if q is not None: try: - q.put_nowait(_FRAME_SENTINEL) + q.put(_FRAME_SENTINEL, block=True, timeout=timeout) except queue.Full: - pass + log.warning("Recording frame queue full while stopping dispatcher; dispatcher may not stop promptly.") if t is not None: t.join(timeout=timeout) @@ -138,8 +135,9 @@ def _stop_dispatcher(self, timeout: float = 2.0) -> None: log.warning("Recording frame dispatcher did not stop within %.1fs", timeout) with self._lock: - self._dispatch_thread = None - self._frame_queue = None + if self._dispatch_thread is t: + self._dispatch_thread = None + self._frame_queue = None self._dispatch_stop.clear() def _dispatch_loop(self) -> None: @@ -149,11 +147,8 @@ def _dispatch_loop(self) -> None: if q is None: return - while not self._dispatch_stop.is_set(): - try: - item = q.get(timeout=0.1) - except queue.Empty: - continue + while True: + item = q.get() try: if item is _FRAME_SENTINEL: @@ -270,7 +265,6 @@ def start_all( self._run_dir = None return None - self._start_dispatcher() return run_dir def stop_all(self) -> None: @@ -326,13 +320,23 @@ def _write_frame_now( log.exception("Failed to stop recorder for %s after write error.", cam_id) def write_frame( - self, cam_id: str, frame: np.ndarray, timestamp: float | None = None, timestamp_metadata: object | None = None + self, + cam_id: str, + frame: np.ndarray, + timestamp: float | None = None, + timestamp_metadata: object | None = None, ) -> None: with self._lock: - q = self._frame_queue active = cam_id in self._recorders + if not active: + return + + if self._frame_queue is None or self._dispatch_thread is None or not self._dispatch_thread.is_alive(): + self._start_dispatcher() - if not active or q is None: + q = self._frame_queue + + if q is None: return try: @@ -345,6 +349,27 @@ def write_frame( getattr(frame, "dtype", None), ) + def flush(self, timeout: float = 2.0) -> bool: + """Wait until all currently queued recording frames have been dispatched. + + Returns True if the queue drained before timeout, False otherwise. + """ + with self._lock: + q = self._frame_queue + + if q is None: + return True + + done = threading.Event() + + def waiter() -> None: + q.join() + done.set() + + t = threading.Thread(target=waiter, name="RecordingManagerFlush", daemon=True) + t.start() + return done.wait(timeout) + def get_stats_summary(self) -> str: totals = { "enqueued": 0, From 7f1625c47c3a82c62395f52b82782c8a7424973e Mon Sep 17 00:00:00 2001 From: C-Achard Date: Fri, 3 Jul 2026 16:19:52 +0200 Subject: [PATCH 17/21] Update tests for CapturedFrame integration Adjust camera backend and factory tests to return `CapturedFrame` objects instead of `(frame, timestamp)` tuples, matching the updated camera API. Extend `tests/conftest.py` fake DLCLive doubles with runner, processing, and post-processing behavior so `DLCLiveProcessor._process_frame` paths are exercised under the new flow. --- tests/cameras/test_backend_discovery.py | 4 ++-- tests/cameras/test_factory.py | 19 ++++++++-------- tests/cameras/test_fake_backend.py | 2 +- tests/conftest.py | 29 +++++++++++++++++++++++++ 4 files changed, 42 insertions(+), 12 deletions(-) diff --git a/tests/cameras/test_backend_discovery.py b/tests/cameras/test_backend_discovery.py index 610a90c62..0b86e9520 100644 --- a/tests/cameras/test_backend_discovery.py +++ b/tests/cameras/test_backend_discovery.py @@ -26,7 +26,7 @@ def _write_temp_backend_package(tmp_path: Path, pkg_name: str = "test_backends_p # A backend module which registers itself as "lazyfake" backend_code = textwrap.dedent( """ - from dlclivegui.cameras.base import register_backend, CameraBackend + from dlclivegui.cameras.base import register_backend, CameraBackend, CapturedFrame from dlclivegui.config import CameraSettings import numpy as np import time @@ -44,7 +44,7 @@ def open(self) -> None: def read(self): # Small deterministic frame + timestamp frame = np.zeros((2, 3, 3), dtype=np.uint8) - return frame, time.time() + return CapturedFrame(frame, time.time(), None) def close(self) -> None: self._opened = False diff --git a/tests/cameras/test_factory.py b/tests/cameras/test_factory.py index cc1d798de..43516b487 100644 --- a/tests/cameras/test_factory.py +++ b/tests/cameras/test_factory.py @@ -3,6 +3,7 @@ import pytest from dlclivegui.cameras import CameraFactory, DetectedCamera, base +from dlclivegui.cameras.base import CapturedFrame from dlclivegui.config import CameraSettings @@ -69,7 +70,7 @@ def open(self): raise AssertionError("Probing path should not open when rich discovery returns a list") def read(self): - return None, 0.0 + return CapturedFrame(None, 0.0, None) def close(self): pass @@ -112,7 +113,7 @@ def open(self): pass def read(self): - return None, 0.0 + return CapturedFrame(None, 0.0, None) def close(self): pass @@ -150,7 +151,7 @@ def open(self): raise RuntimeError("no device") def read(self): - return None, 0.0 + return CapturedFrame(None, 0.0, None) def close(self): pass @@ -182,7 +183,7 @@ def open(self): raise RuntimeError("no device") def read(self): - return None, 0.0 + return CapturedFrame(None, 0.0, None) def close(self): pass @@ -220,7 +221,7 @@ def open(self): pass def read(self): - return None, 0.0 + return CapturedFrame(None, 0.0, None) def close(self): pass @@ -252,7 +253,7 @@ def open(self): pass def read(self): - return None, 0.0 + return CapturedFrame(None, 0.0, None) def close(self): pass @@ -280,7 +281,7 @@ def open(self): pass def read(self): - return None, 0.0 + return CapturedFrame(None, 0.0, None) def close(self): pass @@ -311,7 +312,7 @@ def open(self): pass def read(self): - return None, 0.0 + return CapturedFrame(None, 0.0, None) def close(self): pass @@ -341,7 +342,7 @@ def open(self): raise RuntimeError("no device") def read(self): - return None, 0.0 + return CapturedFrame(None, 0.0, None) def close(self): pass diff --git a/tests/cameras/test_fake_backend.py b/tests/cameras/test_fake_backend.py index d85616bcc..eac6e6015 100644 --- a/tests/cameras/test_fake_backend.py +++ b/tests/cameras/test_fake_backend.py @@ -26,7 +26,7 @@ def open(self): def read(self): assert self._opened img = np.zeros((10, 20, 3), dtype=np.uint8) - return img, 123.456 + return base.CapturedFrame(img, 123.456, None) def close(self): self._opened = False diff --git a/tests/conftest.py b/tests/conftest.py index 20752cd8a..04992894d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -154,6 +154,18 @@ def _factory(settings: CameraSettings): # --------------------------------------------------------------------- # Test doubles # --------------------------------------------------------------------- +class FakeRunner: + """Minimal fake DLCLive runner used by DLCLiveProcessor._process_frame.""" + + def __init__(self, parent): + self._parent = parent + self.device = "cpu" + self.model = None + self.net = None + + def get_pose(self, processed_frame): + self._parent.pose_calls += 1 + return np.ones((2, 3), dtype=float) class FakeDLCLive: @@ -163,14 +175,31 @@ def __init__(self, **opts): self.opts = opts self.init_called = False self.pose_calls = 0 + self.process_frame_calls = 0 + + self.processor = opts.get("processor") + self.cfg = {"fake": True} + self.runner = FakeRunner(self) + self.pose = None def init_inference(self, frame): self.init_called = True + def process_frame(self, frame): + self.process_frame_calls += 1 + return frame + def get_pose(self, frame, frame_time=None): + # Keep this for compatibility with older tests, but production code now + # uses self.runner.get_pose(...). self.pose_calls += 1 return np.ones((2, 3), dtype=float) + def _post_process_pose(self, processed_frame, frame_time=None): + if self.pose is None: + self.pose = self.runner.get_pose(processed_frame) + return self.pose + @pytest.fixture def fake_dlclive_factory(): From 825d6394418a6977e442e63e2910cb737d443114 Mon Sep 17 00:00:00 2001 From: C-Achard Date: Fri, 3 Jul 2026 16:20:16 +0200 Subject: [PATCH 18/21] Stabilize recording and camera tests Update tests to match recent recording/capture API changes and reduce flakiness. Camera dialog E2E stubs now return `CapturedFrame`, multicam tests validate the new recording sink path (including timestamp metadata forwarding), and recording manager tests consistently clean up with `stop_all()` plus queue flushes before assertions. GUI overlay tests tied to removed behavior are now skipped, and config tests now expect numeric `-input_framerate` values instead of formatted strings. --- .../gui/camera_config/test_cam_dialog_e2e.py | 6 +- tests/gui/test_pose_overlay.py | 2 + tests/gui/test_rec_manager.py | 476 ++++++++++-------- tests/services/test_multicam_controller.py | 113 +++-- tests/test_config.py | 6 +- 5 files changed, 345 insertions(+), 258 deletions(-) diff --git a/tests/gui/camera_config/test_cam_dialog_e2e.py b/tests/gui/camera_config/test_cam_dialog_e2e.py index df1c357e8..9556efc72 100644 --- a/tests/gui/camera_config/test_cam_dialog_e2e.py +++ b/tests/gui/camera_config/test_cam_dialog_e2e.py @@ -8,7 +8,7 @@ from PySide6.QtCore import Qt from PySide6.QtWidgets import QMessageBox -from dlclivegui.cameras.base import CameraBackend +from dlclivegui.cameras.base import CameraBackend, CapturedFrame from dlclivegui.cameras.factory import CameraFactory, DetectedCamera from dlclivegui.config import CameraSettings, MultiCameraSettings from dlclivegui.gui.camera_config.camera_config_dialog import CameraConfigDialog @@ -194,7 +194,7 @@ def close(self): self._opened = False def read(self): - return np.zeros((30, 40, 3), dtype=np.uint8), 0.1 + return CapturedFrame(np.zeros((30, 40, 3), dtype=np.uint8), 0.1, None) CountingBackend.opens = 0 monkeypatch.setattr(CameraFactory, "create", staticmethod(lambda s: CountingBackend(s))) @@ -238,7 +238,7 @@ def close(self): self._opened = False def read(self): - return np.zeros((30, 40, 3), dtype=np.uint8), 0.1 + return CapturedFrame(np.zeros((30, 40, 3), dtype=np.uint8), 0.1, None) CountingBackend.opens = 0 monkeypatch.setattr(CameraFactory, "create", staticmethod(lambda s: CountingBackend(s))) diff --git a/tests/gui/test_pose_overlay.py b/tests/gui/test_pose_overlay.py index 511d44552..bef210b3c 100644 --- a/tests/gui/test_pose_overlay.py +++ b/tests/gui/test_pose_overlay.py @@ -9,6 +9,7 @@ def stop(self): @pytest.mark.gui @pytest.mark.timeout(10) +@pytest.mark.skip("Removed functionality.") def test_record_overlay_uses_identity_transform_for_per_camera_recording(window, draw_pose_stub): # Disable event timers to avoid GUI rendering pipelines interfering with test window._display_timer.stop() @@ -47,6 +48,7 @@ def test_record_overlay_uses_identity_transform_for_per_camera_recording(window, @pytest.mark.gui @pytest.mark.timeout(10) +@pytest.mark.skip("Removed functionality.") def test_record_overlay_toggle_affects_frames_sent_to_recorder(window, recording_frame_spy, draw_pose_stub): # Disable event timers to avoid GUI rendering pipelines interfering with test window._display_timer.stop() diff --git a/tests/gui/test_rec_manager.py b/tests/gui/test_rec_manager.py index 914132d35..01853a8bd 100644 --- a/tests/gui/test_rec_manager.py +++ b/tests/gui/test_rec_manager.py @@ -43,37 +43,40 @@ def test_start_all_creates_recorders_and_returns_run_dir( spy, expected_run_dir = patch_build_run_dir mgr = RecordingManager() - run_dir = mgr.start_all( - recording_settings, - _active_cams_two, - current_frames, - session_name="Sess", - use_timestamp=True, - all_or_nothing=False, - ) - - assert run_dir == expected_run_dir - assert mgr.is_active is True - assert mgr.run_dir == expected_run_dir - assert mgr.session_dir is not None - assert len(mgr.recorders) == 2 - - # build_run_dir called with correct use_timestamp - assert spy["use_timestamp"] is True - assert spy["session_dir"] is not None + try: + run_dir = mgr.start_all( + recording_settings, + _active_cams_two, + current_frames, + session_name="Sess", + use_timestamp=True, + all_or_nothing=False, + ) - # Validate per-cam recorder construction - for cam in _active_cams_two: - cam_id = get_camera_id(cam) - rec = mgr.recorders[cam_id] - assert rec.codec == recording_settings.codec - assert rec.crf == recording_settings.crf - assert rec.frame_rate == float(cam.fps) - assert rec.is_running is True - # output file should be inside run dir - assert rec.output.parent == expected_run_dir - # filename should include backend + cam index - assert f"_{cam.backend}_cam{cam.index}" in rec.output.name + assert run_dir == expected_run_dir + assert mgr.is_active is True + assert mgr.run_dir == expected_run_dir + assert mgr.session_dir is not None + assert len(mgr.recorders) == 2 + + # build_run_dir called with correct use_timestamp + assert spy["use_timestamp"] is True + assert spy["session_dir"] is not None + + # Validate per-cam recorder construction + for cam in _active_cams_two: + cam_id = get_camera_id(cam) + rec = mgr.recorders[cam_id] + assert rec.codec == recording_settings.codec + assert rec.crf == recording_settings.crf + assert rec.frame_rate == float(cam.fps) + assert rec.is_running is True + # output file should be inside run dir + assert rec.output.parent == expected_run_dir + # filename should include backend + cam index + assert f"_{cam.backend}_cam{cam.index}" in rec.output.name + finally: + mgr.stop_all() @pytest.mark.unit @@ -83,8 +86,11 @@ def test_start_all_passes_use_timestamp_flag( spy, _expected_run_dir = patch_build_run_dir mgr = RecordingManager() - mgr.start_all(recording_settings, _active_cams_two, current_frames, session_name="Sess", use_timestamp=False) - assert spy["use_timestamp"] is False + try: + mgr.start_all(recording_settings, _active_cams_two, current_frames, session_name="Sess", use_timestamp=False) + assert spy["use_timestamp"] is False + finally: + mgr.stop_all() @pytest.mark.unit @@ -92,14 +98,18 @@ def test_frame_size_is_inferred_from_current_frames( recording_settings, _active_cams_two, current_frames, patch_video_recorder, patch_build_run_dir ): mgr = RecordingManager() - mgr.start_all(recording_settings, _active_cams_two, current_frames, session_name="Sess") - # cam0 -> 480x640, cam1 -> 720x1280 - for cam in _active_cams_two: - cam_id = get_camera_id(cam) - rec = mgr.recorders[cam_id] - frame = current_frames[cam_id] - assert rec.frame_size == (frame.shape[0], frame.shape[1]) + try: + mgr.start_all(recording_settings, _active_cams_two, current_frames, session_name="Sess") + + # cam0 -> 480x640, cam1 -> 720x1280 + for cam in _active_cams_two: + cam_id = get_camera_id(cam) + rec = mgr.recorders[cam_id] + frame = current_frames[cam_id] + assert rec.frame_size == (frame.shape[0], frame.shape[1]) + finally: + mgr.stop_all() @pytest.mark.unit @@ -111,10 +121,14 @@ def test_missing_frame_results_in_none_frame_size( current_frames.pop(cam1_id) mgr = RecordingManager() - mgr.start_all(recording_settings, _active_cams_two, current_frames, session_name="Sess") - rec1 = mgr.recorders[cam1_id] - assert rec1.frame_size is None + try: + mgr.start_all(recording_settings, _active_cams_two, current_frames, session_name="Sess") + + rec1 = mgr.recorders[cam1_id] + assert rec1.frame_size is None + finally: + mgr.stop_all() @pytest.mark.unit @@ -175,6 +189,7 @@ def start_with_failure(self): assert mgr.session_dir is None finally: patch_video_recorder.start = original_start + mgr.stop_all() @pytest.mark.unit @@ -197,14 +212,19 @@ def test_write_frame_uses_given_timestamp( recording_settings, _active_cams_two, current_frames, patch_video_recorder, patch_build_run_dir ): mgr = RecordingManager() - mgr.start_all(recording_settings, _active_cams_two, current_frames, session_name="Sess") - cam0_id = get_camera_id(_active_cams_two[0]) - frame = current_frames[cam0_id] - mgr.write_frame(cam0_id, frame, timestamp=123.0) + try: + mgr.start_all(recording_settings, _active_cams_two, current_frames, session_name="Sess") - rec = mgr.recorders[cam0_id] - assert rec.write_calls[-1][1] == 123.0 + cam0_id = get_camera_id(_active_cams_two[0]) + frame = current_frames[cam0_id] + mgr.write_frame(cam0_id, frame, timestamp=123.0) + assert mgr.flush(timeout=2.0) + + rec = mgr.recorders[cam0_id] + assert rec.write_calls[-1][1] == 123.0 + finally: + mgr.stop_all() @pytest.mark.unit @@ -212,18 +232,23 @@ def test_write_frame_uses_time_when_timestamp_missing( recording_settings, _active_cams_two, current_frames, patch_video_recorder, patch_build_run_dir, monkeypatch ): mgr = RecordingManager() - mgr.start_all(recording_settings, _active_cams_two, current_frames, session_name="Sess") - import dlclivegui.services.recording_manager as rm_mod # noqa: E402 + try: + mgr.start_all(recording_settings, _active_cams_two, current_frames, session_name="Sess") - monkeypatch.setattr(rm_mod.time, "time", lambda: 999.0) + import dlclivegui.services.recording_manager as rm_mod # noqa: E402 - cam0_id = get_camera_id(_active_cams_two[0]) - frame = current_frames[cam0_id] - mgr.write_frame(cam0_id, frame, timestamp=None) + monkeypatch.setattr(rm_mod.time, "time", lambda: 999.0) + + cam0_id = get_camera_id(_active_cams_two[0]) + frame = current_frames[cam0_id] + mgr.write_frame(cam0_id, frame, timestamp=None) + assert mgr.flush(timeout=2.0) - rec = mgr.recorders[cam0_id] - assert rec.write_calls[-1][1] == 999.0 + rec = mgr.recorders[cam0_id] + assert rec.write_calls[-1][1] == 999.0 + finally: + mgr.stop_all() @pytest.mark.unit @@ -231,14 +256,20 @@ def test_write_frame_removes_recorder_on_exception( recording_settings, _active_cams_two, current_frames, patch_video_recorder, patch_build_run_dir ): mgr = RecordingManager() - mgr.start_all(recording_settings, _active_cams_two, current_frames, session_name="Sess") - cam0_id = get_camera_id(_active_cams_two[0]) - rec = mgr.recorders[cam0_id] - rec.raise_on_write = True + try: + mgr.start_all(recording_settings, _active_cams_two, current_frames, session_name="Sess") - mgr.write_frame(cam0_id, current_frames[cam0_id], timestamp=1.0) - assert cam0_id not in mgr.recorders + cam0_id = get_camera_id(_active_cams_two[0]) + rec = mgr.recorders[cam0_id] + rec.raise_on_write = True + + mgr.write_frame(cam0_id, current_frames[cam0_id], timestamp=1.0) + assert mgr.flush(timeout=2.0) + + assert cam0_id not in mgr.recorders + finally: + mgr.stop_all() @pytest.mark.unit @@ -246,17 +277,21 @@ def test_get_stats_summary_single_recorder_uses_formatter( recording_settings, _active_cams_two, current_frames, patch_video_recorder, patch_build_run_dir, monkeypatch ): mgr = RecordingManager() - mgr.start_all(recording_settings, [_active_cams_two[0]], current_frames, session_name="Sess") - cam0_id = get_camera_id(_active_cams_two[0]) - mgr.recorders[cam0_id]._stats = RecorderStats(frames_written=10, frames_enqueued=12) + try: + mgr.start_all(recording_settings, [_active_cams_two[0]], current_frames, session_name="Sess") + + cam0_id = get_camera_id(_active_cams_two[0]) + mgr.recorders[cam0_id]._stats = RecorderStats(frames_written=10, frames_enqueued=12) - # Patch formatter to avoid depending on formatting implementation - import dlclivegui.utils.stats as stats_mod + # Patch formatter to avoid depending on formatting implementation + import dlclivegui.utils.stats as stats_mod - monkeypatch.setattr(stats_mod, "format_recorder_stats", lambda s: "OK_SINGLE") + monkeypatch.setattr(stats_mod, "format_recorder_stats", lambda s: "OK_SINGLE") - assert mgr.get_stats_summary() == "OK_SINGLE" + assert mgr.get_stats_summary() == "OK_SINGLE" + finally: + mgr.stop_all() @pytest.mark.unit @@ -264,39 +299,43 @@ def test_get_stats_summary_multi_aggregates( recording_settings, _active_cams_two, current_frames, patch_video_recorder, patch_build_run_dir ): mgr = RecordingManager() - mgr.start_all(recording_settings, _active_cams_two, current_frames, session_name="Sess") - ids = [get_camera_id(c) for c in _active_cams_two] - - mgr.recorders[ids[0]]._stats = RecorderStats( - frames_enqueued=12, - frames_written=10, - dropped_frames=1, - queue_size=2, - buffer_size=10, - average_latency=0.01, - last_latency=0.02, - write_fps=25.0, - ) - mgr.recorders[ids[1]]._stats = RecorderStats( - frames_enqueued=24, - frames_written=20, - dropped_frames=3, - queue_size=4, - buffer_size=10, - average_latency=0.03, - last_latency=0.05, - write_fps=30.0, - ) - - summary = mgr.get_stats_summary() - - assert "2 cams" in summary - assert "30/36 frames" in summary - assert "writer 55.0 fps" in summary - assert "dropped 4" in summary - assert "queue 6/20" in summary - assert "backlog 6" in summary + try: + mgr.start_all(recording_settings, _active_cams_two, current_frames, session_name="Sess") + + ids = [get_camera_id(c) for c in _active_cams_two] + + mgr.recorders[ids[0]]._stats = RecorderStats( + frames_enqueued=12, + frames_written=10, + dropped_frames=1, + queue_size=2, + buffer_size=10, + average_latency=0.01, + last_latency=0.02, + write_fps=25.0, + ) + mgr.recorders[ids[1]]._stats = RecorderStats( + frames_enqueued=24, + frames_written=20, + dropped_frames=3, + queue_size=4, + buffer_size=10, + average_latency=0.03, + last_latency=0.05, + write_fps=30.0, + ) + + summary = mgr.get_stats_summary() + + assert "2 cams" in summary + assert "30/36 frames" in summary + assert "writer 55.0 fps" in summary + assert "dropped 4" in summary + assert "queue 6/20" in summary + assert "backlog 6" in summary + finally: + mgr.stop_all() @pytest.mark.unit @@ -307,51 +346,58 @@ def test_recording_manager_uses_stable_camera_id_not_display_id( ): mgr = RecordingManager() - cam = CameraSettings( - name="GenTL cam", - backend="gentl", - index=0, - fps=30.0, - enabled=True, - properties={ - "gentl": { - "device_id": "serial:SER0", - "serial_number": "SER0", - } - }, - ).apply_defaults() - - stable_id = get_camera_id(cam) - display_id = get_display_id(cam) - - assert stable_id == "gentl:serial:SER0" - assert display_id == "GenTL cam" - assert stable_id != display_id - - frame = np.zeros((480, 640, 3), dtype=np.uint8) - current_frames = {stable_id: frame} - - run_dir = mgr.start_all( - recording_settings, - [cam], - current_frames, - session_name="Sess", - ) + try: + cam = CameraSettings( + name="GenTL cam", + backend="gentl", + index=0, + fps=30.0, + enabled=True, + properties={ + "gentl": { + "device_id": "serial:SER0", + "serial_number": "SER0", + } + }, + ).apply_defaults() + + stable_id = get_camera_id(cam) + display_id = get_display_id(cam) + + assert stable_id == "gentl:serial:SER0" + assert display_id == "GenTL cam" + assert stable_id != display_id + + frame = np.zeros((480, 640, 3), dtype=np.uint8) + current_frames = {stable_id: frame} - assert run_dir is not None - assert stable_id in mgr.recorders - assert display_id not in mgr.recorders + run_dir = mgr.start_all( + recording_settings, + [cam], + current_frames, + session_name="Sess", + ) + + assert run_dir is not None + assert stable_id in mgr.recorders + assert display_id not in mgr.recorders - rec = mgr.recorders[stable_id] - assert rec.frame_size == (480, 640) + rec = mgr.recorders[stable_id] + assert rec.frame_size == (480, 640) + + mgr.write_frame(stable_id, frame, timestamp=123.0) + assert mgr.flush(timeout=2.0) + + assert len(rec.write_calls) == 1 + assert rec.write_calls[-1][1] == 123.0 - mgr.write_frame(stable_id, frame, timestamp=123.0) - assert len(rec.write_calls) == 1 - assert rec.write_calls[-1][1] == 123.0 + # Display ID is GUI-only and must not route frames internally. + mgr.write_frame(display_id, frame, timestamp=456.0) + assert mgr.flush(timeout=2.0) - # Display ID is GUI-only and must not route frames internally. - mgr.write_frame(display_id, frame, timestamp=456.0) - assert len(rec.write_calls) == 1 + assert len(rec.write_calls) == 1 + finally: + mgr.stop_all() @pytest.mark.unit @@ -362,41 +408,44 @@ def test_start_all_does_not_infer_frame_size_from_display_id( ): mgr = RecordingManager() - cam = CameraSettings( - name="GenTL cam", - backend="gentl", - index=0, - fps=30.0, - enabled=True, - properties={ - "gentl": { - "device_id": "serial:SER0", - "serial_number": "SER0", - } - }, - ).apply_defaults() - - stable_id = get_camera_id(cam) - display_id = get_display_id(cam) - - frame = np.zeros((480, 640, 3), dtype=np.uint8) - - # Simulate the buggy situation: frames are keyed by display ID. - current_frames = {display_id: frame} - - mgr.start_all( - recording_settings, - [cam], - current_frames, - session_name="Sess", - ) + try: + cam = CameraSettings( + name="GenTL cam", + backend="gentl", + index=0, + fps=30.0, + enabled=True, + properties={ + "gentl": { + "device_id": "serial:SER0", + "serial_number": "SER0", + } + }, + ).apply_defaults() + + stable_id = get_camera_id(cam) + display_id = get_display_id(cam) + + frame = np.zeros((480, 640, 3), dtype=np.uint8) + + # Simulate the buggy situation: frames are keyed by display ID. + current_frames = {display_id: frame} + + mgr.start_all( + recording_settings, + [cam], + current_frames, + session_name="Sess", + ) - assert stable_id in mgr.recorders - assert display_id not in mgr.recorders + assert stable_id in mgr.recorders + assert display_id not in mgr.recorders - # Since RecordingManager uses stable IDs internally, it should not find this frame. - rec = mgr.recorders[stable_id] - assert rec.frame_size is None + # Since RecordingManager uses stable IDs internally, it should not find this frame. + rec = mgr.recorders[stable_id] + assert rec.frame_size is None + finally: + mgr.stop_all() @pytest.mark.unit @@ -412,18 +461,22 @@ def test_start_all_passes_writegear_options( recording_settings.fast_encoding = True mgr = RecordingManager() - mgr.start_all(recording_settings, _active_cams_two, current_frames, session_name="Sess") - for cam in _active_cams_two: - cam_id = get_camera_id(cam) - rec = mgr.recorders[cam_id] + try: + mgr.start_all(recording_settings, _active_cams_two, current_frames, session_name="Sess") - opts_ovrr = rec.writer_options_overrides - assert opts_ovrr is not None - assert opts_ovrr["-vcodec"] == "libx264" - assert opts_ovrr["-crf"] == "23" - assert opts_ovrr["-preset"] == "ultrafast" - assert opts_ovrr["-tune"] == "zerolatency" + for cam in _active_cams_two: + cam_id = get_camera_id(cam) + rec = mgr.recorders[cam_id] + + opts_ovrr = rec.writer_options_overrides + assert opts_ovrr is not None + assert opts_ovrr["-vcodec"] == "libx264" + assert opts_ovrr["-crf"] == "23" + assert opts_ovrr["-preset"] == "ultrafast" + assert opts_ovrr["-tune"] == "zerolatency" + finally: + mgr.stop_all() class TestRecordingManagerTimestampMetadata: @@ -437,28 +490,33 @@ def test_write_frame_passes_timestamp_metadata( patch_build_run_dir, ): mgr = RecordingManager() - mgr.start_all(recording_settings, _active_cams_two, current_frames, session_name="Sess") - - cam0_id = get_camera_id(_active_cams_two[0]) - frame = current_frames[cam0_id] - - meta = FrameTimestampMetadata( - source="grab_result.GetTimeStamp", - backend="basler", - default_reported="seconds", - seconds=0.001, - raw_value=1_000_000, - raw_unit="ticks", - tick_frequency_hz=1_000_000_000.0, - kind="camera_clock", - ) - - mgr.write_frame(cam0_id, frame, timestamp=123.0, timestamp_metadata=meta) - - rec = mgr.recorders[cam0_id] - assert len(rec.write_calls) == 1 - written_frame, written_timestamp, written_metadata = rec.write_calls[0] - assert written_frame is frame - assert written_timestamp == 123.0 - assert written_metadata is meta + try: + mgr.start_all(recording_settings, _active_cams_two, current_frames, session_name="Sess") + + cam0_id = get_camera_id(_active_cams_two[0]) + frame = current_frames[cam0_id] + + meta = FrameTimestampMetadata( + source="grab_result.GetTimeStamp", + backend="basler", + default_reported="seconds", + seconds=0.001, + raw_value=1_000_000, + raw_unit="ticks", + tick_frequency_hz=1_000_000_000.0, + kind="camera_clock", + ) + + mgr.write_frame(cam0_id, frame, timestamp=123.0, timestamp_metadata=meta) + assert mgr.flush(timeout=2.0) + + rec = mgr.recorders[cam0_id] + assert len(rec.write_calls) == 1 + + written_frame, written_timestamp, written_metadata = rec.write_calls[0] + assert written_frame is frame + assert written_timestamp == 123.0 + assert written_metadata is meta + finally: + mgr.stop_all() diff --git a/tests/services/test_multicam_controller.py b/tests/services/test_multicam_controller.py index 783b02240..1f8d0f17a 100644 --- a/tests/services/test_multicam_controller.py +++ b/tests/services/test_multicam_controller.py @@ -504,7 +504,7 @@ def _create(settings): @pytest.mark.unit -def test_recording_frame_ready_only_emits_when_enabled(qtbot, patch_factory): +def test_recording_sink_receives_frames_when_enabled(qtbot, patch_factory): mc = MultiCameraController() cam = CameraSettings( @@ -516,26 +516,25 @@ def test_recording_frame_ready_only_emits_when_enabled(qtbot, patch_factory): ).apply_defaults() cam_id = get_camera_id(cam) - seen: list[tuple[str, tuple, float]] = [] + seen: list[tuple[str, tuple, float, object]] = [] - def on_recording_frame(camera_id, frame, timestamp, timestamp_metadata=None): - seen.append((camera_id, frame.shape, timestamp)) - - mc.recording_frame_ready.connect(on_recording_frame) + def sink(camera_id, frame, timestamp, timestamp_metadata=None): + seen.append((camera_id, frame.shape, timestamp, timestamp_metadata)) try: with qtbot.waitSignal(mc.all_started, timeout=1500): mc.start([cam]) - # Disabled by default: should not emit recording frames. + # Disabled by default. qtbot.wait(300) assert seen == [] + mc.set_recording_sink(sink) mc.set_recording_frame_do_emit(True) qtbot.waitUntil(lambda: bool(seen), timeout=2000) - camera_id, shape, timestamp = seen[-1] + camera_id, shape, timestamp, timestamp_metadata = seen[-1] assert camera_id == cam_id assert isinstance(timestamp, float) assert len(shape) in (2, 3) @@ -551,48 +550,76 @@ def on_recording_frame(camera_id, frame, timestamp, timestamp_metadata=None): mc.stop(wait=True) -class TestRecordingFrameTimestamps: - @pytest.mark.unit - def test_recording_frame_ready_forwards_timestamp_metadata(self, qtbot): - mc = MultiCameraController() - mc._running = True - mc._recording_frame_emission_enabled = True +@pytest.mark.unit +def test_recording_sink_forwards_timestamp_metadata(qtbot, monkeypatch): + from dlclivegui.cameras.base import CapturedFrame + from dlclivegui.cameras.factory import CameraFactory + + meta = FrameTimestampMetadata( + source="grab_result.GetTimeStamp", + backend="basler", + default_reported="seconds", + seconds=0.001, + raw_value=1_000_000, + raw_unit="ticks", + tick_frequency_hz=1_000_000_000.0, + kind="camera_clock", + ) + + class TimestampBackend: + waits_for_hardware_trigger = False + + def __init__(self, settings): + self.settings = settings + self._count = 0 + + def open(self): + pass + + def read(self): + self._count += 1 + return CapturedFrame( + frame=np.zeros((10, 10), dtype=np.uint8), + software_timestamp=123.0 + self._count, + timestamp_metadata=meta, + ) + + def close(self): + pass - cam_id = "basler:0815-0000" - mc._settings[cam_id] = CameraSettings( - name="C", - backend="basler", - index=0, - enabled=True, - ).apply_defaults() - mc._camera_display_order = [cam_id] - mc._display_ids[cam_id] = "C" + monkeypatch.setattr(CameraFactory, "create", staticmethod(lambda settings: TimestampBackend(settings))) - frame = np.zeros((10, 10), dtype=np.uint8) - meta = FrameTimestampMetadata( - source="grab_result.GetTimeStamp", - backend="basler", - default_reported="seconds", - seconds=0.001, - raw_value=1_000_000, - raw_unit="ticks", - tick_frequency_hz=1_000_000_000.0, - kind="camera_clock", - ) + mc = MultiCameraController() + cam = CameraSettings( + name="C", + backend="basler", + index=0, + enabled=True, + properties={"basler": {"device_id": "0815-0000"}}, + ).apply_defaults() - seen = [] + cam_id = get_camera_id(cam) + seen = [] - def on_recording_frame(camera_id, emitted_frame, timestamp, timestamp_metadata): - seen.append((camera_id, emitted_frame, timestamp, timestamp_metadata)) + def sink(camera_id, frame, timestamp, timestamp_metadata=None): + seen.append((camera_id, frame, timestamp, timestamp_metadata)) - mc.recording_frame_ready.connect(on_recording_frame) + try: + with qtbot.waitSignal(mc.all_started, timeout=1500): + mc.start([cam]) - mc._on_frame_captured(cam_id, frame, 123.0, meta) + # Recording is disabled by start(); enable the new sink path after cameras are running. + mc.set_recording_sink(sink) + mc.set_recording_frame_do_emit(True) - assert len(seen) == 1 + qtbot.waitUntil(lambda: bool(seen), timeout=2000) - camera_id, emitted_frame, timestamp, timestamp_metadata = seen[0] + camera_id, frame, timestamp, timestamp_metadata = seen[-1] assert camera_id == cam_id - assert emitted_frame is frame - assert timestamp == 123.0 + assert frame.shape == (10, 10) + assert isinstance(timestamp, float) assert timestamp_metadata is meta + + finally: + with qtbot.waitSignal(mc.all_stopped, timeout=2000): + mc.stop(wait=True) diff --git a/tests/test_config.py b/tests/test_config.py index 7c2f64bf0..2a967a01d 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -81,7 +81,7 @@ def test_recording_settings_writegear_options_default(): opts = settings.writegear_options(100.0) - assert opts["-input_framerate"] == "100.000000" + assert opts["-input_framerate"] == 100.0 assert opts["-vcodec"] == "libx264" assert opts["-crf"] == "23" assert "-preset" not in opts @@ -93,7 +93,7 @@ def test_recording_settings_writegear_options_fast_encoding_x264(): opts = settings.writegear_options(100.0) - assert opts["-input_framerate"] == "100.000000" + assert opts["-input_framerate"] == 100.0 assert opts["-vcodec"] == "libx264" assert opts["-crf"] == "23" assert opts["-preset"] == "ultrafast" @@ -115,4 +115,4 @@ def test_recording_settings_writegear_options_invalid_fps_falls_back_to_30(): opts = settings.writegear_options(None) - assert opts["-input_framerate"] == "30.000000" + assert opts["-input_framerate"] == 30.0 From f08371673fefc6b6b06ad35ec4eee03ca4145e60 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Fri, 14 Aug 2026 10:01:26 +0200 Subject: [PATCH 19/21] Harden recording dispatcher shutdown flow Refactors dispatcher lifecycle management to make stop/start behavior safer under load. Introduces explicit dispatcher state flags, bounded constants for queue size/stop timeout, and makes `_stop_dispatcher` return success instead of silently forcing teardown. Shutdown now only enqueues a sentinel once, preserves thread/queue state when sentinel enqueue or join times out, and logs these cases clearly. The dispatch loop now blocks on queue reads, handles full frame payloads consistently, logs unexpected dispatch errors, and warns on invalid `task_done` usage. `stop_all` now returns a boolean to report incomplete shutdown, and `write_frame` only enqueues while the dispatcher is actively accepting frames. --- dlclivegui/services/recording_manager.py | 149 +++++++++++++++++------ 1 file changed, 110 insertions(+), 39 deletions(-) diff --git a/dlclivegui/services/recording_manager.py b/dlclivegui/services/recording_manager.py index 1cfa28632..37bead192 100644 --- a/dlclivegui/services/recording_manager.py +++ b/dlclivegui/services/recording_manager.py @@ -16,6 +16,9 @@ log = logging.getLogger(__name__) +DISPATCH_STOP_TIMEOUT = 2.0 +DISPATCH_QUEUE_MAXSIZE = 4096 + _FRAME_SENTINEL = object() @@ -30,7 +33,8 @@ def __init__(self): self._lock = threading.RLock() self._frame_queue: queue.Queue | None = None self._dispatch_thread: threading.Thread | None = None - self._dispatch_stop = threading.Event() + self._dispatch_accepting: bool = False + self._dispatch_sentinel_enqueued: bool = False @property def is_active(self) -> bool: @@ -106,39 +110,79 @@ def pop(self, cam_id: str, default=None) -> VideoRecorder | None: return self._recorders.pop(cam_id, default) def _start_dispatcher(self) -> None: - if self._dispatch_thread is not None and self._dispatch_thread.is_alive(): - return + with self._lock: + if self._dispatch_thread is not None and self._dispatch_thread.is_alive(): + return - self._dispatch_stop.clear() - self._frame_queue = queue.Queue(maxsize=4096) - self._dispatch_thread = threading.Thread( - target=self._dispatch_loop, - name="RecordingManagerDispatcher", - daemon=True, - ) - self._dispatch_thread.start() + self._frame_queue = queue.Queue(maxsize=DISPATCH_QUEUE_MAXSIZE) + self._dispatch_accepting = True + self._dispatch_sentinel_enqueued = False + self._dispatch_thread = threading.Thread( + target=self._dispatch_loop, + name="RecordingManagerDispatcher", + daemon=True, + ) + self._dispatch_thread.start() - def _stop_dispatcher(self, timeout: float = 2.0) -> None: + def _stop_dispatcher( + self, + timeout: float = DISPATCH_STOP_TIMEOUT, + ) -> bool: with self._lock: q = self._frame_queue t = self._dispatch_thread - if q is not None: + if t is None: + self._frame_queue = None + self._dispatch_accepting = False + self._dispatch_sentinel_enqueued = False + return True + + # Establish the stop boundary while holding the same lock used + # by write_frame(). No new frame can be accepted after this. + self._dispatch_accepting = False + + should_enqueue_sentinel = not self._dispatch_sentinel_enqueued + self._dispatch_sentinel_enqueued = True + + if should_enqueue_sentinel and q is not None: try: - q.put(_FRAME_SENTINEL, block=True, timeout=timeout) + q.put( + _FRAME_SENTINEL, + block=True, + timeout=timeout, + ) except queue.Full: - log.warning("Recording frame queue full while stopping dispatcher; dispatcher may not stop promptly.") + # No sentinel was inserted, so allow a later stop attempt + # to retry once the dispatcher has freed queue capacity. + with self._lock: + if self._dispatch_thread is t: + self._dispatch_sentinel_enqueued = False - if t is not None: - t.join(timeout=timeout) - if t.is_alive(): - log.warning("Recording frame dispatcher did not stop within %.1fs", timeout) + log.warning( + "Could not enqueue recording dispatcher sentinel within %.1fs; preserving live dispatcher state", + timeout, + ) + return False + + t.join(timeout=timeout) + + if t.is_alive(): + log.warning( + "Recording frame dispatcher did not stop within %.1fs; preserving its queue and thread state", + timeout, + ) + return False with self._lock: + # Do not clear a newer dispatcher if one was somehow started. if self._dispatch_thread is t: self._dispatch_thread = None self._frame_queue = None - self._dispatch_stop.clear() + self._dispatch_accepting = False + self._dispatch_sentinel_enqueued = False + + return True def _dispatch_loop(self) -> None: with self._lock: @@ -152,16 +196,30 @@ def _dispatch_loop(self) -> None: try: if item is _FRAME_SENTINEL: - break + return + + ( + cam_id, + frame, + timestamp, + timestamp_metadata, + ) = item + + self._write_frame_now( + cam_id, + frame, + timestamp, + timestamp_metadata, + ) - cam_id, frame, timestamp, timestamp_metadata = item - self._write_frame_now(cam_id, frame, timestamp, timestamp_metadata) + except Exception: + log.exception("Unhandled error dispatching a recording frame") finally: try: q.task_done() except ValueError: - pass + log.warning("Recording dispatcher called task_done() too many times") def start_all( self, @@ -267,8 +325,10 @@ def start_all( return run_dir - def stop_all(self) -> None: - self._stop_dispatcher() + def stop_all(self) -> bool: + if not self._stop_dispatcher(): + log.warning("Recording stop is incomplete, frame dispatcher is still draining.") + return False with self._lock: recorders = list(self._recorders.items()) @@ -285,6 +345,8 @@ def stop_all(self) -> None: self._session_dir = None self._run_dir = None + return True + def _write_frame_now( self, cam_id: str, frame: np.ndarray, timestamp: float | None = None, timestamp_metadata: object | None = None ) -> None: @@ -326,9 +388,15 @@ def write_frame( timestamp: float | None = None, timestamp_metadata: object | None = None, ) -> None: + payload = ( + cam_id, + frame, + timestamp if timestamp is not None else time.time(), + timestamp_metadata, + ) + with self._lock: - active = cam_id in self._recorders - if not active: + if cam_id not in self._recorders: return if self._frame_queue is None or self._dispatch_thread is None or not self._dispatch_thread.is_alive(): @@ -336,18 +404,21 @@ def write_frame( q = self._frame_queue - if q is None: - return + if q is None or not self._dispatch_accepting: + return - try: - q.put_nowait((cam_id, frame, timestamp if timestamp is not None else time.time(), timestamp_metadata)) - except queue.Full: - log.warning( - "Recording manager frame queue full; dropping frame for %s. frame_shape=%s dtype=%s", - cam_id, - getattr(frame, "shape", None), - getattr(frame, "dtype", None), - ) + try: + q.put_nowait(payload) + return + except queue.Full: + pass + + log.warning( + "Recording manager frame queue full; dropping frame for %s. frame_shape=%s dtype=%s", + cam_id, + getattr(frame, "shape", None), + getattr(frame, "dtype", None), + ) def flush(self, timeout: float = 2.0) -> bool: """Wait until all currently queued recording frames have been dispatched. From 7f9840c71a008e5827797f816b56f97b41598261 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Fri, 14 Aug 2026 10:03:12 +0200 Subject: [PATCH 20/21] Retry recorder stop until success Make recording shutdown more robust by retrying `stop_all()` in both async stop and app close paths with a configurable interval (`RECORD_STOP_RETRY_INTERVAL`). Also disable recording frame emission before shutdown stop attempts and log stop/shutdown errors for better diagnostics. --- dlclivegui/config.py | 1 + dlclivegui/gui/main_window.py | 20 ++++++++++++++++---- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/dlclivegui/config.py b/dlclivegui/config.py index c948af0a6..962158a37 100644 --- a/dlclivegui/config.py +++ b/dlclivegui/config.py @@ -27,6 +27,7 @@ DEFAULT_RECORDING_FPS: float = 30.0 ALLOWED_VIDEO_CONTAINERS: set[str] = {"mp4", "avi", "mov"} DEFAULT_RECORDING_CONTAINER: str = "mp4" +RECORD_STOP_RETRY_INTERVAL: float = 0.25 ## Debug diff --git a/dlclivegui/gui/main_window.py b/dlclivegui/gui/main_window.py index 4af17d600..b62fb4067 100644 --- a/dlclivegui/gui/main_window.py +++ b/dlclivegui/gui/main_window.py @@ -54,6 +54,7 @@ DEFAULT_RECORDING_CONTAINER, DLC_DO_LOG_TIMING, GUI_MAX_DISPLAY_FPS, + RECORD_STOP_RETRY_INTERVAL, ApplicationSettings, BoundingBoxSettings, CameraSettings, @@ -1652,9 +1653,14 @@ def _stop_multi_camera_recording(self) -> None: def worker(): try: - self._rec_manager.stop_all() - finally: - self._recording_stopped_async.emit() + while not self._rec_manager.stop_all(): + logger.info("Retrying recorder stop...") + time.sleep(RECORD_STOP_RETRY_INTERVAL) + except Exception as e: + logger.exception("Error while stopping recording: %s", e) + return + + self._recording_stopped_async.emit() threading.Thread( target=worker, @@ -2256,7 +2262,13 @@ def closeEvent(self, event: QCloseEvent) -> None: # pragma: no cover - GUI beha if hasattr(self, "_camera_validation_timer") and self._camera_validation_timer.isActive(): self._camera_validation_timer.stop() # Stop all multi-camera recorders - self._rec_manager.stop_all() + try: + self.multi_camera_controller.set_recording_frame_do_emit(False) + except Exception: + logger.exception("Failed to disable recording frame emission during shutdown") + while not self._rec_manager.stop_all(): + logger.info("Retrying recorder stop during shutdown...") + time.sleep(RECORD_STOP_RETRY_INTERVAL) # Close the camera dialog if open (ensures its worker thread is canceled) if getattr(self, "_cam_dialog", None) is not None and self._cam_dialog.isVisible(): From 87e549e33e9125388454e63da6ebecf97ccc923e Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Fri, 14 Aug 2026 10:54:29 +0200 Subject: [PATCH 21/21] Disable debug timing log --- dlclivegui/config.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dlclivegui/config.py b/dlclivegui/config.py index 962158a37..a6a06051c 100644 --- a/dlclivegui/config.py +++ b/dlclivegui/config.py @@ -35,7 +35,7 @@ SINGLE_CAMERA_WORKER_DO_LOG_TIMING: bool = False MULTI_CAMERA_WORKER_DO_LOG_TIMING: bool = False REC_DO_LOG_TIMING: bool = False -DLC_DO_LOG_TIMING: bool = True +DLC_DO_LOG_TIMING: bool = False ### Trigger debug logging DEBUG_TRIGGER_LOGS = False # MAIN_WINDOW_DO_LOG_TIMING: bool = False