From 6d2a168d714c978798dabd77d786192a9a49e339 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Wed, 1 Jul 2026 09:50:45 +0200 Subject: [PATCH 01/23] 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 e47dbe2f8..c33858a44 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 3010eeae4f61a2cb970a5aa317b79e86b4afb5f9 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Wed, 1 Jul 2026 10:06:38 +0200 Subject: [PATCH 02/23] 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 44d6281abe2b676d83df55619a9aed26e6cc1cca Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Wed, 1 Jul 2026 10:08:11 +0200 Subject: [PATCH 03/23] 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 | 5 +- 2 files changed, 50 insertions(+), 4 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 7ed769198..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__) @@ -165,7 +164,6 @@ def process(self, pose, **kwargs): # Calculate head angle relative to body cross = body_axis[0] * head_axis[1] - head_axis[0] * body_axis[1] sign = copysign(1, cross) # Positive when looking left - try: head_angle = acos(body_axis @ head_axis) * sign except ValueError: @@ -335,7 +333,6 @@ def process(self, pose, **kwargs): # Calculate head angle relative to body cross = body_axis[0] * head_axis[1] - head_axis[0] * body_axis[1] sign = copysign(1, cross) # Positive when looking left - try: head_angle = acos(body_axis @ head_axis) * sign except ValueError: From a568617a7b651b0af157464e78e44dbe2798d219 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Wed, 1 Jul 2026 10:12:22 +0200 Subject: [PATCH 04/23] 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 c33858a44..70653b0d7 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 9179b4e0ea428f297775126ffb0d629d09e27b7f Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Wed, 1 Jul 2026 10:13:13 +0200 Subject: [PATCH 05/23] 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 70653b0d7..e47dbe2f8 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 798b7de0097f3b2e99fb53b5489cd483b0e84216 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Wed, 1 Jul 2026 10:21:40 +0200 Subject: [PATCH 06/23] 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 f2a0204fc9819803a022e84f8180abc51b180d5e Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Wed, 1 Jul 2026 10:49:10 +0200 Subject: [PATCH 07/23] 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 bf6468b2538282f026b23378e4366516b48f5d4f Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 2 Jul 2026 10:01:59 +0200 Subject: [PATCH 08/23] Normalize recording container handling Centralized recording container configuration by introducing shared allowed/default container constants and using them across settings, UI, and utilities. The recording UI now populates container options from config and keeps the filename extension aligned with the selected container when switching between known video formats, so saved settings and path previews stay consistent. --- dlclivegui/config.py | 5 ++- dlclivegui/gui/main_window.py | 81 ++++++++++++++++++++++++----------- dlclivegui/utils/utils.py | 4 +- 3 files changed, 62 insertions(+), 28 deletions(-) diff --git a/dlclivegui/config.py b/dlclivegui/config.py index c24a0e085..85a48210c 100644 --- a/dlclivegui/config.py +++ b/dlclivegui/config.py @@ -25,6 +25,9 @@ GUI_MAX_DISPLAY_FPS: float = 30.0 ## Recording DEFAULT_RECORDING_FPS: float = 30.0 +ALLOWED_VIDEO_CONTAINERS: set[str] = {"mp4", "avi", "mov"} +DEFAULT_RECORDING_CONTAINER: str = "mp4" + ## Debug ### Timing logs @@ -508,7 +511,7 @@ class RecordingSettings(BaseModel): enabled: bool = False directory: str = Field(default_factory=lambda: str(Path.home() / "Videos" / "deeplabcut-live")) filename: str = "session.mp4" - container: Literal["mp4", "avi", "mov"] = "mp4" + container: Literal["mp4", "avi", "mov"] = DEFAULT_RECORDING_CONTAINER codec: str = "libx264" crf: int = Field(default=23, ge=0, le=51) fast_encoding: bool = False diff --git a/dlclivegui/gui/main_window.py b/dlclivegui/gui/main_window.py index 157d547cf..327ac308f 100644 --- a/dlclivegui/gui/main_window.py +++ b/dlclivegui/gui/main_window.py @@ -48,8 +48,10 @@ from dlclivegui.cameras import CameraFactory from dlclivegui.config import ( + ALLOWED_VIDEO_CONTAINERS, DEFAULT_CONFIG, GUI_MAX_DISPLAY_FPS, + DEFAULT_RECORDING_CONTAINER, ApplicationSettings, BoundingBoxSettings, CameraSettings, @@ -604,7 +606,7 @@ def _build_recording_group(self) -> QGroupBox: self.container_combo.setToolTip("Select the video container/format") self.container_combo.setSizePolicy(QSizePolicy.MinimumExpanding, QSizePolicy.Preferred) self.container_combo.setEditable(True) - self.container_combo.addItems(["mp4", "avi", "mov"]) + self.container_combo.addItems(sorted(ALLOWED_VIDEO_CONTAINERS)) # Ensure it never becomes unreadable: self.container_combo.setMinimumContentsLength(8) self.container_combo.setSizeAdjustPolicy(QComboBox.SizeAdjustPolicy.AdjustToMinimumContentsLengthWithIcon) @@ -820,7 +822,7 @@ def _connect_signals(self) -> None: self.use_timestamp_checkbox.stateChanged.connect(self._on_use_timestamp_changed) self.output_directory_edit.textChanged.connect(lambda _t: self._update_recording_path_preview()) self.filename_edit.textChanged.connect(lambda _t: self._update_recording_path_preview()) - self.container_combo.currentTextChanged.connect(lambda _t: self._update_recording_path_preview()) + self.container_combo.currentTextChanged.connect(self._on_container_changed) self.fast_encoding_checkbox.stateChanged.connect(self._on_fast_encoding_changed) # ------------------------------------------------------------------ @@ -958,11 +960,13 @@ def _dlc_settings_from_ui(self, *, allow_empty_model_path=False) -> DLCProcessor return DLCProcessorSettings.model_validate(updated_dlc) def _recording_settings_from_ui(self) -> RecordingSettings: + container = self.container_combo.currentText().strip() or DEFAULT_RECORDING_CONTAINER + filename = self._filename_matching_container(self.filename_edit.text().strip(), container) return RecordingSettings( enabled=True, # Always enabled - recording controlled by button directory=self.output_directory_edit.text().strip(), - filename=self.filename_edit.text().strip() or "session.mp4", - container=self.container_combo.currentText().strip() or "mp4", + filename=filename, + container=container, codec=self.codec_combo.currentText().strip() or "libx264", crf=int(self.crf_spin.value()), fast_encoding=bool( @@ -1178,31 +1182,52 @@ def _refresh_processors(self) -> None: # ------------------------------------------------------------------ # Recording path preview and session name persistence + def _known_recording_extensions(self) -> set[str]: + """Return known recording container extensions without leading dots.""" + known = ALLOWED_VIDEO_CONTAINERS.copy() + if hasattr(self, "container_combo"): + known.update( + self.container_combo.itemText(i).strip().lower().lstrip(".") + for i in range(self.container_combo.count()) + if self.container_combo.itemText(i).strip() + ) + return known + + def _filename_matching_container(self, filename: str, container: str) -> str: + """ + Adjust filename extension to match selected container, but only when + the existing extension is another known recording container. + """ + name = filename.strip() or "recording" + selected_ext = container.strip().lower().lstrip(".") + suffix = Path(name).suffix + + if not suffix or not selected_ext: + return name + + current_ext = suffix.lower().lstrip(".") + if current_ext in self._known_recording_extensions() and current_ext != selected_ext: + return str(Path(name).with_suffix(f".{selected_ext}")) + + return name + + def _on_container_changed(self, text: str) -> None: + """Keep filename extension aligned with selected container when safe.""" + if hasattr(self, "filename_edit"): + current = self.filename_edit.text() + updated = self._filename_matching_container(current, text) + if updated != current: + self.filename_edit.blockSignals(True) + self.filename_edit.setText(updated) + self.filename_edit.blockSignals(False) + + self._update_recording_path_preview() + def _on_session_name_editing_finished(self) -> None: name = self.session_name_edit.text().strip() self._settings_store.set_session_name(name) self._update_recording_path_preview() - # def _update_recording_path_preview(self) -> None: - # """Update the label showing where files will go (best-effort).""" - # if not hasattr(self, "recording_path_preview"): - # return - # out_dir = self.output_directory_edit.text().strip() - # sess = self.session_name_edit.text().strip() if hasattr(self, "session_name_edit") else "" - # base = self.filename_edit.text().strip() - # container = self.container_combo.currentText().strip() if hasattr(self, "container_combo") else "mp4" - # use_ts = self.use_timestamp_checkbox.isChecked() if hasattr(self, "use_timestamp_checkbox") else True - - # # Preview is approximate (since run index/time is decided at start). - # sess_safe = sess.strip() or "session" - # run_hint = "run_" if use_ts else "run_" - # stem_hint = Path(base).stem if base.strip() else "recording" # shows user-provided stem or default - # full_hint = str(Path(out_dir).expanduser() / sess_safe / run_hint / f"{stem_hint}_.{container}") - # self.recording_path_preview.setText(f"{full_hint}") - # self.recording_path_preview.setToolTip( - # f"Click to copy to clipboard :
{full_hint.replace('', '*')}" - # ) - def _update_recording_path_preview(self) -> None: """Update the label showing where files will go (best-effort).""" if not hasattr(self, "recording_path_preview"): @@ -1210,8 +1235,12 @@ def _update_recording_path_preview(self) -> None: out_dir = self.output_directory_edit.text().strip() sess = self.session_name_edit.text().strip() if hasattr(self, "session_name_edit") else "" - base = self.filename_edit.text().strip() - container = self.container_combo.currentText().strip() if hasattr(self, "container_combo") else "mp4" + container = ( + self.container_combo.currentText().strip() + if hasattr(self, "container_combo") + else DEFAULT_RECORDING_CONTAINER + ) + base = self._filename_matching_container(self.filename_edit.text(), container) use_ts = self.use_timestamp_checkbox.isChecked() if hasattr(self, "use_timestamp_checkbox") else True # Preview is approximate (since run index/time is decided at start). diff --git a/dlclivegui/utils/utils.py b/dlclivegui/utils/utils.py index 6af003dad..534e5732f 100644 --- a/dlclivegui/utils/utils.py +++ b/dlclivegui/utils/utils.py @@ -8,6 +8,8 @@ from datetime import datetime from pathlib import Path +from dlclivegui.config import DEFAULT_RECORDING_CONTAINER + _INVALID_CHARS = re.compile(r"[^A-Za-z0-9._-]+") @@ -36,7 +38,7 @@ def split_stem_ext(base_filename: str, container: str) -> tuple[str, str]: If user typed an extension, keep it. Else use container. """ base = (base_filename or "").strip() - container = (container or "mp4").strip().lstrip(".") or "mp4" + container = (container or DEFAULT_RECORDING_CONTAINER).strip().lstrip(".") or DEFAULT_RECORDING_CONTAINER if not base: base = "recording" From 6819af1edc3eb81ccad7358ef82f252c4425e5a5 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 2 Jul 2026 10:50:25 +0200 Subject: [PATCH 09/23] Add DLC timing instrumentation in GUI path Introduce a new `DLC_DO_LOG_TIMING` config flag and wire `WorkerTimingStats` into `DLCLiveMainWindow` for DLC enqueue and pose-ready callback timing. The pose callback now logs camera-to-GUI latency in debug mode, marks display state dirty instead of forcing an immediate redraw, and emits periodic timing stats via `maybe_log()`. --- dlclivegui/config.py | 1 + dlclivegui/gui/main_window.py | 34 ++++++++++++++++++++++++++++------ 2 files changed, 29 insertions(+), 6 deletions(-) diff --git a/dlclivegui/config.py b/dlclivegui/config.py index 85a48210c..8f80464e3 100644 --- a/dlclivegui/config.py +++ b/dlclivegui/config.py @@ -34,6 +34,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 ### Trigger debug logging DEBUG_TRIGGER_LOGS = False # MAIN_WINDOW_DO_LOG_TIMING: bool = False diff --git a/dlclivegui/gui/main_window.py b/dlclivegui/gui/main_window.py index 327ac308f..2c02e7de2 100644 --- a/dlclivegui/gui/main_window.py +++ b/dlclivegui/gui/main_window.py @@ -52,6 +52,7 @@ DEFAULT_CONFIG, GUI_MAX_DISPLAY_FPS, DEFAULT_RECORDING_CONTAINER, + DLC_DO_LOG_TIMING, ApplicationSettings, BoundingBoxSettings, CameraSettings, @@ -71,7 +72,7 @@ from ..services.multi_camera_controller import MultiCameraController, MultiFrameData, get_camera_id, get_display_id 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 format_dlc_stats +from ..utils.stats import WorkerTimingStats, format_dlc_stats from ..utils.utils import FPSTracker from .camera_config.camera_config_dialog import CameraConfigDialog from .misc import color_dropdowns as color_ui @@ -132,6 +133,10 @@ def __init__(self, config: ApplicationSettings | None = None): self._rec_manager = RecordingManager() self._dlc = DLCLiveProcessor() self.multi_camera_controller = MultiCameraController() + ### Time debug + self._dlc_timing = WorkerTimingStats( + "GUI - DLC Worker", logger=logger, log_interval=2.0, enabled=DLC_DO_LOG_TIMING + ) self._config = config self._inference_camera_id: str | None = None # Camera ID used for inference @@ -1520,7 +1525,11 @@ def _on_multi_frame_processing_ready(self, frame_data: MultiFrameData) -> None: if self._dlc_active and is_dlc_camera_frame and dlc_cam_id in frame_data.frames: frame = frame_data.frames[dlc_cam_id] timestamp = frame_data.timestamps.get(dlc_cam_id, time.time()) - self._dlc.enqueue_frame(frame, timestamp) + with self._dlc_timing.measure("enqueue_frame"): + self._dlc.enqueue_frame(frame, timestamp) + + self._dlc_timing.note_frame() + self._dlc_timing.maybe_log() def _on_multi_frame_display_ready(self, frame_data: MultiFrameData) -> None: """Throttled UI/display path. @@ -2080,10 +2089,23 @@ def _stop_recording(self) -> None: def _on_pose_ready(self, result: PoseResult) -> None: if not self._dlc_active: return - self._last_pose = result - # logger.debug(f"Pose result: {result.pose}, Timestamp: {result.timestamp}") - if self._current_frame is not None: - self._display_frame(self._current_frame, force=True) + + with self._dlc_timing.measure("DLC.pose_ready_callback"): + self._last_pose = result + + try: + latency_ms = (time.time() - float(result.timestamp)) * 1000.0 + if logger.isEnabledFor(logging.DEBUG): + logger.debug("DLC pose latency camera_timestamp_to_gui=%.2f ms", latency_ms) + except Exception: + pass + + if self._current_frame is not None: + self._display_dirty = True + # with self._dlc_timing.measure("DLC.display_after_pose"): + # self._display_frame(self._current_frame, force=True) + + self._dlc_timing.maybe_log() def _on_dlc_error(self, message: str) -> None: self._stop_inference(show_message=False) From 4fa11638a279f5b4ff6b8452651f323a21bcae3b Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 2 Jul 2026 10:54:05 +0200 Subject: [PATCH 10/23] Prioritize latest frame when queue is full Update `_enqueue_frame` to keep enqueueing the newest frame by removing one queued item when `put_nowait` hits `queue.Full`, instead of dropping the incoming frame. This makes processing more real-time under load and keeps enqueue/drop stats consistent, including safe `task_done()` handling. --- dlclivegui/services/dlc_processor.py | 29 +++++++++++++++++++++------- 1 file changed, 22 insertions(+), 7 deletions(-) diff --git a/dlclivegui/services/dlc_processor.py b/dlclivegui/services/dlc_processor.py index b4476e116..0ecf8abca 100644 --- a/dlclivegui/services/dlc_processor.py +++ b/dlclivegui/services/dlc_processor.py @@ -258,13 +258,28 @@ def enqueue_frame(self, frame: np.ndarray, timestamp: float) -> None: if q is None: return - try: - q.put_nowait((frame_c, timestamp, enq_time)) - with self._stats_lock: - self._frames_enqueued += 1 - except queue.Full: - with self._stats_lock: - self._frames_dropped += 1 + item = (frame_c, timestamp, enq_time) + + while True: + try: + q.put_nowait(item) + with self._stats_lock: + self._frames_enqueued += 1 + return + + except queue.Full: + try: + q.get_nowait() + try: + q.task_done() + except ValueError: + pass + + with self._stats_lock: + self._frames_dropped += 1 + + except queue.Empty: + continue def get_stats(self) -> ProcessorStats: """Get current processing statistics.""" From 4c3ef7f8d0ac019a232e64848846fcf49a0ec236 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 2 Jul 2026 11:16:52 +0200 Subject: [PATCH 11/23] Lower camera backend logs to debug Demoted multiple verbose runtime messages from INFO to DEBUG in Basler and GenTL backends. This keeps normal logs cleaner by moving routine configuration/readback details (FPS setup, converter mode, exposure/gain settings, trigger configuration, and startup/close diagnostics) out of INFO-level output while preserving the diagnostics when DEBUG is enabled. --- dlclivegui/cameras/backends/basler_backend.py | 26 +++++++++---------- dlclivegui/cameras/backends/gentl_backend.py | 18 ++++++------- 2 files changed, 22 insertions(+), 22 deletions(-) diff --git a/dlclivegui/cameras/backends/basler_backend.py b/dlclivegui/cameras/backends/basler_backend.py index 09d23c7bb..319f5e5f7 100644 --- a/dlclivegui/cameras/backends/basler_backend.py +++ b/dlclivegui/cameras/backends/basler_backend.py @@ -447,7 +447,7 @@ def _configure_frame_rate(self) -> None: fps = self._positive_float(getattr(self.settings, "fps", 0.0)) if fps is None: - LOG.info("[Basler] FPS: auto/free-run, not forcing AcquisitionFrameRate") + LOG.debug("[Basler] FPS: auto/free-run, not forcing AcquisitionFrameRate") return enable = self._feature("AcquisitionFrameRateEnable") @@ -464,7 +464,7 @@ def _configure_frame_rate(self) -> None: try: min_v = rate.GetMin() max_v = rate.GetMax() - LOG.info("[Basler] AcquisitionFrameRate range: min=%s max=%s requested=%s", min_v, max_v, fps) + LOG.debug("[Basler] AcquisitionFrameRate range: min=%s max=%s requested=%s", min_v, max_v, fps) except Exception: pass @@ -495,7 +495,7 @@ def _configure_frame_rate(self) -> None: if feature is not None: readbacks[name] = self._feature_value(feature, None) - LOG.info("[Basler] FPS readback requested=%s values=%s", fps, readbacks) + LOG.debug("[Basler] Readback requested=%s values=%s", fps, readbacks) try: self._actual_fps = float(readbacks.get("AcquisitionFrameRate")) @@ -520,14 +520,14 @@ def _configure_converter(self) -> None: if self._should_output_mono(): self._converter.OutputPixelFormat = pylon.PixelType_Mono8 - LOG.info( + LOG.debug( "[Basler] Converter configured for Mono8 output (camera PixelFormat=%s preserve_mono=%s)", camera_pixel_format, self._preserve_mono, ) else: self._converter.OutputPixelFormat = pylon.PixelType_BGR8packed - LOG.info( + LOG.debug( "[Basler] Converter configured for BGR8 output (camera PixelFormat=%s preserve_mono=%s)", camera_pixel_format, self._preserve_mono, @@ -558,7 +558,7 @@ def open(self) -> None: self._camera.ExposureTime.SetValue(float(self.settings.exposure)) if hasattr(self._camera, "ExposureTimeAbs"): self._camera.ExposureTimeAbs.SetValue(float(self.settings.exposure)) - LOG.info("[Basler] Exposure set to %s us (auto off)", self.settings.exposure) + LOG.debug("[Basler] Exposure set to %s us (auto off)", self.settings.exposure) except Exception as exc: LOG.warning("[Basler] Failed to set exposure: %s", exc) @@ -568,7 +568,7 @@ def open(self) -> None: if hasattr(self._camera, "GainAuto"): self._camera.GainAuto.SetValue("Off") self._camera.Gain.SetValue(float(self.settings.gain)) - LOG.info("[Basler] Gain set to %s dB (auto off)", self.settings.gain) + LOG.debug("[Basler] Gain set to %s dB (auto off)", self.settings.gain) except Exception as exc: LOG.warning("[Basler] Failed to set gain: %s", exc) @@ -648,7 +648,7 @@ def open(self) -> None: # pylon.GrabStrategy_LatestImageOnly, pylon.GrabStrategy_OneByOne, ) - LOG.info( + LOG.debug( "[Basler] grabbing=%s max_buffers=%s", self._camera.IsGrabbing(), self._camera.MaxNumBuffer.GetValue() if hasattr(self._camera, "MaxNumBuffer") else "N/A", @@ -656,7 +656,7 @@ def open(self) -> None: else: LOG.debug("Fast-start probe: skipping StartGrabbing and converter") - LOG.info( + LOG.debug( "[Basler] open device_id=%s index=%s fast_start=%s requested=(%sx%s @ %s fps exp=%s gain=%s)", getattr(self, "_device_id", None), getattr(self.settings, "index", None), @@ -766,7 +766,7 @@ def read(self) -> CapturedFrame: if not self._logged_first_frame: self._logged_first_frame = True - LOG.info( + LOG.debug( "[Basler] first frame device_id=%s shape=%s dtype=%s nbytes=%.2f MB " "camera_pixel_format=%s output_format=%s preserve_mono=%s", self._device_id, @@ -817,7 +817,7 @@ def read(self) -> CapturedFrame: raise RuntimeError("Failed to retrieve image from Basler camera.") from exc def close(self) -> None: - LOG.info( + LOG.debug( "[Basler] close called camera_exists=%s grabbing=%s open=%s", self._camera is not None, bool(self._camera and self._camera.IsGrabbing()), @@ -1226,7 +1226,7 @@ def _configure_trigger_input(self, cfg, *, strict: bool = False) -> None: self._trigger = CameraTriggerSettings() return - LOG.info( + LOG.debug( "Basler trigger input configured: role=%s selector=%s source=%s activation=%s " "selector_ok=%s source_ok=%s activation_ok=%s", role, @@ -1291,7 +1291,7 @@ def _configure_trigger_master(self, cfg, *, strict: bool = False) -> None: source_ok = self._set_enum_feature("LineSource", output_source, strict=strict) if mode_ok and source_ok: - LOG.info( + LOG.debug( "Basler trigger master configured via Line*: output_line=%s output_source=%s", output_line, output_source, diff --git a/dlclivegui/cameras/backends/gentl_backend.py b/dlclivegui/cameras/backends/gentl_backend.py index 76d0be5ae..89dac6511 100644 --- a/dlclivegui/cameras/backends/gentl_backend.py +++ b/dlclivegui/cameras/backends/gentl_backend.py @@ -1290,7 +1290,7 @@ def _resolve_trigger_source(self, node_map, requested: str, *, strict: bool) -> if requested.lower() == "auto": for candidate in ("Line0", "Line1", "Line2", "Any"): if candidate in available: - LOG.info( + LOG.debug( "GenTL TriggerSource auto-selected '%s'. Available: %s", candidate, available, @@ -1479,7 +1479,7 @@ def _configure_trigger_input(self, node_map, cfg, *, strict: bool = False) -> No self._trigger = CameraTriggerSettings() return - LOG.info( + LOG.debug( "GenTL trigger input configured: role=%s selector=%s source_requested=%s " "source=%s activation=%s selector_ok=%s source_ok=%s activation_ok=%s", role, @@ -1540,7 +1540,7 @@ def _configure_trigger_master(self, node_map, cfg, *, strict: bool = False) -> N node = self._node(node_map, "StrobeDuration") if node is not None: node.value = int(strobe_duration) - LOG.info("Configured GenTL StrobeDuration=%s", int(strobe_duration)) + LOG.debug("Configured GenTL StrobeDuration=%s", int(strobe_duration)) except Exception as exc: if strict: raise RuntimeError(f"Failed to set StrobeDuration={strobe_duration}: {exc}") from exc @@ -1551,7 +1551,7 @@ def _configure_trigger_master(self, node_map, cfg, *, strict: bool = False) -> N node = self._node(node_map, "StrobeDelay") if node is not None: node.value = int(strobe_delay) - LOG.info("Configured GenTL StrobeDelay=%s", int(strobe_delay)) + LOG.debug("Configured GenTL StrobeDelay=%s", int(strobe_delay)) except Exception as exc: if strict: raise RuntimeError(f"Failed to set StrobeDelay={strobe_delay}: {exc}") from exc @@ -1565,7 +1565,7 @@ def _configure_trigger_master(self, node_map, cfg, *, strict: bool = False) -> N ) if enable_ok: - LOG.info( + LOG.debug( "GenTL trigger master configured via Strobe*: " "StrobeEnable=On StrobePolarity=%s polarity_ok=%s " "StrobeOperation=%s operation_ok=%s", @@ -1605,7 +1605,7 @@ def _configure_trigger_master(self, node_map, cfg, *, strict: bool = False) -> N source_ok = self._set_enum_node(node_map, "LineSource", output_source, strict=strict) if mode_ok and source_ok: - LOG.info( + LOG.debug( "GenTL trigger master configured via Line*: output_line=%s output_source=%s", output_line, output_source, @@ -1724,7 +1724,7 @@ def _configure_frame_rate(self, node_map) -> None: return target = float(self.settings.fps) - LOG.info("Configuring GenTL frame rate: requested %.3f FPS", target) + LOG.debug("Configuring GenTL frame rate: requested %.3f FPS", target) for attr in ("AcquisitionFrameRateEnable", "AcquisitionFrameRateControlEnable"): try: @@ -1732,7 +1732,7 @@ def _configure_frame_rate(self, node_map) -> None: before = getattr(node, "value", None) node.value = True after = getattr(node, "value", None) - LOG.info("Enabled GenTL %s: before=%r after=%r", attr, before, after) + LOG.debug("Enabled GenTL %s: before=%r after=%r", attr, before, after) break except Exception: pass @@ -1744,7 +1744,7 @@ def _configure_frame_rate(self, node_map) -> None: node.value = target after = getattr(node, "value", None) - LOG.info( + LOG.debug( "Set GenTL %s: before=%r requested=%.3f after=%r", attr, before, From a789bf8158391d9b4747be8f38b554f129ff1e3a Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 2 Jul 2026 13:30:00 +0200 Subject: [PATCH 12/23] Harden DLC worker startup and timing logs Adds fine-grained `WorkerTimingStats` instrumentation across enqueue, initialization, inference, and emit paths, with error/frame accounting and optional timing logging. It also makes worker startup/stop behavior safer by deferring queue creation until RUNNING, blocking enqueue during STARTING, normalizing input frames before inference, and adding richer debug diagnostics (CUDA/runner state and thread stack dumps on stuck shutdown). --- dlclivegui/services/dlc_processor.py | 286 ++++++++++++++++++++++----- dlclivegui/utils/stats.py | 5 +- dlclivegui/utils/utils.py | 16 ++ 3 files changed, 256 insertions(+), 51 deletions(-) diff --git a/dlclivegui/services/dlc_processor.py b/dlclivegui/services/dlc_processor.py index 0ecf8abca..4d4df7584 100644 --- a/dlclivegui/services/dlc_processor.py +++ b/dlclivegui/services/dlc_processor.py @@ -16,9 +16,11 @@ import numpy as np from PySide6.QtCore import QObject, Signal -from dlclivegui.config import DLCProcessorSettings, ModelType +from dlclivegui.config import DLC_DO_LOG_TIMING, DLCProcessorSettings, ModelType from dlclivegui.processors.processor_utils import instantiate_from_scan from dlclivegui.temp import Engine # type: ignore # TODO use main package enum when released +from dlclivegui.utils.stats import WorkerTimingStats +from dlclivegui.utils.utils import format_thread_stack logger = logging.getLogger(__name__) STOP_WORKER_TIMEOUT = 10.0 # # seconds to wait in STOPPING state before scheduling background reaping @@ -181,6 +183,13 @@ def __init__(self) -> None: self._gpu_inference_times: deque[float] = deque(maxlen=60) self._processor_overhead_times: deque[float] = deque(maxlen=60) + self._timing = WorkerTimingStats( + "DLCLiveProcessor", + logger=logger, + log_interval=1.0, + enabled=bool(DLC_DO_LOG_TIMING or ENABLE_PROFILING), + ) + @staticmethod def get_model_backend(model_path: str) -> Engine: return Engine.from_model_path(model_path) @@ -232,24 +241,40 @@ def shutdown(self) -> None: def enqueue_frame(self, frame: np.ndarray, timestamp: float) -> None: # Keep lifecycle lock held only for quick state checks and snapshots. with self._lifecycle_lock: - if self._state in (WorkerState.STOPPING, WorkerState.FAULTED) or self._stop_event.is_set(): + if ( + self._state in (WorkerState.STOPPING, WorkerState.FAULTED, WorkerState.STARTING) + or self._stop_event.is_set() + ): return t = self._worker_thread q = self._queue should_start = t is None or not t.is_alive() - frame_c = frame.copy() + with self._timing.measure("DLC.enqueue.copy_frame"): + frame_c = frame.copy() enq_time = time.perf_counter() if should_start: # Re-acquire the lifecycle lock to safely (re)start the worker if needed. with self._lifecycle_lock: # Re-check state in case it changed while we were copying the frame. - if self._state in (WorkerState.STOPPING, WorkerState.FAULTED) or self._stop_event.is_set(): + if ( + self._state in (WorkerState.STOPPING, WorkerState.FAULTED, WorkerState.STARTING) + or self._stop_event.is_set() + ): return t = self._worker_thread if t is None or not t.is_alive(): - # _start_worker_locked expects the lifecycle lock to be held. + if logger.isEnabledFor(logging.DEBUG): + logger.debug( + "Starting DLC worker from first frame: " + "shape=%s dtype=%s contiguous=%s strides=%s timestamp=%.6f", + frame_c.shape, + frame_c.dtype, + frame_c.flags["C_CONTIGUOUS"], + frame_c.strides, + timestamp, + ) self._start_worker_locked(frame_c, timestamp) return # Worker is now running; refresh queue snapshot. @@ -262,18 +287,20 @@ def enqueue_frame(self, frame: np.ndarray, timestamp: float) -> None: while True: try: - q.put_nowait(item) + with self._timing.measure("DLC.enqueue.put"): + q.put_nowait(item) with self._stats_lock: self._frames_enqueued += 1 return except queue.Full: try: - q.get_nowait() - try: - q.task_done() - except ValueError: - pass + with self._timing.measure("DLC.enqueue.drop_stale"): + q.get_nowait() + try: + q.task_done() + except ValueError: + pass with self._stats_lock: self._frames_dropped += 1 @@ -332,11 +359,91 @@ def get_stats(self) -> ProcessorStats: avg_processor_overhead=avg_proc_overhead, ) + def _debug_log_dlc_runner_device(self) -> None: + if not logger.isEnabledFor(logging.DEBUG): + return + + try: + import torch + + logger.debug( + "Torch CUDA state: available=%s built=%s device_count=%s current_device=%s device_name=%s " + "allocated=%.2fMB reserved=%.2fMB", + torch.cuda.is_available(), + torch.backends.cuda.is_built(), + torch.cuda.device_count(), + torch.cuda.current_device() if torch.cuda.is_available() else None, + torch.cuda.get_device_name(0) if torch.cuda.is_available() and torch.cuda.device_count() else None, + torch.cuda.memory_allocated(0) / (1024 * 1024) if torch.cuda.is_available() else 0.0, + torch.cuda.memory_reserved(0) / (1024 * 1024) if torch.cuda.is_available() else 0.0, + ) + except Exception: + logger.debug("Could not query torch CUDA state", exc_info=True) + + dlc = self._dlc + runner = getattr(dlc, "runner", None) + + logger.debug( + "DLCLive runner: type=%s runner.device=%r runner.model=%r runner.net=%r", + type(runner).__name__ if runner is not None else None, + getattr(runner, "device", None), + type(getattr(runner, "model", None)).__name__ if getattr(runner, "model", None) is not None else None, + type(getattr(runner, "net", None)).__name__ if getattr(runner, "net", None) is not None else None, + ) + + seen: set[int] = set() + + def walk(obj, path: str, depth: int = 0) -> None: + if obj is None or depth > 7: + return + + oid = id(obj) + if oid in seen: + return + seen.add(oid) + + try: + params = getattr(obj, "parameters", None) + if callable(params): + first_param = next(iter(params()), None) + if first_param is not None: + logger.debug( + "Torch module at %s: parameter device=%s is_cuda=%s dtype=%s shape=%s", + path, + first_param.device, + first_param.is_cuda, + first_param.dtype, + tuple(first_param.shape), + ) + except Exception: + pass + + for name in ( + "runner", + "model", + "net", + "pose_model", + "dlc_model", + "module", + "engine", + "predictor", + "detector", + "backbone", + ): + try: + child = getattr(obj, name, None) + except Exception: + child = None + if child is not None: + walk(child, f"{path}.{name}", depth + 1) + + walk(self._dlc, "self._dlc") + def _start_worker_locked(self, init_frame: np.ndarray, init_timestamp: float) -> None: # lifecycle_lock must already be held if self._worker_thread is not None and self._worker_thread.is_alive(): return - self._queue = queue.Queue(maxsize=1) + self._queue = None self._stop_event.clear() self._state = WorkerState.STARTING self._worker_thread = threading.Thread( @@ -364,7 +471,7 @@ def _stop_worker(self) -> bool: t.join(timeout=STOP_WORKER_TIMEOUT) if t.is_alive(): qsize = self._queue.qsize() if self._queue is not None else -1 - logger.warning("DLC worker thread did not terminate cleanly (qsize=%s)", qsize) + logger.warning("DLC worker thread did not terminate cleanly (qsize=%s)\n%s", qsize, format_thread_stack(t)) self._schedule_reap(t) return False @@ -427,7 +534,8 @@ def _timed_processor(self): def timed_process(pose, _op=original, _holder=holder, **kwargs): start = time.perf_counter() try: - return _op(pose, **kwargs) + with self._timing.measure("DLC.processor.process"): + return _op(pose, **kwargs) finally: _holder[0] = time.perf_counter() - start @@ -438,6 +546,24 @@ def timed_process(pose, _op=original, _holder=holder, **kwargs): # Restore even if inference/errors occur self._processor.process = original + @staticmethod + def _prepare_input_frame(frame: np.ndarray) -> np.ndarray: + """Normalize camera frames for DLCLive inference.""" + arr = np.asarray(frame) + + if arr.ndim == 2: + # Mono8 / grayscale -> 3-channel + arr = np.repeat(arr[:, :, None], 3, axis=2) + elif arr.ndim == 3 and arr.shape[2] == 4: + arr = arr[:, :, :3] + elif arr.ndim != 3 or arr.shape[2] != 3: + raise ValueError(f"Unsupported DLCLive input frame shape: {arr.shape}") + + if arr.dtype != np.uint8: + arr = np.clip(arr, 0, 255).astype(np.uint8, copy=False) + + return np.ascontiguousarray(arr) + def _process_frame( self, frame: np.ndarray, @@ -453,11 +579,23 @@ def _process_frame( if self._dlc is None: raise RuntimeError("DLCLive instance is not initialized.") # Time GPU inference (and processor overhead when present) + with self._timing.measure("DLC.prepare_frame"): + frame = self._prepare_input_frame(frame) with self._timed_processor() as proc_holder: inference_start = time.perf_counter() - raw_pose: Any = self._dlc.get_pose(frame, frame_time=timestamp) + + with self._timing.measure("DLC.process_frame"): + processed_frame = self._dlc.process_frame(frame) + + with self._timing.measure("DLC.runner.get_pose"): + self._dlc.pose = self._dlc.runner.get_pose(processed_frame) + + with self._timing.measure("DLC.post_process_pose"): + raw_pose: Any = self._dlc._post_process_pose(processed_frame, frame_time=timestamp) + inference_time = time.perf_counter() - inference_start - pose_arr: np.ndarray = validate_pose_array(raw_pose, source_backend=PoseBackends.DLC_LIVE) + with self._timing.measure("DLC.validate_pose"): + pose_arr: np.ndarray = validate_pose_array(raw_pose, source_backend=PoseBackends.DLC_LIVE) pose_packet = PosePacket( schema_version=0, keypoints=pose_arr, @@ -475,7 +613,8 @@ def _process_frame( # Emit pose (measure signal overhead) signal_start = time.perf_counter() - self.pose_ready.emit(PoseResult(pose=pose_packet.keypoints, timestamp=timestamp, packet=pose_packet)) + with self._timing.measure("DLC.emit.pose_ready"): + self.pose_ready.emit(PoseResult(pose=pose_packet.keypoints, timestamp=timestamp, packet=pose_packet)) signal_time = time.perf_counter() - signal_start end_ts = time.perf_counter() @@ -496,6 +635,8 @@ def _process_frame( self._gpu_inference_times.append(gpu_inference_time) self._processor_overhead_times.append(processor_overhead) + self._timing.note_frame() + self._timing.maybe_log() self.frame_processed.emit() def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None: @@ -504,60 +645,98 @@ def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None: if not self._settings.model_path: raise RuntimeError("No DLCLive model path configured.") - init_start = time.perf_counter() - dyn = self._settings.dynamic - if not isinstance(dyn, (list, tuple)) or len(dyn) != 3: - try: - dyn = dyn.to_tuple() - except Exception as e: - raise RuntimeError("Invalid dynamic crop settings format.") from e - enabled, margin, max_missing = dyn - - options = { - "model_path": self._settings.model_path, - "model_type": self._settings.model_type, - "processor": self._processor, - "dynamic": [enabled, margin, max_missing], - "resize": self._settings.resize, - "precision": self._settings.precision, - "single_animal": self._settings.single_animal, - } - if self._settings.device is not None: - options["device"] = self._settings.device + with self._timing.measure("DLC.build_options"): + dyn = self._settings.dynamic + if not isinstance(dyn, (list, tuple)) or len(dyn) != 3: + try: + dyn = dyn.to_tuple() + except Exception as e: + raise RuntimeError("Invalid dynamic crop settings format.") from e + enabled, margin, max_missing = dyn + + options = { + "model_path": self._settings.model_path, + "model_type": self._settings.model_type, + "processor": self._processor, + "dynamic": [enabled, margin, max_missing], + "resize": self._settings.resize, + "precision": self._settings.precision, + "single_animal": self._settings.single_animal, + } + if self._settings.device is not None: + options["device"] = self._settings.device + + if logger.isEnabledFor(logging.DEBUG): + logger.debug( + "DLC worker starting: model_path=%s model_type=%s device=%s " + "init_frame_shape=%s dtype=%s contiguous=%s", + self._settings.model_path, + self._settings.model_type, + self._settings.device, + init_frame.shape, + init_frame.dtype, + init_frame.flags["C_CONTIGUOUS"], + ) try: if DLCLive is None: raise RuntimeError( "DLCLive class is not available. Ensure the dlclive package is installed and can be imported." ) - self._dlc = DLCLive(**options) + with self._timing.measure("DLC.construct"): + self._dlc = DLCLive(**options) + self._timing.maybe_log() except Exception as exc: + self._timing.note_error() + self._timing.maybe_log() with self._lifecycle_lock: self._state = WorkerState.FAULTED raise RuntimeError( f"Failed to initialize DLCLive with model '{self._settings.model_path}': {exc}" ) from exc + if self._stop_event.is_set(): + logger.debug("DLC worker stop requested during construction; exiting before init_inference.") + return + + with self._timing.measure("DLC.prepare_init_frame"): + init_frame = self._prepare_input_frame(init_frame) + + if logger.isEnabledFor(logging.DEBUG): + logger.debug( + "Calling DLCLive.init_inference with frame shape=%s dtype=%s contiguous=%s", + init_frame.shape, + init_frame.dtype, + init_frame.flags["C_CONTIGUOUS"], + ) # First inference to initialize - init_inference_start = time.perf_counter() - self._dlc.init_inference(init_frame) - init_inference_time = time.perf_counter() - init_inference_start + with self._timing.measure("DLC.init_inference"): + self._dlc.init_inference(init_frame) + + self._debug_log_dlc_runner_device() + self._timing.note_frame() + self._timing.maybe_log() + + if self._stop_event.is_set(): + logger.debug("DLC worker stop requested after init_inference; exiting before RUNNING state.") + return # Pass DLCLive cfg to processor if available if hasattr(self._dlc, "processor") and hasattr(self._dlc.processor, "set_dlc_cfg"): - self._dlc.processor.set_dlc_cfg(getattr(self._dlc, "cfg", None)) + with self._timing.measure("DLC.processor.set_dlc_cfg"): + self._dlc.processor.set_dlc_cfg(getattr(self._dlc, "cfg", None)) self._initialized = True self.initialized.emit(True) with self._lifecycle_lock: + if self._stop_event.is_set(): + logger.debug("DLC worker stop requested before RUNNING state; exiting.") + return + + self._queue = queue.Queue(maxsize=1) self._state = WorkerState.RUNNING - total_init_time = time.perf_counter() - init_start - logger.info( - "DLCLive model initialized successfully (total: %.3fs, init_inference: %.3fs)", - total_init_time, - init_inference_time, - ) + logger.info("DLCLive model initialized successfully") # Emit pose for init frame & update stats (not dequeued) self._process_frame(init_frame, init_timestamp, time.perf_counter(), queue_wait_time=0.0) @@ -598,6 +777,8 @@ def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None: try: self._process_frame(frame, ts, enq, queue_wait_time=0.0) except Exception as exc: + self._timing.note_error() + self._timing.maybe_log() logger.exception("Pose inference failed", exc_info=exc) self.error.emit(str(exc)) finally: @@ -610,11 +791,14 @@ def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None: # Normal operation: timed get try: wait_start = time.perf_counter() - item = q.get(timeout=0.05) + with self._timing.measure("DLC.queue_get"): + item = q.get(timeout=0.05) queue_wait_time = time.perf_counter() - wait_start except queue.Empty: continue except Exception as exc: + self._timing.note_error() + self._timing.maybe_log() logger.exception("Error getting item from queue", exc_info=exc) with self._lifecycle_lock: self._state = WorkerState.FAULTED @@ -625,6 +809,8 @@ def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None: frame, ts, enq = item self._process_frame(frame, ts, enq, queue_wait_time=queue_wait_time) except Exception as exc: + self._timing.note_error() + self._timing.maybe_log() logger.exception("Pose inference failed", exc_info=exc) self.error.emit(str(exc)) finally: @@ -635,6 +821,8 @@ def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None: logger.info("DLC worker thread exiting") + self._timing.maybe_log() + class DLCService: """Wrap DLCLiveProcessor lifecycle & configuration.""" diff --git a/dlclivegui/utils/stats.py b/dlclivegui/utils/stats.py index 1edbf7890..0ef0528e2 100644 --- a/dlclivegui/utils/stats.py +++ b/dlclivegui/utils/stats.py @@ -70,6 +70,7 @@ def __init__(self, parent: WorkerTimingStats, name: str): self.parent = parent self.name = name self.t0 = 0.0 + self.elapsed = 0.0 def __enter__(self): if self.parent.enabled: @@ -80,8 +81,8 @@ def __exit__(self, exc_type, exc, tb): if not self.parent.enabled: return False - dt = time.perf_counter() - self.t0 - self.parent._totals[self.name] = self.parent._totals.get(self.name, 0.0) + dt + self.elapsed = time.perf_counter() - self.t0 + self.parent._totals[self.name] = self.parent._totals.get(self.name, 0.0) + self.elapsed self.parent._counts[self.name] = self.parent._counts.get(self.name, 0) + 1 return False diff --git a/dlclivegui/utils/utils.py b/dlclivegui/utils/utils.py index 534e5732f..bd72958cd 100644 --- a/dlclivegui/utils/utils.py +++ b/dlclivegui/utils/utils.py @@ -1,7 +1,10 @@ from __future__ import annotations import re +import sys +import threading import time +import traceback from collections import deque from collections.abc import Iterable from dataclasses import dataclass @@ -87,6 +90,19 @@ def build_run_dir(session_dir: Path, *, use_timestamp: bool) -> Path: return run_dir +def format_thread_stack(thread: threading.Thread) -> str: + ident = thread.ident + if ident is None: + return f"Thread {thread.name!r} has no ident." + + frame = sys._current_frames().get(ident) + if frame is None: + return f"No Python stack frame found for thread {thread.name!r} ident={ident}." + + stack = "".join(traceback.format_stack(frame)) + return f"Stack for thread {thread.name!r} ident={ident}:\n{stack}" + + @dataclass(frozen=True) class RecordingPlan: session_dir: Path From 2f7fbf3cc2e7d4ab1a1384ea0ac8558b4e3d086b Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 2 Jul 2026 13:30:37 +0200 Subject: [PATCH 13/23] Disable pose latency debug logging Comment out the camera-to-GUI pose latency calculation and debug log in `pose_ready_callback`. This removes the try/except-wrapped timing log path while leaving pose handling and display update behavior unchanged. --- dlclivegui/gui/main_window.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/dlclivegui/gui/main_window.py b/dlclivegui/gui/main_window.py index 2c02e7de2..b4d10831a 100644 --- a/dlclivegui/gui/main_window.py +++ b/dlclivegui/gui/main_window.py @@ -2093,12 +2093,12 @@ def _on_pose_ready(self, result: PoseResult) -> None: with self._dlc_timing.measure("DLC.pose_ready_callback"): self._last_pose = result - try: - latency_ms = (time.time() - float(result.timestamp)) * 1000.0 - if logger.isEnabledFor(logging.DEBUG): - logger.debug("DLC pose latency camera_timestamp_to_gui=%.2f ms", latency_ms) - except Exception: - pass + # try: + # latency_ms = (time.time() - float(result.timestamp)) * 1000.0 + # if logger.isEnabledFor(logging.DEBUG): + # logger.debug("DLC pose latency camera_timestamp_to_gui=%.2f ms", latency_ms) + # except Exception: + # pass if self._current_frame is not None: self._display_dirty = True From 0271f1139d4f31019f2874040efef0cee479b4be Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 2 Jul 2026 13:31:01 +0200 Subject: [PATCH 14/23] Fix shutdown order in camera preview stop Reorders preview teardown so inference is stopped before stopping the multi-camera controller. This avoids stopping the controller while inference is still active and keeps shutdown state cleanup consistent. --- dlclivegui/gui/main_window.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dlclivegui/gui/main_window.py b/dlclivegui/gui/main_window.py index b4d10831a..c4e7053a4 100644 --- a/dlclivegui/gui/main_window.py +++ b/dlclivegui/gui/main_window.py @@ -1739,9 +1739,9 @@ def _stop_preview(self) -> None: # Stop any active recording first self._stop_multi_camera_recording() - self.multi_camera_controller.stop(wait=True) self._pending_recording_after_preview = False self._stop_inference(show_message=False) + self.multi_camera_controller.stop() self._fps_tracker.clear() self._last_display_time = 0.0 if hasattr(self, "camera_stats_label"): From c9a7b48fd8c21bed91a7247d48c1ec9cb492708d Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 2 Jul 2026 10:15:13 +0200 Subject: [PATCH 15/23] Add mono-preserving output mode to GenTL Adds a `preserve_mono` option to the GenTL camera backend so Mono pixel formats can remain 2D Mono8 output instead of always converting to BGR. The backend now reports preserve-mono capability, exposes recommended/actual output format based on camera format, and persists detected pixel/output format metadata in settings. It also logs first-frame format details to make runtime format behavior easier to inspect. --- dlclivegui/cameras/backends/gentl_backend.py | 54 +++++++++++++++++++- 1 file changed, 52 insertions(+), 2 deletions(-) diff --git a/dlclivegui/cameras/backends/gentl_backend.py b/dlclivegui/cameras/backends/gentl_backend.py index 89dac6511..680b9e4a5 100644 --- a/dlclivegui/cameras/backends/gentl_backend.py +++ b/dlclivegui/cameras/backends/gentl_backend.py @@ -90,6 +90,8 @@ def __init__(self, settings): ns = {} self._fast_start: bool = bool(ns.get("fast_start", False)) + self._preserve_mono: bool = bool(getattr(settings, "preserve_mono", False) or ns.get("preserve_mono", False)) + self._logged_first_frame: bool = False raw_device_id = ns.get("device_id") or props.get("device_id") legacy_serial = ns.get("serial_number") or ns.get("serial") or props.get("serial_number") or props.get("serial") @@ -177,10 +179,20 @@ def actual_pixel_format(self) -> str | None: """Camera/native pixel format selected on the GenICam PixelFormat node.""" return self._camera_pixel_format or (self._pixel_format if self._pixel_format != "auto" else None) + @property + def recommended_preserve_mono(self) -> bool | None: + if not self._camera_pixel_format: + return None + return self._is_camera_mono() + @property def actual_output_format(self) -> str | None: - """Current GenTL backend emits OpenCV-native BGR uint8 frames.""" - return self._actual_output_format or "BGR8" + """Backend output frame format emitted to the app, e.g. 'Mono8' or 'BGR8'.""" + if self._actual_output_format: + return self._actual_output_format + if not self._camera_pixel_format: + return None + return "Mono8" if self._should_output_mono() else "BGR8" @classmethod def is_available(cls) -> bool: @@ -196,6 +208,7 @@ def static_capabilities(cls) -> dict[str, SupportLevel]: "device_discovery": SupportLevel.SUPPORTED, "stable_identity": SupportLevel.SUPPORTED, "hardware_trigger": SupportLevel.BEST_EFFORT, + "preserve_mono": SupportLevel.SUPPORTED, } def _debug_trigger_nodes(self, node_map, *, context: str = "") -> None: @@ -609,6 +622,13 @@ def waits_for_hardware_trigger(self) -> bool: role = str(self._trigger_attr(getattr(self, "_trigger", None), "role", "off") or "off").lower() return role in {"external", "follower"} + def _is_camera_mono(self) -> bool: + fmt = str(self._camera_pixel_format or self._pixel_format or "").strip() + return fmt.startswith("Mono") + + def _should_output_mono(self) -> bool: + return bool(self._preserve_mono and self._is_camera_mono()) + @staticmethod def _output_format_for_frame(frame: np.ndarray) -> str: if frame.ndim == 2: @@ -664,6 +684,25 @@ def read(self) -> CapturedFrame: pass if self._actual_output_format is None: self._actual_output_format = self._output_format_for_frame(frame) + try: + ns = self._ensure_settings_ns() + ns["actual_output_format"] = self._actual_output_format + ns["preserve_mono"] = self._preserve_mono + except Exception: + pass + if not self._logged_first_frame: + self._logged_first_frame = True + LOG.info( + "[GenTL] first frame device_id=%s shape=%s dtype=%s nbytes=%.2f MB " + "camera_pixel_format=%s output_format=%s preserve_mono=%s", + self._device_id, + frame.shape, + frame.dtype, + frame.nbytes / (1024 * 1024), + self._camera_pixel_format, + self.actual_output_format, + self._preserve_mono, + ) return CapturedFrame( frame=frame, @@ -1378,6 +1417,14 @@ def _configure_pixel_format(self, node_map) -> None: pixel_format_node.value = selected self._pixel_format = str(pixel_format_node.value) self._camera_pixel_format = self._pixel_format + try: + ns = self._ensure_settings_ns() + ns["actual_pixel_format"] = self._camera_pixel_format + ns["detected_pixel_format"] = self._camera_pixel_format + ns["actual_output_format"] = self.actual_output_format + ns["preserve_mono"] = self._preserve_mono + except Exception: + pass LOG.debug("GenTL pixel format selected: %s", self._pixel_format) @@ -1883,6 +1930,9 @@ def _convert_frame(self, frame: np.ndarray) -> np.ndarray: frame = cv2.cvtColor(frame, cv2.COLOR_BayerGR2BGR) elif fmt == "BayerBG8": frame = cv2.cvtColor(frame, cv2.COLOR_BayerBG2BGR) + elif self._should_output_mono(): + # Keep Mono* cameras as 2D uint8 frames when explicitly requested. + pass else: frame = cv2.cvtColor(frame, cv2.COLOR_GRAY2BGR) From 82b0eb9b0396850691192508f0ea05f3c19e7fc4 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 2 Jul 2026 15:18:25 +0200 Subject: [PATCH 16/23] Async recording stop and queued frame dispatch Move recording shutdown off the UI thread and finalize it via a Qt signal so stop actions no longer block the interface. RecordingManager now uses a lock-protected recorder map plus a bounded background frame-dispatch queue/thread to decouple frame intake from disk writes, drop frames under sustained backpressure, and cleanly stop/tear down recorders. The recording-with-overlays UI path is also disabled in this change. --- dlclivegui/gui/main_window.py | 54 +++++++-- dlclivegui/gui/recording_manager.py | 180 +++++++++++++++++++++++----- 2 files changed, 191 insertions(+), 43 deletions(-) diff --git a/dlclivegui/gui/main_window.py b/dlclivegui/gui/main_window.py index c4e7053a4..cda2def5a 100644 --- a/dlclivegui/gui/main_window.py +++ b/dlclivegui/gui/main_window.py @@ -6,12 +6,13 @@ import json import logging import os +import threading import time from pathlib import Path import cv2 import numpy as np -from PySide6.QtCore import QRect, QSettings, Qt, QTimer, QUrl +from PySide6.QtCore import QRect, QSettings, Qt, QTimer, QUrl, Signal from PySide6.QtGui import ( QAction, QActionGroup, @@ -88,6 +89,8 @@ class DLCLiveMainWindow(QMainWindow): """Main application window.""" + _recording_stopped_async = Signal() + def __init__(self, config: ApplicationSettings | None = None): super().__init__() self.setWindowTitle("DeepLabCut Live GUI") @@ -179,6 +182,9 @@ def __init__(self, config: ApplicationSettings | None = None): self._dlc_tile_scale: tuple[float, float] = (1.0, 1.0) # (scale_x, scale_y) # Display flag (decoupled from frame capture for performance) self._display_dirty: bool = False + # Recording state + self._recording_stopping = False + self._recording_stopped_async.connect(self._on_recording_stopped_async) self._load_icons() self._preview_pixmap = QPixmap(LOGO_ALPHA) @@ -652,11 +658,11 @@ def _build_recording_group(self) -> QGroupBox: form.addRow(grid) # Recording options - self.record_with_overlays_checkbox = QCheckBox("Record video with overlays") - self.record_with_overlays_checkbox.setToolTip( - "Enable to include pose overlays in recorded video (keypoints & bounding boxes)" - ) - self.record_with_overlays_checkbox.setChecked(False) + # self.record_with_overlays_checkbox = QCheckBox("Record video with overlays") + # self.record_with_overlays_checkbox.setToolTip( + # "Enable to include pose overlays in recorded video (keypoints & bounding boxes)" + # ) + # self.record_with_overlays_checkbox.setChecked(False) self.fast_encoding_checkbox = QCheckBox("Use faster encoding parameters") self.fast_encoding_checkbox.setToolTip( @@ -669,7 +675,7 @@ def _build_recording_group(self) -> QGroupBox: recording_options = QWidget() recording_options_layout = QHBoxLayout(recording_options) recording_options_layout.setContentsMargins(0, 0, 0, 0) - recording_options_layout.addWidget(self.record_with_overlays_checkbox) + # recording_options_layout.addWidget(self.record_with_overlays_checkbox) recording_options_layout.addWidget(self.fast_encoding_checkbox) recording_options_layout.addStretch(1) @@ -1470,8 +1476,8 @@ def _on_recording_frame_ready( if not self._rec_manager.is_active: return - if self.record_with_overlays_checkbox.isChecked(): - frame = self._render_overlays_for_recording(camera_id, frame) + # if self.record_with_overlays_checkbox.isChecked(): + # frame = self._render_overlays_for_recording(camera_id, frame) self._rec_manager.write_frame(camera_id, frame, timestamp, timestamp_metadata=timestamp_metadata) @@ -1627,9 +1633,35 @@ def _stop_multi_camera_recording(self) -> None: if not self._rec_manager.is_active: return - self.multi_camera_controller.set_recording_frame_do_emit(False) + if getattr(self, "_recording_stopping", False): + return + + self._recording_stopping = True - self._rec_manager.stop_all() + self.start_record_button.setEnabled(False) + self.stop_record_button.setEnabled(False) + self.statusBar().showMessage("Stopping multi-camera recording…", 3000) + + # Stop frame emission immediately so no new frames enter recording pipeline. + try: + self.multi_camera_controller.set_recording_frame_do_emit(False) + except Exception: + logger.exception("Failed to disable recording frame emission") + + def worker(): + try: + self._rec_manager.stop_all() + finally: + self._recording_stopped_async.emit() + + threading.Thread( + target=worker, + name="StopRecordingWorker", + daemon=True, + ).start() + + def _on_recording_stopped_async(self) -> None: + self._recording_stopping = False self.start_record_button.setEnabled(True) self.stop_record_button.setEnabled(False) self.statusBar().showMessage("Multi-camera recording stopped", 3000) diff --git a/dlclivegui/gui/recording_manager.py b/dlclivegui/gui/recording_manager.py index 9c7a523aa..6f5f4f43a 100644 --- a/dlclivegui/gui/recording_manager.py +++ b/dlclivegui/gui/recording_manager.py @@ -2,6 +2,8 @@ import logging import math +import queue +import threading import time from pathlib import Path @@ -15,6 +17,8 @@ log = logging.getLogger(__name__) +_FRAME_SENTINEL = object() + class RecordingManager: """Handle multi-camera recording lifecycle and filenames.""" @@ -24,21 +28,30 @@ def __init__(self): self._session_dir: Path | None = None self._run_dir: Path | None = None + self._lock = threading.RLock() + self._frame_queue: queue.Queue | None = None + self._dispatch_thread: threading.Thread | None = None + self._dispatch_stop = threading.Event() + @property def is_active(self) -> bool: - return bool(self._recorders) + with self._lock: + return bool(self._recorders) @property def recorders(self) -> dict[str, VideoRecorder]: - return self._recorders + with self._lock: + return dict(self._recorders) @property def session_dir(self) -> Path | None: - return self._session_dir + with self._lock: + return self._session_dir @property def run_dir(self) -> Path | None: - return self._run_dir + with self._lock: + return self._run_dir @staticmethod def _backend_ns(cam: CameraSettings) -> dict: @@ -79,7 +92,71 @@ def _resolve_recording_fps( return cls._valid_fps(cam.fps) def pop(self, cam_id: str, default=None) -> VideoRecorder | None: - return self._recorders.pop(cam_id, default) + with self._lock: + 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 + + 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) + except queue.Full: + pass + + if t is not None: + t.join(timeout=timeout) + if t.is_alive(): + log.warning("Recording frame dispatcher did not stop within %.1fs", timeout) + + with self._lock: + self._dispatch_thread = None + self._frame_queue = None + self._dispatch_stop.clear() + + def _dispatch_loop(self) -> None: + with self._lock: + q = self._frame_queue + + if q is None: + return + + while not self._dispatch_stop.is_set(): + try: + item = q.get(timeout=0.1) + except queue.Empty: + continue + + try: + if item is _FRAME_SENTINEL: + break + + cam_id, frame, timestamp = item + self._write_frame_now(cam_id, frame, timestamp) + + finally: + try: + q.task_done() + except ValueError: + pass def start_all( self, @@ -107,8 +184,9 @@ def start_all( Returns: run_dir if at least one recorder started, else None. """ - if self._recorders: - return self._run_dir + with self._lock: + if self._recorders: + return self._run_dir if not active_cams: return None @@ -125,8 +203,9 @@ def start_all( log.error("Failed to create run dir: %s", exc) return None - self._session_dir = session_dir - self._run_dir = run_dir + with self._lock: + self._session_dir = session_dir + self._run_dir = run_dir started_any = False @@ -164,7 +243,8 @@ def start_all( ) try: recorder.start() - self._recorders[cam_id] = recorder + with self._lock: + self._recorders[cam_id] = recorder started_any = True log.info("Started recording %s -> %s", cam_id, cam_path) except Exception as exc: @@ -174,30 +254,40 @@ def start_all( return None if not started_any: - self._recorders.clear() - self._session_dir = None - self._run_dir = None + with self._lock: + self._recorders.clear() + self._session_dir = None + self._run_dir = None return None + self._start_dispatcher() return run_dir def stop_all(self) -> None: - for cam_id, rec in self._recorders.items(): + self._stop_dispatcher() + + with self._lock: + recorders = list(self._recorders.items()) + self._recorders.clear() + + for cam_id, rec in recorders: try: rec.stop() log.info("Stopped recording %s", cam_id) except Exception as exc: log.warning("Error stopping recorder for %s: %s", cam_id, exc) - self._recorders.clear() - self._session_dir = None - self._run_dir = None - - def write_frame( - self, cam_id: str, frame: np.ndarray, timestamp: float | None = None, timestamp_metadata: object | None = None - ) -> None: - rec = self._recorders.get(cam_id) + + with self._lock: + self._session_dir = None + self._run_dir = None + + def _write_frame_now(self, cam_id: str, frame: np.ndarray, timestamp: float | None = None, timestamp_metadata: object | None = None) -> None: + with self._lock: + rec = self._recorders.get(cam_id) + if not rec or not rec.is_running: return + try: rec.write( frame, @@ -206,18 +296,40 @@ def write_frame( ) except Exception as exc: log.warning( - "Failed to write frame for %s: %s: %s frame_shape=%s dtype=%s", + "Failed to write frame for %s: %s: %s frame_shape=%s dtype=%s. Removing recorder.", cam_id, type(exc).__name__, str(exc) or repr(exc), getattr(frame, "shape", None), getattr(frame, "dtype", None), ) - try: - rec.stop() - except Exception: - log.exception("Failed to stop recorder for %s after write error.") - self._recorders.pop(cam_id, None) + + with self._lock: + rec = self._recorders.pop(cam_id, None) + + if rec is not None: + try: + rec.stop() + except Exception: + 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) -> None: + with self._lock: + q = self._frame_queue + active = cam_id in self._recorders + + if not active or q is None: + 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), + ) def get_stats_summary(self) -> str: totals = { @@ -231,7 +343,11 @@ def get_stats_summary(self) -> str: "max_latency": 0.0, "avg_latencies": [], } - for rec in self._recorders.values(): + + with self._lock: + recorders = list(self._recorders.values()) + + for rec in recorders: stats: RecorderStats | None = rec.get_stats() if not stats: continue @@ -245,8 +361,8 @@ def get_stats_summary(self) -> str: totals["max_latency"] = max(totals["max_latency"], stats.last_latency) totals["avg_latencies"].append(stats.average_latency) - if len(self._recorders) == 1: - rec = next(iter(self._recorders.values())) + if len(recorders) == 1: + rec = recorders[0] stats = rec.get_stats() if stats: from dlclivegui.utils.stats import format_recorder_stats @@ -261,7 +377,7 @@ def get_stats_summary(self) -> str: fill_pct = (100.0 * totals["queue"] / buffer) if buffer > 0 else 0.0 return ( - f"{len(self._recorders)} cams | {totals['written']}/{totals['enqueued']} frames | " + f"{len(recorders)} cams | {totals['written']}/{totals['enqueued']} frames | " f"writer {totals['write_fps']:.1f} fps | " f"latency {totals['max_latency'] * 1000:.1f}ms (avg {avg * 1000:.1f}ms) | " f"queue {queue_text} ({fill_pct:.0f}%) | " From 80bd1212a9fd99698020eec7a440d6559616a7d9 Mon Sep 17 00:00:00 2001 From: C-Achard Date: Fri, 3 Jul 2026 16:54:38 +0200 Subject: [PATCH 17/23] Update recording_manager.py --- dlclivegui/gui/recording_manager.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/dlclivegui/gui/recording_manager.py b/dlclivegui/gui/recording_manager.py index 6f5f4f43a..dd1deadc5 100644 --- a/dlclivegui/gui/recording_manager.py +++ b/dlclivegui/gui/recording_manager.py @@ -281,7 +281,9 @@ def stop_all(self) -> None: self._session_dir = None self._run_dir = None - def _write_frame_now(self, cam_id: str, frame: np.ndarray, timestamp: float | None = None, timestamp_metadata: object | None = None) -> None: + def _write_frame_now( + self, cam_id: str, frame: np.ndarray, timestamp: float | None = None, timestamp_metadata: object | None = None + ) -> None: with self._lock: rec = self._recorders.get(cam_id) @@ -313,7 +315,9 @@ def _write_frame_now(self, cam_id: str, frame: np.ndarray, timestamp: float | No except Exception: 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) -> None: + def write_frame( + 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 From 48ef319b58cd575fccbbeaf17442955e1763e1aa Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 13 Aug 2026 14:05:25 +0200 Subject: [PATCH 18/23] Use runner pose directly in DLC processor Switch pose handling to validate `self._dlc.pose` from `runner.get_pose` instead of calling the private `_post_process_pose` path. The emitted `PosePacket.raw` now carries the validated pose array, keeping packet contents aligned with the processed keypoints. --- dlclivegui/services/dlc_processor.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/dlclivegui/services/dlc_processor.py b/dlclivegui/services/dlc_processor.py index 4d4df7584..ce45dd6a4 100644 --- a/dlclivegui/services/dlc_processor.py +++ b/dlclivegui/services/dlc_processor.py @@ -590,19 +590,19 @@ def _process_frame( with self._timing.measure("DLC.runner.get_pose"): self._dlc.pose = self._dlc.runner.get_pose(processed_frame) - with self._timing.measure("DLC.post_process_pose"): - raw_pose: Any = self._dlc._post_process_pose(processed_frame, frame_time=timestamp) + # with self._timing.measure("DLC.post_process_pose"): + # raw_pose: Any = self._dlc._post_process_pose(processed_frame, frame_time=timestamp) inference_time = time.perf_counter() - inference_start with self._timing.measure("DLC.validate_pose"): - pose_arr: np.ndarray = validate_pose_array(raw_pose, source_backend=PoseBackends.DLC_LIVE) + pose_arr: np.ndarray = validate_pose_array(self._dlc.pose, source_backend=PoseBackends.DLC_LIVE) pose_packet = PosePacket( schema_version=0, keypoints=pose_arr, keypoint_names=None, individual_ids=None, source=PoseSource(backend=PoseBackends.DLC_LIVE, model_type=self._settings.model_type), - raw=raw_pose, + raw=pose_arr, ) processor_overhead = 0.0 From 025b6f67a84f76998eb159ea76233c00c84bf47c Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 13 Aug 2026 17:07:59 +0200 Subject: [PATCH 19/23] Split timing loggers across worker thread --- dlclivegui/services/dlc_processor.py | 76 ++++++++++++++++------------ 1 file changed, 44 insertions(+), 32 deletions(-) diff --git a/dlclivegui/services/dlc_processor.py b/dlclivegui/services/dlc_processor.py index ce45dd6a4..ec92461e7 100644 --- a/dlclivegui/services/dlc_processor.py +++ b/dlclivegui/services/dlc_processor.py @@ -183,8 +183,14 @@ def __init__(self) -> None: self._gpu_inference_times: deque[float] = deque(maxlen=60) self._processor_overhead_times: deque[float] = deque(maxlen=60) - self._timing = WorkerTimingStats( - "DLCLiveProcessor", + self._enqueue_timing = WorkerTimingStats( + "DLCLiveEnqueue", + logger=logger, + log_interval=1.0, + enabled=bool(DLC_DO_LOG_TIMING or ENABLE_PROFILING), + ) + self._worker_timing = WorkerTimingStats( + "DLCLiveWorker", logger=logger, log_interval=1.0, enabled=bool(DLC_DO_LOG_TIMING or ENABLE_PROFILING), @@ -250,7 +256,7 @@ def enqueue_frame(self, frame: np.ndarray, timestamp: float) -> None: q = self._queue should_start = t is None or not t.is_alive() - with self._timing.measure("DLC.enqueue.copy_frame"): + with self._enqueue_timing.measure("DLC.enqueue.copy_frame"): frame_c = frame.copy() enq_time = time.perf_counter() @@ -276,6 +282,9 @@ def enqueue_frame(self, frame: np.ndarray, timestamp: float) -> None: timestamp, ) self._start_worker_locked(frame_c, timestamp) + + self._enqueue_timing.note_frame() + self._enqueue_timing.maybe_log() return # Worker is now running; refresh queue snapshot. q = self._queue @@ -287,15 +296,18 @@ def enqueue_frame(self, frame: np.ndarray, timestamp: float) -> None: while True: try: - with self._timing.measure("DLC.enqueue.put"): + with self._enqueue_timing.measure("DLC.enqueue.put"): q.put_nowait(item) with self._stats_lock: self._frames_enqueued += 1 + + self._enqueue_timing.note_frame() + self._enqueue_timing.maybe_log() return except queue.Full: try: - with self._timing.measure("DLC.enqueue.drop_stale"): + with self._enqueue_timing.measure("DLC.enqueue.drop_stale"): q.get_nowait() try: q.task_done() @@ -534,7 +546,7 @@ def _timed_processor(self): def timed_process(pose, _op=original, _holder=holder, **kwargs): start = time.perf_counter() try: - with self._timing.measure("DLC.processor.process"): + with self._worker_timing.measure("DLC.processor.process"): return _op(pose, **kwargs) finally: _holder[0] = time.perf_counter() - start @@ -579,22 +591,22 @@ def _process_frame( if self._dlc is None: raise RuntimeError("DLCLive instance is not initialized.") # Time GPU inference (and processor overhead when present) - with self._timing.measure("DLC.prepare_frame"): + with self._worker_timing.measure("DLC.prepare_frame"): frame = self._prepare_input_frame(frame) with self._timed_processor() as proc_holder: inference_start = time.perf_counter() - with self._timing.measure("DLC.process_frame"): + with self._worker_timing.measure("DLC.process_frame"): processed_frame = self._dlc.process_frame(frame) - with self._timing.measure("DLC.runner.get_pose"): + with self._worker_timing.measure("DLC.runner.get_pose"): self._dlc.pose = self._dlc.runner.get_pose(processed_frame) - # with self._timing.measure("DLC.post_process_pose"): + # with self._worker_timing.measure("DLC.post_process_pose"): # raw_pose: Any = self._dlc._post_process_pose(processed_frame, frame_time=timestamp) inference_time = time.perf_counter() - inference_start - with self._timing.measure("DLC.validate_pose"): + with self._worker_timing.measure("DLC.validate_pose"): pose_arr: np.ndarray = validate_pose_array(self._dlc.pose, source_backend=PoseBackends.DLC_LIVE) pose_packet = PosePacket( schema_version=0, @@ -613,7 +625,7 @@ def _process_frame( # Emit pose (measure signal overhead) signal_start = time.perf_counter() - with self._timing.measure("DLC.emit.pose_ready"): + with self._worker_timing.measure("DLC.emit.pose_ready"): self.pose_ready.emit(PoseResult(pose=pose_packet.keypoints, timestamp=timestamp, packet=pose_packet)) signal_time = time.perf_counter() - signal_start @@ -635,8 +647,8 @@ def _process_frame( self._gpu_inference_times.append(gpu_inference_time) self._processor_overhead_times.append(processor_overhead) - self._timing.note_frame() - self._timing.maybe_log() + self._worker_timing.note_frame() + self._worker_timing.maybe_log() self.frame_processed.emit() def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None: @@ -645,7 +657,7 @@ def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None: if not self._settings.model_path: raise RuntimeError("No DLCLive model path configured.") - with self._timing.measure("DLC.build_options"): + with self._worker_timing.measure("DLC.build_options"): dyn = self._settings.dynamic if not isinstance(dyn, (list, tuple)) or len(dyn) != 3: try: @@ -683,12 +695,12 @@ def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None: raise RuntimeError( "DLCLive class is not available. Ensure the dlclive package is installed and can be imported." ) - with self._timing.measure("DLC.construct"): + with self._worker_timing.measure("DLC.construct"): self._dlc = DLCLive(**options) - self._timing.maybe_log() + self._worker_timing.maybe_log() except Exception as exc: - self._timing.note_error() - self._timing.maybe_log() + self._worker_timing.note_error() + self._worker_timing.maybe_log() with self._lifecycle_lock: self._state = WorkerState.FAULTED raise RuntimeError( @@ -699,7 +711,7 @@ def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None: logger.debug("DLC worker stop requested during construction; exiting before init_inference.") return - with self._timing.measure("DLC.prepare_init_frame"): + with self._worker_timing.measure("DLC.prepare_init_frame"): init_frame = self._prepare_input_frame(init_frame) if logger.isEnabledFor(logging.DEBUG): @@ -710,12 +722,12 @@ def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None: init_frame.flags["C_CONTIGUOUS"], ) # First inference to initialize - with self._timing.measure("DLC.init_inference"): + with self._worker_timing.measure("DLC.init_inference"): self._dlc.init_inference(init_frame) self._debug_log_dlc_runner_device() - self._timing.note_frame() - self._timing.maybe_log() + self._worker_timing.note_frame() + self._worker_timing.maybe_log() if self._stop_event.is_set(): logger.debug("DLC worker stop requested after init_inference; exiting before RUNNING state.") @@ -723,7 +735,7 @@ def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None: # Pass DLCLive cfg to processor if available if hasattr(self._dlc, "processor") and hasattr(self._dlc.processor, "set_dlc_cfg"): - with self._timing.measure("DLC.processor.set_dlc_cfg"): + with self._worker_timing.measure("DLC.processor.set_dlc_cfg"): self._dlc.processor.set_dlc_cfg(getattr(self._dlc, "cfg", None)) self._initialized = True @@ -777,8 +789,8 @@ def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None: try: self._process_frame(frame, ts, enq, queue_wait_time=0.0) except Exception as exc: - self._timing.note_error() - self._timing.maybe_log() + self._worker_timing.note_error() + self._worker_timing.maybe_log() logger.exception("Pose inference failed", exc_info=exc) self.error.emit(str(exc)) finally: @@ -791,14 +803,14 @@ def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None: # Normal operation: timed get try: wait_start = time.perf_counter() - with self._timing.measure("DLC.queue_get"): + with self._worker_timing.measure("DLC.queue_wait"): item = q.get(timeout=0.05) queue_wait_time = time.perf_counter() - wait_start except queue.Empty: continue except Exception as exc: - self._timing.note_error() - self._timing.maybe_log() + self._worker_timing.note_error() + self._worker_timing.maybe_log() logger.exception("Error getting item from queue", exc_info=exc) with self._lifecycle_lock: self._state = WorkerState.FAULTED @@ -809,8 +821,8 @@ def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None: frame, ts, enq = item self._process_frame(frame, ts, enq, queue_wait_time=queue_wait_time) except Exception as exc: - self._timing.note_error() - self._timing.maybe_log() + self._worker_timing.note_error() + self._worker_timing.maybe_log() logger.exception("Pose inference failed", exc_info=exc) self.error.emit(str(exc)) finally: @@ -821,7 +833,7 @@ def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None: logger.info("DLC worker thread exiting") - self._timing.maybe_log() + self._worker_timing.maybe_log() class DLCService: From 2dd41869986ffd5055fdbfea8d75438e1956d419 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 13 Aug 2026 17:10:07 +0200 Subject: [PATCH 20/23] pre-commit --- dlclivegui/gui/main_window.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dlclivegui/gui/main_window.py b/dlclivegui/gui/main_window.py index cda2def5a..f73081498 100644 --- a/dlclivegui/gui/main_window.py +++ b/dlclivegui/gui/main_window.py @@ -51,9 +51,9 @@ from dlclivegui.config import ( ALLOWED_VIDEO_CONTAINERS, DEFAULT_CONFIG, - GUI_MAX_DISPLAY_FPS, DEFAULT_RECORDING_CONTAINER, DLC_DO_LOG_TIMING, + GUI_MAX_DISPLAY_FPS, ApplicationSettings, BoundingBoxSettings, CameraSettings, From a2c748d2591cc6fcf39c230c61b85fb4530aacca Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Fri, 14 Aug 2026 10:09:52 +0200 Subject: [PATCH 21/23] Remove duplicate node value reader --- dlclivegui/cameras/backends/gentl_backend.py | 13 ------------- 1 file changed, 13 deletions(-) diff --git a/dlclivegui/cameras/backends/gentl_backend.py b/dlclivegui/cameras/backends/gentl_backend.py index 680b9e4a5..c07c6cb67 100644 --- a/dlclivegui/cameras/backends/gentl_backend.py +++ b/dlclivegui/cameras/backends/gentl_backend.py @@ -1217,19 +1217,6 @@ def _node_symbolics(node) -> list[str]: except Exception: return [] - @staticmethod - def _node_value(node_map, name: str, default=None): - """Best-effort read of a GenICam node value.""" - try: - node = getattr(node_map, name) - except Exception: - return default - - try: - return node.value - except Exception: - return default - @classmethod def _node_float(cls, node_map, *names: str, allow_zero: bool = False) -> float | None: """Return the first positive float value from a list of GenICam node names.""" From 3f7979092d3e2640313f7d24ecb66e6839dd0a2d Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Tue, 18 Aug 2026 15:52:37 +0200 Subject: [PATCH 22/23] Pass timestamp metadata to frame writer Update the recording queue consumer to unpack `timestamp_metadata` from each frame item and forward it to `_write_frame_now`. This keeps queued frame payload handling aligned with the expanded tuple format and preserves metadata during recording. --- dlclivegui/gui/recording_manager.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/dlclivegui/gui/recording_manager.py b/dlclivegui/gui/recording_manager.py index dd1deadc5..cf9be9eb5 100644 --- a/dlclivegui/gui/recording_manager.py +++ b/dlclivegui/gui/recording_manager.py @@ -149,8 +149,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 40061a6feac1df5d3a1948c5286e0d9d2bf621ba Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Tue, 18 Aug 2026 16:01:40 +0200 Subject: [PATCH 23/23] Stabilize rec manager async write tests Add a small `_wait_until` polling helper and use it in recorder-manager tests that assert write calls or recorder removal after `write_frame`. This removes race-prone immediate assertions against asynchronous work, making timestamp and stable-camera-ID behavior tests deterministic. --- tests/gui/test_rec_manager.py | 20 +++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/tests/gui/test_rec_manager.py b/tests/gui/test_rec_manager.py index 5ba5e2dca..d11969983 100644 --- a/tests/gui/test_rec_manager.py +++ b/tests/gui/test_rec_manager.py @@ -1,5 +1,7 @@ from __future__ import annotations +import time + import numpy as np import pytest @@ -36,6 +38,16 @@ def current_frames(_active_cams_two): return frames +def _wait_until(predicate, timeout: float = 1.0) -> None: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return + time.sleep(0.01) + + raise AssertionError("Condition was not reached before timeout") + + @pytest.mark.unit def test_start_all_creates_recorders_and_returns_run_dir( recording_settings, _active_cams_two, current_frames, patch_video_recorder, patch_build_run_dir @@ -204,6 +216,7 @@ def test_write_frame_uses_given_timestamp( mgr.write_frame(cam0_id, frame, timestamp=123.0) rec = mgr.recorders[cam0_id] + _wait_until(lambda: len(rec.write_calls) > 0) assert rec.write_calls[-1][1] == 123.0 @@ -223,6 +236,7 @@ def test_write_frame_uses_time_when_timestamp_missing( mgr.write_frame(cam0_id, frame, timestamp=None) rec = mgr.recorders[cam0_id] + _wait_until(lambda: len(rec.write_calls) > 0) assert rec.write_calls[-1][1] == 999.0 @@ -238,6 +252,7 @@ def test_write_frame_removes_recorder_on_exception( rec.raise_on_write = True mgr.write_frame(cam0_id, current_frames[cam0_id], timestamp=1.0) + _wait_until(lambda: cam0_id not in mgr.recorders) assert cam0_id not in mgr.recorders @@ -346,11 +361,14 @@ def test_recording_manager_uses_stable_camera_id_not_display_id( assert rec.frame_size == (480, 640) mgr.write_frame(stable_id, frame, timestamp=123.0) + _wait_until(lambda: len(rec.write_calls) > 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) + # No async work for unknown ID + time.sleep(0.05) assert len(rec.write_calls) == 1 @@ -461,7 +479,7 @@ def test_write_frame_passes_timestamp_metadata( mgr.write_frame(cam0_id, frame, timestamp=123.0, timestamp_metadata=meta) rec = mgr.recorders[cam0_id] - assert len(rec.write_calls) == 1 + _wait_until(lambda: len(rec.write_calls) > 0) written_frame, written_timestamp, written_metadata = rec.write_calls[0] assert written_frame is frame