diff --git a/dlclivegui/gui/main_window.py b/dlclivegui/gui/main_window.py
index 2eef9fb11..5fab2e41e 100644
--- a/dlclivegui/gui/main_window.py
+++ b/dlclivegui/gui/main_window.py
@@ -446,7 +446,7 @@ def _build_dlc_group(self) -> QGroupBox:
# Processor selection
processor_path_layout = QHBoxLayout()
self.processor_folder_edit = QLineEdit()
- self.processor_folder_edit.setText(default_processors_dir())
+ self.processor_folder_edit.setText(self._settings_store.get_processor_folder(default=default_processors_dir()))
processor_path_layout.addWidget(self.processor_folder_edit)
self.browse_processor_folder_button = QPushButton("Browse...")
@@ -481,13 +481,34 @@ def _build_dlc_group(self) -> QGroupBox:
processing_sttgs = lyts.make_two_field_row(
"Inference camera",
self.dlc_camera_combo,
- "Processor",
+ "Custom processor",
self.processor_combo,
key_width=None,
)
self.dlc_camera_combo.update_shrink_width()
form.addRow(processing_sttgs)
+ self.processor_status_label = QLabel("Processor: No clients | Recording: No")
+ self.processor_status_label.setWordWrap(True)
+ # form.addRow("Processor Status", self.processor_status_label)
+ self.use_custom_proc_checkbox = QCheckBox("Use custom processor")
+ self.use_custom_proc_checkbox.setChecked(False)
+ self.use_custom_proc_checkbox.setToolTip(
+ "If enabled, the GUI will load and interact with the selected processor plugin.\n"
+ )
+ self.processor_toggle_row = lyts.make_two_field_row(
+ "Processor status",
+ self.processor_status_label,
+ None,
+ self.use_custom_proc_checkbox,
+ key_width=None,
+ left_stretch=0,
+ right_stretch=0,
+ style_values=False,
+ )
+ self.processor_toggle_row.setVisible(False) # Hide until a processor is selected
+ form.addRow(self.processor_toggle_row)
+
# Wrap inference buttons in a widget to prevent shifting
inference_button_widget = QWidget()
inference_buttons = QHBoxLayout(inference_button_widget)
@@ -508,17 +529,6 @@ def _build_dlc_group(self) -> QGroupBox:
# self.show_predictions_checkbox.setChecked(True)
# form.addRow(self.show_predictions_checkbox)
- self.allow_processor_ctrl_checkbox = QCheckBox("Allow processor-based control")
- self.allow_processor_ctrl_checkbox.setChecked(False)
- self.allow_processor_ctrl_checkbox.setToolTip(
- "If enabled, the GUI will load and interact with the selected processor plugin.\n"
- )
- form.addRow(self.allow_processor_ctrl_checkbox)
-
- self.processor_status_label = QLabel("Processor: No clients | Recording: No")
- self.processor_status_label.setWordWrap(True)
- form.addRow("Processor Status", self.processor_status_label)
-
return group
def _build_recording_group(self) -> QGroupBox:
@@ -801,8 +811,8 @@ def _connect_signals(self) -> None:
self._dlc.initialized.connect(self._on_dlc_initialised)
self.dlc_camera_combo.currentIndexChanged.connect(self._on_dlc_camera_changed)
self.dlc_camera_combo.currentTextChanged.connect(self.dlc_camera_combo.update_shrink_width)
- self.allow_processor_ctrl_checkbox.stateChanged.connect(lambda _s: self._update_dlc_controls_enabled())
- self.allow_processor_ctrl_checkbox.stateChanged.connect(lambda _s: self._update_processor_status())
+ self.processor_combo.currentIndexChanged.connect(self._on_processor_selection_changed)
+ self.use_custom_proc_checkbox.stateChanged.connect(lambda _s: self._update_processor_status())
# Recording settings
## Session name persistence + preview updates
@@ -1085,10 +1095,11 @@ def _action_browse_directory(self) -> None:
def _action_browse_processor_folder(self) -> None:
"""Browse for processor folder."""
- current_path = self.processor_folder_edit.text() or default_processors_dir()
+ current_path = self.processor_folder_edit.text().strip() or default_processors_dir()
directory = QFileDialog.getExistingDirectory(self, "Select processor folder", current_path)
if directory:
self.processor_folder_edit.setText(directory)
+ self._settings_store.set_processor_folder(directory)
self._refresh_processors()
def _action_open_recording_folder(self) -> None:
@@ -1132,9 +1143,11 @@ def _action_open_recording_folder(self) -> None:
logger.error(f"Failed to open folder: {exc}")
self.statusBar().showMessage("Could not open recording folder.", 5000)
- def _processor_control_enabled(self) -> bool:
+ def _custom_processor_enabled(self) -> bool:
return bool(
- getattr(self, "allow_processor_ctrl_checkbox", None) and self.allow_processor_ctrl_checkbox.isChecked()
+ getattr(self, "use_custom_proc_checkbox", None)
+ and self.use_custom_proc_checkbox.isChecked()
+ and self.processor_combo.currentData() is not None
)
def _refresh_processors(self) -> None:
@@ -1142,10 +1155,17 @@ def _refresh_processors(self) -> None:
self.processor_combo.addItem("No Processor", None)
selected_folder = self.processor_folder_edit.text().strip()
- if Path(selected_folder).exists():
- self._scanned_processors = scan_processor_folder(selected_folder)
+ selected_path = Path(selected_folder).expanduser() if selected_folder else None
+
+ if selected_path is not None and selected_path.is_dir():
+ resolved_folder = str(selected_path.resolve())
+ self._settings_store.set_processor_folder(resolved_folder)
+ self._scanned_processors = scan_processor_folder(resolved_folder)
+ source_text = resolved_folder
else:
self._scanned_processors = scan_processor_package("dlclivegui.processors")
+ source_text = "package dlclivegui.processors"
+
self._processor_keys = list(self._scanned_processors.keys())
for key in self._processor_keys:
@@ -1154,9 +1174,7 @@ def _refresh_processors(self) -> None:
self.processor_combo.addItem(display_name, key)
self.processor_combo.update_shrink_width()
- self.statusBar().showMessage(
- f"Found {len(self._processor_keys)} processor(s) in package dlclivegui.processors", 3000
- )
+ self.statusBar().showMessage(f"Found {len(self._processor_keys)} processor(s) in {source_text}", 3000)
# ------------------------------------------------------------------
# Recording path preview and session name persistence
@@ -1704,23 +1722,20 @@ def _configure_dlc(self) -> bool:
# Instantiate processor if selected
processor = None
- if self._processor_control_enabled():
- selected_key = self.processor_combo.currentData()
- if selected_key is not None and self._scanned_processors:
- try:
- # For now, instantiate with no parameters
- processor = instantiate_from_scan(self._scanned_processors, selected_key)
- processor_name = self._scanned_processors[selected_key]["name"]
- self.statusBar().showMessage(f"Loaded processor: {processor_name}", 3000)
- except Exception as e:
- error_msg = f"Failed to instantiate processor: {e}"
- self._show_error(error_msg)
- logger.error(error_msg)
- return False
- else:
- selected_key = self.processor_combo.currentData()
- if selected_key is not None:
- self.statusBar().showMessage(f"Processor selection ignored (control disabled): {selected_key}", 3000)
+ selected_key = self.processor_combo.currentData()
+ if self._custom_processor_enabled():
+ try:
+ # For now, instantiate with no parameters
+ processor = instantiate_from_scan(self._scanned_processors, selected_key)
+ processor_name = self._scanned_processors[selected_key]["name"]
+ self.statusBar().showMessage(f"Loaded processor: {processor_name}", 3000)
+ except Exception as e:
+ error_msg = f"Failed to instantiate processor: {e}"
+ self._show_error(error_msg)
+ logger.error(error_msg)
+ return False
+ elif selected_key is not None:
+ self.statusBar().showMessage(f"Custom processor disabled: {selected_key}", 3000)
self._dlc.configure(settings, processor=processor)
self._model_path_store.save_if_valid(settings.model_path)
@@ -1734,24 +1749,28 @@ def _update_inference_buttons(self) -> None:
def _update_dlc_controls_enabled(self) -> None:
"""Enable/disable DLC settings based on inference state."""
allow_changes = not self._dlc_active
- processor_controls = allow_changes and self._processor_control_enabled()
widgets = [
self.model_path_edit,
self.browse_model_button,
self.dlc_camera_combo,
- # self.additional_options_edit,
]
+
processor_widgets = [
self.processor_folder_edit,
self.browse_processor_folder_button,
self.refresh_processors_button,
self.processor_combo,
]
+
for widget in widgets:
widget.setEnabled(allow_changes)
+
for widget in processor_widgets:
- widget.setEnabled(processor_controls)
+ widget.setEnabled(allow_changes)
+
+ if hasattr(self, "use_custom_proc_checkbox"):
+ self.use_custom_proc_checkbox.setEnabled(allow_changes)
def _update_camera_controls_enabled(self) -> None:
multi_cam_recording = self._rec_manager.is_active
@@ -1841,7 +1860,7 @@ def _update_metrics(self) -> None:
self.dlc_stats_label.setText("DLC processor idle")
# Update processor status (connection and recording state)
- if hasattr(self, "processor_status_label") and self._processor_control_enabled():
+ if hasattr(self, "processor_status_label") and self._custom_processor_enabled():
self._update_processor_status()
# --- Recorder stats ---
@@ -1853,26 +1872,40 @@ def _update_metrics(self) -> None:
else:
self.recording_stats_label.setText(self._last_recorder_summary)
+ def _on_processor_selection_changed(
+ self,
+ _index: int,
+ ) -> None:
+ """Enable custom processing when a processor is selected."""
+ has_selection = self.processor_combo.currentData() is not None
+ self.processor_toggle_row.setVisible(has_selection)
+
+ self.use_custom_proc_checkbox.blockSignals(True)
+ self.use_custom_proc_checkbox.setChecked(has_selection)
+ self.use_custom_proc_checkbox.blockSignals(False)
+
+ self._update_processor_status()
+
def _update_processor_status(self) -> None:
"""Update processor connection and recording status, handle auto-recording."""
- if not self._processor_control_enabled():
- self.processor_status_label.setText("Processor control disabled")
+ if not self._custom_processor_enabled():
+ self.processor_status_label.setText("Disabled")
return
if not self._dlc_active or not self._dlc_initialized:
- self.processor_status_label.setText("Processor: Not active")
+ self.processor_status_label.setText("Not active")
return
# Get processor instance from _dlc
processor = self._dlc._processor
if processor is None:
- self.processor_status_label.setText("Processor: None loaded")
+ self.processor_status_label.setText("None loaded")
return
# Check if processor has the required attributes (socket-based processors)
if not hasattr(processor, "conns") or not hasattr(processor, "_recording"):
- self.processor_status_label.setText("Processor: No status info")
+ self.processor_status_label.setText("No status info")
return
# Get connection count and recording state
@@ -1885,7 +1918,7 @@ def _update_processor_status(self) -> None:
self.processor_status_label.setText(f"Clients: {client_str} | Recording: {recording_str}")
# Handle auto-recording based on processor's video recording flag
- if hasattr(processor, "_vid_recording") and self.allow_processor_ctrl_checkbox.isChecked():
+ if hasattr(processor, "_vid_recording") and self.use_custom_proc_checkbox.isChecked():
current_vid_recording = processor.video_recording
# Check if video recording state changed
@@ -2157,6 +2190,9 @@ def closeEvent(self, event: QCloseEvent) -> None: # pragma: no cover - GUI beha
# Remember model path on exit
self._model_path_store.save_if_valid(self.model_path_edit.text().strip())
+ # Remember processor folder on exit
+ if hasattr(self, "processor_folder_edit"):
+ self._settings_store.set_processor_folder(self.processor_folder_edit.text().strip())
# Close the window
super().closeEvent(event)
diff --git a/dlclivegui/gui/recording_manager.py b/dlclivegui/gui/recording_manager.py
index b41cbb845..556d963e9 100644
--- a/dlclivegui/gui/recording_manager.py
+++ b/dlclivegui/gui/recording_manager.py
@@ -215,7 +215,14 @@ def write_frame(
timestamp_metadata=timestamp_metadata,
)
except Exception as exc:
- log.warning("Failed to write frame for %s: %s", cam_id, exc)
+ log.warning(
+ "Failed to write frame for %s: %s: %s frame_shape=%s dtype=%s",
+ cam_id,
+ type(exc).__name__,
+ str(exc) or repr(exc),
+ getattr(frame, "shape", None),
+ getattr(frame, "dtype", None),
+ )
try:
rec.stop()
except Exception:
diff --git a/dlclivegui/processors/PLUGIN_SYSTEM.md b/dlclivegui/processors/PLUGIN_SYSTEM.md
index 9e975e01c..5c3b2e320 100644
--- a/dlclivegui/processors/PLUGIN_SYSTEM.md
+++ b/dlclivegui/processors/PLUGIN_SYSTEM.md
@@ -1,59 +1,63 @@
-# DeepLabCut Live GUI — Processor Plugin System
+# DeepLabCut Live GUI: Processor Plugin System
This repository includes a **plugin-style processor system** that lets the GUI discover and instantiate **DLCLive processors** dynamically.
-Processors are Python classes (typically subclasses of `dlclive.Processor`) that can optionally:
+Processors are Python classes that subclass `dlclive.processor.Processor`, directly or indirectly, and can optionally:
-- receive pose estimates during inference (via `process(pose, **kwargs)`),
-- broadcast pose-derived data to external clients (e.g., for experiment control),
-- expose metadata so the GUI can list them and (optionally) build simple parameter UIs.
+- Receive pose estimates during inference through `process(pose, **kwargs)`
+- Broadcast pose-derived data, for example for experiment control
+- Expose metadata so the GUI can list them and support processor configuration
-> **Security / control note:** The GUI should treat processors as **optional, user-controlled extensions**. In our current design, the GUI exposes an opt-in toggle (recommended label: **“Allow processor control”**) that gates whether processor plugins are instantiated and whether the GUI reads/acts on processor state.
-
----
+> The GUI should treat processors as **optional, user-controlled extensions**.
+> In our current design, the GUI exposes an opt-in toggle, **Allow processor-based control**, that controls whether processor plugins are instantiated and whether the GUI reads or acts on processor state.
## Overview
### Useful files
-- `dlclivegui/processors/dlc_processor_socket.py` — Example socket-based processor base class + examples
-- `dlclivegui/processors/processor_utils.py` — Scanning + instantiation helpers used by the GUI
-
----
+- `dlclivegui/processors/dlc_processor_socket.py`: Example socket-based processor base class
+- `dlclivegui/processors/examples.py`: Example processor implementations, such as One-Euro filtering
+- `dlclivegui/processors/processor_utils.py`: Scanning and instantiation helpers used by the GUI
## Architecture
-### 1) Processor registry (module-level)
+### 1) Processor class discovery
-A typical processor module defines a registry and a decorator. The decorator registers classes into `PROCESSOR_REGISTRY` using either `PROCESSOR_ID` (if present) or the class name.
+A processor module defines one or more classes that subclass `dlclive.processor.Processor`, directly or indirectly.
-```python
-# Registry for GUI discovery
-PROCESSOR_REGISTRY = {}
+The GUI discovers eligible processor classes by inspecting the imported module.
-def register_processor(cls):
- registry_key = getattr(cls, "PROCESSOR_ID", cls.__name__)
- PROCESSOR_REGISTRY[registry_key] = cls
- return cls
-```
+```python
+from dlclive.processor import Processor
-Register processors by decorating the class:
-```python
-@register_processor
-class ExampleProcessor(BaseProcessorSocket):
+class ExampleProcessor(Processor):
PROCESSOR_NAME = "Example Processor"
PROCESSOR_DESCRIPTION = "Example description"
PROCESSOR_PARAMS = {}
+
+ def process(self, pose, **kwargs):
+ return pose
```
+Only processor classes defined in the scanned module are included. Processor classes imported from another module are ignored to avoid duplicate entries.
+
+Reusable base classes that should not appear in the GUI can explicitly opt out:
+
+```python
+class BaseProcessorSocket(Processor):
+ PROCESSOR_DISCOVERABLE = False
+```
+
+Concrete subclasses of a non-discoverable base class remain discoverable by default.
+
### 2) Processor metadata
Each processor class should define metadata attributes to help GUI discovery:
```python
class MyProcessorSocket(BaseProcessorSocket):
- PROCESSOR_NAME = "Mouse Pose Processor" # Human-readable
+ PROCESSOR_NAME = "Use Pose Processor" # Human-readable
PROCESSOR_DESCRIPTION = "Broadcasts processed pose values"
PROCESSOR_PARAMS = {
"bind": {
@@ -76,64 +80,46 @@ class MyProcessorSocket(BaseProcessorSocket):
> **Recommendation:** For security, prefer binding to `127.0.0.1` unless you explicitly want LAN exposure.
-### 3) Module-level discovery helpers (optional)
-Processor modules can expose:
-
-- `get_available_processors()` — returns a dictionary of available processors and metadata
-
-Example:
-
-```python
-def get_available_processors():
- 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()
- }
-```
-
----
-
-## Discovery & instantiation (current utilities)
+## Discovery & instantiation
The GUI uses utilities from `dlclivegui/processors/processor_utils.py`:
-- `scan_processor_folder(folder_path)` — discover processors from `*.py` files in a folder
-- `scan_processor_package(package_name="dlclivegui.processors")` — discover processors from a package namespace
-- `instantiate_from_scan(processors_dict, processor_key, **kwargs)` — instantiate a processor from scan output
+- `discover_processor_classes(module)`: discover eligible processor classes in an imported module
+- `scan_processor_folder(folder_path)`: discover processors from `*.py` files in a folder
+- `scan_processor_package(package_name="dlclivegui.processors")`: discover processors from a package namespace
+- `instantiate_from_scan(processors_dict, processor_key, **kwargs)`: instantiate a processor from scan output
+
+Package and folder scanning use different module-loading mechanisms, but both use the same class-based processor discovery.
### Key format
Scan results are dictionaries keyed like:
-```
-"some_file.py::SomeProcessorClassOrId"
+```text
+some_file.py::SomeProcessorClass
```
-Each entry contains (at least):
+Each entry contains at least:
- `class`: the processor class object
- `name`: display name
- `description`: description text
- `params`: parameter schema
- `file`: module filename
-- `class_name`: class/registry key
+- `class_name`: processor class name
- `file_path`: full path to the module file
### Example: scanning and instantiating
```python
from dlclivegui.processors.processor_utils import (
- scan_processor_package,
- scan_processor_folder,
instantiate_from_scan,
+ scan_processor_folder,
+ scan_processor_package,
)
+
# Built-in processors
processors = scan_processor_package("dlclivegui.processors")
@@ -142,126 +128,129 @@ processors = scan_processor_package("dlclivegui.processors")
# List
for key, info in processors.items():
- print(f"{info['name']} ({key}) — {info['description']}")
+ print(f"{info['name']} ({key}): {info['description']}")
# Instantiate
-selected_key = next(iter(processors.keys()))
-proc = instantiate_from_scan(processors, selected_key, bind=("127.0.0.1", 6000))
+selected_key = next(iter(processors))
+proc = instantiate_from_scan(
+ processors,
+ selected_key,
+ bind=("127.0.0.1", 6000),
+)
```
----
+### Legacy registration compatibility
-## GUI integration & the “Allow processor control” gate
+Earlier processor modules may still import and use:
-### Recommended behavior
-
-To keep processor behavior explicit and opt-in, the GUI provides a toggle (**Allow processor-based control**) with these semantics:
-
-- **Disabled (default):**
- - the GUI does **not instantiate** any processor plugin;
- - the GUI does **not read or act** on processor state (connections, recording flags, remote commands);
- - inference runs with `processor=None`.
- - *processor code may be imported by the discovery process*
+```python
+from dlclivegui.processors import PROCESSOR_REGISTRY, register_processor
+```
-- **Enabled:**
- - the GUI may instantiate the selected processor and (optionally) reflect processor state in the UI.
- - the processor will be used by the `DLCLive` instance during inference.
+The registry and decorator remain temporarily available for compatibility with existing processor modules. However:
-This lets users decide whether they want to run processor plugins and whether those plugins may influence UI/recording behavior.
+- GUI discovery does not use `PROCESSOR_REGISTRY`
+- GUI discovery does not call `get_available_processors()`
+- Decorating a class is not required for discovery
+- An existing decorated class remains discoverable because the decorator returns the original class
-> We recommend users to follow this design patter when designing their own processors
-> to help ensure predictable behavior and clear user control over processor-based features.
-> **We are not responsible for any unexpected behavior caused by custom processors,**
-> **and the examples are provided as-is with no guarantees.**
+New processor modules should rely on subclass discovery instead of defining a registry or discovery function.
----
+## GUI integration & enabling custom processors
-## Socket-based processors (example base class)
+### Recommended behavior
-The built-in `BaseProcessorSocket` (in `dlc_processor_socket.py`) demonstrates a simple approach for:
+To keep processor behavior explicit and opt-in, the GUI provides an **Use custom processor** toggle with these effects:
-- accepting multiple clients,
-- receiving control messages (e.g., start/stop recording),
-- broadcasting payloads to connected clients,
-- cleaning up reliably on shutdown.
+- **Disabled by default:**
+ - The GUI does **not instantiate** any processor plugin
+ - The GUI does **not read or act** on processor state, such as connections, recording flags, or remote commands
+ - Inference runs with `processor=None`
+ - Processor code may still be imported by the discovery process
-### Key points
+- **Enabled:**
+ - The GUI may instantiate the selected processor and reflect processor state in the UI
+ - The processor is used by the `DLCLive` instance during inference
-- Socket server is optional: `BaseProcessorSocket` supports `start_server(...)`.
-- Connections are tracked in `self.conns`.
-- `broadcast(payload)` sends to all clients; failing clients are dropped.
-- `stop()` closes clients and listener, joins threads, and attempts to wake `accept()` during shutdown.
+This lets users decide whether they want to run processor plugins and whether those plugins may influence UI or recording behavior.
-> **Tip:** If you publish processors for others to use, keep module import side-effect free (define classes/functions only).
+> We recommend that users follow this design pattern when creating processors to help ensure predictable behavior and clear user control over processor-based features.
+> **We are not responsible for unexpected behavior caused by custom processors, and the examples are provided as-is with no guarantees.**
----
+## Socket-based processors
-## Adding a new processor
+The built-in `BaseProcessorSocket` in `dlc_processor_socket.py` demonstrates a simple approach for:
-1) Create a new module file in a processor folder (or inside `dlclivegui/processors/`).
+- Accepting multiple clients
+- Receiving control messages, such as start and stop recording,
+- Broadcasting payloads to connected clients,
+- Cleaning up reliably on shutdown.
-2) Define a processor class and metadata:
+`BaseProcessorSocket` is a reusable base class and is not shown as a selectable processor in the GUI:
```python
-from dlclive import Processor
+PROCESSOR_DISCOVERABLE = False
+```
-PROCESSOR_REGISTRY = {}
+Concrete subclasses defined in processor modules are discovered normally.
-def register_processor(cls):
- PROCESSOR_REGISTRY[getattr(cls, "PROCESSOR_ID", cls.__name__)] = cls
- return cls
+### Key points
-@register_processor
-class MyNewProcessor(Processor):
- PROCESSOR_NAME = "My New Processor"
- PROCESSOR_DESCRIPTION = "Does something cool"
- PROCESSOR_PARAMS = {
- "my_param": {"type": "bool", "default": True, "description": "Enable cool feature"}
- }
+- The socket server is optional: `BaseProcessorSocket` supports `start_server(...)`.
+- Connections are tracked in `self.conns`.
+- `broadcast(payload)` sends to all clients, and failing clients are dropped.
+- `stop()` closes clients and the listener, joins threads, and attempts to wake `accept()` during shutdown.
- def process(self, pose, **kwargs):
- # Do something with pose
- return pose
+> **Tip:** If you publish processors for others to use, keep module imports side-effect free where possible. Define classes and functions during import, and initialize sockets, hardware, or other resources when the processor is instantiated.
+## Adding a new processor
-def get_available_processors():
- return {
- name: {
- "class": cls,
- "name": getattr(cls, "PROCESSOR_NAME", name),
- "description": getattr(cls, "PROCESSOR_DESCRIPTION", ""),
- "params": getattr(cls, "PROCESSOR_PARAMS", {}),
+1. Create a new module file in a processor folder or inside `dlclivegui/processors/`.
+
+2. Define a processor class and metadata:
+ ```python
+ from dlclive.processor import Processor
+
+ class MyNewProcessor(Processor):
+ PROCESSOR_NAME = "My New Processor"
+ PROCESSOR_DESCRIPTION = "Does something useful"
+ PROCESSOR_PARAMS = {
+ "my_param": {
+ "type": "bool",
+ "default": True,
+ "description": "Enable optional behavior",
+ }
}
- for name, cls in PROCESSOR_REGISTRY.items()
- }
-```
-3) Refresh processors in the GUI, select your processor, and start inference (with processor control enabled if required).
+ def __init__(self, my_param: bool = True):
+ super().__init__()
+ self.my_param = my_param
+
+ def process(self, pose, **kwargs):
+ # Do something with pose
+ return pose
+ ```
+ No registration decorator, module-level registry, or `get_available_processors()` function is required.
----
+3. Refresh processors in the GUI, select your processor, and start inference with processor control enabled if required.
## Parameter schema types
Supported `PROCESSOR_PARAMS` types:
-- `"bool"` — checkbox
-- `"int"` — integer input
-- `"float"` — float input
-- `"str"` — string input
-- `"bytes"` — string that gets encoded to bytes
-- `"tuple"` — tuple (e.g., `(host, port)`)
-- `"dict"` — dictionary
-- `"list"` — list
+- `"bool"`: checkbox
+- `"int"`: integer input
+- `"float"`: float input
+- `"str"`: string input
+- `"bytes"`: string that gets encoded to bytes
+- `"tuple"`: tuple, for example `(host, port)`
+- `"dict"`: dictionary
+- `"list"`: list
----
+The processor constructor remains the base definition of accepted arguments and values.
## Notes on external processors
-External processors are arbitrary Python code. Only load processors you trust.
-
-
-
-## License
+External processors are arbitrary Python code and are imported during discovery. Only load processors you trust.
-This project is distributed under its project license.
-See `LICENSE` in the repository.
+Where possible, processor modules should avoid import-time side effects and initialize files, sockets, hardware, or other resources only when the processor is instantiated.
diff --git a/dlclivegui/processors/__init__.py b/dlclivegui/processors/__init__.py
new file mode 100644
index 000000000..8e7717155
--- /dev/null
+++ b/dlclivegui/processors/__init__.py
@@ -0,0 +1,3 @@
+from .registry import PROCESSOR_REGISTRY, register_processor
+
+__all__ = ["register_processor", "PROCESSOR_REGISTRY"]
diff --git a/dlclivegui/processors/dlc_processor_socket.py b/dlclivegui/processors/dlc_processor_socket.py
index 8ded01069..6a91ef1a3 100644
--- a/dlclivegui/processors/dlc_processor_socket.py
+++ b/dlclivegui/processors/dlc_processor_socket.py
@@ -7,14 +7,13 @@
import sys
import time
from collections import deque
-from math import acos, atan2, copysign, degrees, pi, sqrt
from multiprocessing.connection import Client, Listener
from pathlib import Path
from threading import Event, Thread
import numpy as np
import pandas as pd
-from dlclive import Processor # type: ignore
+from dlclive.processor import Processor # type: ignore
logger = logging.getLogger("dlc_processor_socket")
@@ -24,59 +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:
- 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
-
-
-class OneEuroFilter: # pragma: no cover
- def __init__(self, t0, x0, dx0=None, min_cutoff=1.0, beta=0.0, d_cutoff=1.0):
- self.min_cutoff = min_cutoff
- self.beta = beta
- self.d_cutoff = d_cutoff
- self.x_prev = x0
- if dx0 is None:
- dx0 = np.zeros_like(x0)
- self.dx_prev = dx0
- self.t_prev = t0
-
- @staticmethod
- def smoothing_factor(t_e, cutoff):
- r = 2 * pi * cutoff * t_e
- return r / (r + 1)
-
- @staticmethod
- def exponential_smoothing(alpha, x, x_prev):
- return alpha * x + (1 - alpha) * x_prev
-
- def __call__(self, t, x):
- t_e = t - self.t_prev
- if t_e <= 0:
- return x
- a_d = self.smoothing_factor(t_e, self.d_cutoff)
- dx = (x - self.x_prev) / t_e
- dx_hat = self.exponential_smoothing(a_d, dx, self.dx_prev)
-
- cutoff = self.min_cutoff + self.beta * abs(dx_hat)
- a = self.smoothing_factor(t_e, cutoff)
- x_hat = self.exponential_smoothing(a, x, self.x_prev)
-
- self.x_prev = x_hat
- self.dx_prev = dx_hat
- self.t_prev = t
-
- return x_hat
-
# pragma: cover
class BaseProcessorSocket(Processor):
@@ -90,6 +36,7 @@ class BaseProcessorSocket(Processor):
PROCESSOR_NAME = "Base Socket Processor"
PROCESSOR_DESCRIPTION = "Base class for socket-based processors with multi-client support"
PROCESSOR_PARAMS = {}
+ PROCESSOR_DISCOVERABLE = False # base class, not intended to be an example processor
def __init__(
self,
@@ -474,375 +421,3 @@ def get_data(self):
if self.dlc_cfg is not None:
save_dict["dlc_cfg"] = self.dlc_cfg
return save_dict
-
-
-@register_processor
-class ExampleProcessorSocketCalculateMousePose(BaseProcessorSocket): # pragma: no cover
- """
- DLC Processor with pose calculations (center, heading, head angle) and optional filtering.
-
- Calculates:
- - center: Weighted average of head keypoints
- - heading: Body orientation (degrees)
- - head_angle: Head rotation relative to body (radians)
-
- Broadcasts: [timestamp, center_x, center_y, heading, head_angle]
- """
-
- PROCESSOR_NAME = "Example Experiment Pose Processor"
- PROCESSOR_DESCRIPTION = "Calculates mouse center, heading, and head angle with optional One-Euro filtering"
- PROCESSOR_PARAMS = {
- "bind": {
- "type": "tuple",
- "default": ("127.0.0.1", 6000),
- "description": "Server address (host, port)",
- },
- "authkey": {
- "type": "bytes",
- "default": b"secret password",
- "description": "Authentication key for clients",
- },
- "use_perf_counter": {
- "type": "bool",
- "default": False,
- "description": "Use time.perf_counter() instead of time.time()",
- },
- "use_filter": {
- "type": "bool",
- "default": False,
- "description": "Apply One-Euro filter to calculated values",
- },
- "filter_kwargs": {
- "type": "dict",
- "default": {"min_cutoff": 1.0, "beta": 0.02, "d_cutoff": 1.0},
- "description": "One-Euro filter parameters (min_cutoff, beta, d_cutoff)",
- },
- "save_original": {
- "type": "bool",
- "default": False,
- "description": "Save raw pose arrays for analysis",
- },
- }
-
- def __init__(
- self,
- bind=("127.0.0.1", 6000),
- authkey=b"secret password",
- use_perf_counter=False,
- use_filter=False,
- filter_kwargs: dict | None = None,
- save_original=False,
- ):
- super().__init__(
- bind=bind,
- authkey=authkey,
- use_perf_counter=use_perf_counter,
- save_original=save_original,
- )
-
- self.center_x = deque()
- self.center_y = deque()
- self.heading_direction = deque()
- self.head_angle = deque()
-
- self.use_filter = use_filter
- self.filter_kwargs = filter_kwargs if filter_kwargs is not None else {}
- self.filters = None
-
- def _clear_data_queues(self):
- super()._clear_data_queues()
- self.center_x.clear()
- self.center_y.clear()
- self.heading_direction.clear()
- self.head_angle.clear()
-
- def _initialize_filters(self, vals):
- t0 = self.timing_func()
- self.filters = {
- "center_x": OneEuroFilter(t0, vals[0], **self.filter_kwargs),
- "center_y": OneEuroFilter(t0, vals[1], **self.filter_kwargs),
- "heading": OneEuroFilter(t0, vals[2], **self.filter_kwargs),
- "head_angle": OneEuroFilter(t0, vals[3], **self.filter_kwargs),
- }
- logger.debug(f"Initialized One-Euro filters with parameters: {self.filter_kwargs}")
-
- def process(self, pose, **kwargs):
- # Extract keypoints and confidence
- xy = pose[:, :2]
- conf = pose[:, 2]
-
- # Calculate weighted center from head keypoints
- head_xy = xy[[0, 1, 2, 3, 4, 5, 6, 26], :]
- head_conf = conf[[0, 1, 2, 3, 4, 5, 6, 26]]
- center = np.average(head_xy, axis=0, weights=head_conf)
-
- # Calculate body axis (tail_base -> neck)
- body_axis = xy[7] - xy[13]
- body_axis /= sqrt(np.sum(body_axis**2))
-
- # Calculate head axis (neck -> nose)
- head_axis = xy[0] - xy[7]
- head_axis /= sqrt(np.sum(head_axis**2))
-
- # 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
- sign = copysign(1, cross)
- try:
- head_angle = acos(body_axis @ head_axis) * sign
- except ValueError:
- head_angle = 0
-
- # Calculate heading (body orientation)
- heading = degrees(atan2(body_axis[1], body_axis[0]))
-
- # Raw values (heading unwrapped for filtering)
- vals = [center[0], center[1], heading, head_angle]
-
- # Apply filtering if enabled
- curr_time = self.timing_func()
- if self.use_filter:
- if self.filters is None:
- self._initialize_filters(vals)
-
- vals = [
- self.filters["center_x"](curr_time, vals[0]),
- self.filters["center_y"](curr_time, vals[1]),
- self.filters["heading"](curr_time, vals[2]),
- self.filters["head_angle"](curr_time, vals[3]),
- ]
-
- # Wrap heading to [0, 360) after filtering
- vals[2] = vals[2] % 360
- # Update step counter
- self.curr_step = self.curr_step + 1
-
- # Store processed data (only if recording)
- if self.recording:
- if self.save_original and self.original_pose is not None:
- self.original_pose.append(pose.copy())
- self.center_x.append(vals[0])
- self.center_y.append(vals[1])
- self.heading_direction.append(vals[2])
- self.head_angle.append(vals[3])
- self.time_stamp.append(curr_time)
- self.step.append(self.curr_step)
- self.frame_time.append(kwargs.get("frame_time", -1))
- if "pose_time" in kwargs:
- self.pose_time.append(kwargs["pose_time"])
-
- payload = [curr_time, vals[0], vals[1], vals[2], vals[3]]
- self.broadcast(payload)
- return pose
-
- def get_data(self):
- save_dict = super().get_data()
- save_dict["x_pos"] = np.array(self.center_x)
- save_dict["y_pos"] = np.array(self.center_y)
- save_dict["heading_direction"] = np.array(self.heading_direction)
- save_dict["head_angle"] = np.array(self.head_angle)
- save_dict["use_filter"] = self.use_filter
- save_dict["filter_kwargs"] = self.filter_kwargs
- return save_dict
-
-
-@register_processor
-class ExampleProcessorSocketFilterKeypoints(BaseProcessorSocket): # pragma: no cover
- PROCESSOR_NAME = "Mouse Pose with less keypoints"
- PROCESSOR_DESCRIPTION = "Calculates mouse center, heading, and head angle with optional One-Euro filtering"
- PROCESSOR_PARAMS = {
- "bind": {
- "type": "tuple",
- "default": ("127.0.0.1", 6000),
- "description": "Server address (host, port)",
- },
- "authkey": {
- "type": "bytes",
- "default": b"secret password",
- "description": "Authentication key for clients",
- },
- "use_perf_counter": {
- "type": "bool",
- "default": False,
- "description": "Use time.perf_counter() instead of time.time()",
- },
- "use_filter": {
- "type": "bool",
- "default": False,
- "description": "Apply One-Euro filter to calculated values",
- },
- "filter_kwargs": {
- "type": "dict",
- "default": {"min_cutoff": 1.0, "beta": 0.02, "d_cutoff": 1.0},
- "description": "One-Euro filter parameters (min_cutoff, beta, d_cutoff)",
- },
- "save_original": {
- "type": "bool",
- "default": True,
- "description": "Save raw pose arrays for analysis",
- },
- }
-
- def __init__(
- self,
- bind=("127.0.0.1", 6000),
- authkey=b"secret password",
- use_perf_counter=False,
- use_filter=False,
- filter_kwargs: dict | None = None,
- save_original=True,
- p_cutoff=0.4,
- ):
- super().__init__(
- bind=bind,
- authkey=authkey,
- use_perf_counter=use_perf_counter,
- save_original=save_original,
- )
-
- self.center_x = deque()
- self.center_y = deque()
- self.heading_direction = deque()
- self.head_angle = deque()
-
- self.p_cutoff = p_cutoff
-
- self.use_filter = use_filter
- self.filter_kwargs = filter_kwargs if filter_kwargs is not None else {}
- self.filters = None
-
- def _clear_data_queues(self):
- super()._clear_data_queues()
- self.center_x.clear()
- self.center_y.clear()
- self.heading_direction.clear()
- self.head_angle.clear()
-
- def _initialize_filters(self, vals):
- t0 = self.timing_func()
- self.filters = {
- "center_x": OneEuroFilter(t0, vals[0], **self.filter_kwargs),
- "center_y": OneEuroFilter(t0, vals[1], **self.filter_kwargs),
- "heading": OneEuroFilter(t0, vals[2], **self.filter_kwargs),
- "head_angle": OneEuroFilter(t0, vals[3], **self.filter_kwargs),
- }
- logger.debug(f"Initialized One-Euro filters with parameters: {self.filter_kwargs}")
-
- def process(self, pose, **kwargs):
- # Extract keypoints and confidence
- xy = pose[:, :2]
- conf = pose[:, 2]
-
- # Calculate weighted center from head keypoints
- head_xy = xy[[0, 1, 2, 3, 5, 6, 7], :]
- head_conf = conf[[0, 1, 2, 3, 5, 6, 7]]
- # set low confidence keypoints to zero weight
- head_conf = np.where(head_conf < self.p_cutoff, 0, head_conf)
- try:
- center = np.average(head_xy, axis=0, weights=head_conf)
- except ZeroDivisionError:
- # If all keypoints have zero weight, return without processing
- return pose
-
- neck = np.average(xy[[2, 3, 6, 7], :], axis=0, weights=conf[[2, 3, 6, 7]])
-
- # Calculate body axis (tail_base -> neck)
- body_axis = neck - xy[9]
- body_axis /= sqrt(np.sum(body_axis**2))
-
- # Calculate head axis (neck -> nose)
- head_axis = xy[0] - neck
- head_axis /= sqrt(np.sum(head_axis**2))
-
- # 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
- sign = copysign(1, cross)
- try:
- head_angle = acos(body_axis @ head_axis) * sign
- except ValueError:
- head_angle = 0
-
- # Calculate heading (body orientation)
- heading = degrees(atan2(body_axis[1], body_axis[0]))
- vals = [center[0], center[1], heading, head_angle]
-
- curr_time = self.timing_func()
- if self.use_filter:
- if self.filters is None:
- self._initialize_filters(vals)
-
- vals = [
- self.filters["center_x"](curr_time, vals[0]),
- self.filters["center_y"](curr_time, vals[1]),
- self.filters["heading"](curr_time, vals[2]),
- self.filters["head_angle"](curr_time, vals[3]),
- ]
-
- # Wrap heading to [0, 360) after filtering
- vals[2] = vals[2] % 360
- # Update step counter
- self.curr_step = self.curr_step + 1
-
- # Store processed data (only if recording)
- if self.recording:
- if self.save_original and self.original_pose is not None:
- self.original_pose.append(pose.copy())
- self.center_x.append(vals[0])
- self.center_y.append(vals[1])
- self.heading_direction.append(vals[2])
- self.head_angle.append(vals[3])
- self.time_stamp.append(curr_time)
- self.step.append(self.curr_step)
- self.frame_time.append(kwargs.get("frame_time", -1))
- if "pose_time" in kwargs:
- self.pose_time.append(kwargs["pose_time"])
-
- payload = [curr_time, vals[0], vals[1], vals[2], vals[3]]
- self.broadcast(payload)
- return pose
-
- def get_data(self):
- save_dict = super().get_data()
- save_dict["x_pos"] = np.array(self.center_x)
- save_dict["y_pos"] = np.array(self.center_y)
- save_dict["heading_direction"] = np.array(self.heading_direction)
- save_dict["head_angle"] = np.array(self.head_angle)
- save_dict["use_filter"] = self.use_filter
- save_dict["filter_kwargs"] = self.filter_kwargs
- 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
new file mode 100644
index 000000000..7ed769198
--- /dev/null
+++ b/dlclivegui/processors/examples.py
@@ -0,0 +1,391 @@
+from __future__ import annotations
+
+import logging
+from collections import deque
+from math import acos, atan2, copysign, degrees, pi, sqrt
+
+import numpy as np
+
+from dlclivegui.processors import register_processor
+from dlclivegui.processors.dlc_processor_socket import BaseProcessorSocket
+
+logger = logging.getLogger(__name__)
+
+
+class OneEuroFilter: # pragma: no cover
+ def __init__(self, t0, x0, dx0=None, min_cutoff=1.0, beta=0.0, d_cutoff=1.0):
+ self.min_cutoff = min_cutoff
+ self.beta = beta
+ self.d_cutoff = d_cutoff
+ self.x_prev = x0
+ if dx0 is None:
+ dx0 = np.zeros_like(x0)
+ self.dx_prev = dx0
+ self.t_prev = t0
+
+ @staticmethod
+ def smoothing_factor(t_e, cutoff):
+ r = 2 * pi * cutoff * t_e
+ return r / (r + 1)
+
+ @staticmethod
+ def exponential_smoothing(alpha, x, x_prev):
+ return alpha * x + (1 - alpha) * x_prev
+
+ def __call__(self, t, x):
+ t_e = t - self.t_prev
+ if t_e <= 0:
+ return x
+ a_d = self.smoothing_factor(t_e, self.d_cutoff)
+ dx = (x - self.x_prev) / t_e
+ dx_hat = self.exponential_smoothing(a_d, dx, self.dx_prev)
+
+ cutoff = self.min_cutoff + self.beta * abs(dx_hat)
+ a = self.smoothing_factor(t_e, cutoff)
+ x_hat = self.exponential_smoothing(a, x, self.x_prev)
+
+ self.x_prev = x_hat
+ self.dx_prev = dx_hat
+ self.t_prev = t
+
+ return x_hat
+
+
+@register_processor
+class ExampleProcessorSocketCalculateMousePose(BaseProcessorSocket): # pragma: no cover
+ """
+ DLC Processor with pose calculations (center, heading, head angle) and optional filtering.
+
+ Calculates:
+ - center: Weighted average of head keypoints
+ - heading: Body orientation (degrees)
+ - head_angle: Head rotation relative to body (radians)
+
+ Broadcasts: [timestamp, center_x, center_y, heading, head_angle]
+ """
+
+ PROCESSOR_NAME = "Example Experiment Pose Processor"
+ PROCESSOR_DESCRIPTION = "Calculates mouse center, heading, and head angle with optional One-Euro filtering"
+ PROCESSOR_PARAMS = {
+ "bind": {
+ "type": "tuple",
+ "default": ("127.0.0.1", 6000),
+ "description": "Server address (host, port)",
+ },
+ "authkey": {
+ "type": "bytes",
+ "default": b"secret password",
+ "description": "Authentication key for clients",
+ },
+ "use_perf_counter": {
+ "type": "bool",
+ "default": False,
+ "description": "Use time.perf_counter() instead of time.time()",
+ },
+ "use_filter": {
+ "type": "bool",
+ "default": False,
+ "description": "Apply One-Euro filter to calculated values",
+ },
+ "filter_kwargs": {
+ "type": "dict",
+ "default": {"min_cutoff": 1.0, "beta": 0.02, "d_cutoff": 1.0},
+ "description": "One-Euro filter parameters (min_cutoff, beta, d_cutoff)",
+ },
+ "save_original": {
+ "type": "bool",
+ "default": False,
+ "description": "Save raw pose arrays for analysis",
+ },
+ }
+
+ def __init__(
+ self,
+ bind=("127.0.0.1", 6000),
+ authkey=b"secret password",
+ use_perf_counter=False,
+ use_filter=False,
+ filter_kwargs: dict | None = None,
+ save_original=False,
+ ):
+ super().__init__(
+ bind=bind,
+ authkey=authkey,
+ use_perf_counter=use_perf_counter,
+ save_original=save_original,
+ )
+
+ self.center_x = deque()
+ self.center_y = deque()
+ self.heading_direction = deque()
+ self.head_angle = deque()
+
+ self.use_filter = use_filter
+ self.filter_kwargs = filter_kwargs if filter_kwargs is not None else {}
+ self.filters = None
+
+ def _clear_data_queues(self):
+ super()._clear_data_queues()
+ self.center_x.clear()
+ self.center_y.clear()
+ self.heading_direction.clear()
+ self.head_angle.clear()
+
+ def _initialize_filters(self, vals):
+ t0 = self.timing_func()
+ self.filters = {
+ "center_x": OneEuroFilter(t0, vals[0], **self.filter_kwargs),
+ "center_y": OneEuroFilter(t0, vals[1], **self.filter_kwargs),
+ "heading": OneEuroFilter(t0, vals[2], **self.filter_kwargs),
+ "head_angle": OneEuroFilter(t0, vals[3], **self.filter_kwargs),
+ }
+ logger.debug(f"Initialized One-Euro filters with parameters: {self.filter_kwargs}")
+
+ def process(self, pose, **kwargs):
+ # Extract keypoints and confidence
+ xy = pose[:, :2]
+ conf = pose[:, 2]
+
+ # Calculate weighted center from head keypoints
+ head_xy = xy[[0, 1, 2, 3, 4, 5, 6, 26], :]
+ head_conf = conf[[0, 1, 2, 3, 4, 5, 6, 26]]
+ try:
+ center = np.average(head_xy, axis=0, weights=head_conf)
+ except ZeroDivisionError:
+ center = np.zeros(2)
+
+ # Calculate body axis (tail_base -> neck)
+ body_axis = xy[7] - xy[13]
+ body_axis /= sqrt(np.sum(body_axis**2))
+
+ # Calculate head axis (neck -> nose)
+ head_axis = xy[0] - xy[7]
+ head_axis /= sqrt(np.sum(head_axis**2))
+
+ # 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:
+ head_angle = 0
+
+ # Calculate heading (body orientation)
+ heading = degrees(atan2(body_axis[1], body_axis[0]))
+
+ # Raw values (heading unwrapped for filtering)
+ vals = [center[0], center[1], heading, head_angle]
+
+ # Apply filtering if enabled
+ curr_time = self.timing_func()
+ if self.use_filter:
+ if self.filters is None:
+ self._initialize_filters(vals)
+
+ vals = [
+ self.filters["center_x"](curr_time, vals[0]),
+ self.filters["center_y"](curr_time, vals[1]),
+ self.filters["heading"](curr_time, vals[2]),
+ self.filters["head_angle"](curr_time, vals[3]),
+ ]
+
+ # Wrap heading to [0, 360) after filtering
+ vals[2] = vals[2] % 360
+ # Update step counter
+ self.curr_step = self.curr_step + 1
+
+ # Store processed data (only if recording)
+ if self.recording:
+ if self.save_original and self.original_pose is not None:
+ self.original_pose.append(pose.copy())
+ self.center_x.append(vals[0])
+ self.center_y.append(vals[1])
+ self.heading_direction.append(vals[2])
+ self.head_angle.append(vals[3])
+ self.time_stamp.append(curr_time)
+ self.step.append(self.curr_step)
+ self.frame_time.append(kwargs.get("frame_time", -1))
+ if "pose_time" in kwargs:
+ self.pose_time.append(kwargs["pose_time"])
+
+ payload = [curr_time, vals[0], vals[1], vals[2], vals[3]]
+ self.broadcast(payload)
+ return pose
+
+ def get_data(self):
+ save_dict = super().get_data()
+ save_dict["x_pos"] = np.array(self.center_x)
+ save_dict["y_pos"] = np.array(self.center_y)
+ save_dict["heading_direction"] = np.array(self.heading_direction)
+ save_dict["head_angle"] = np.array(self.head_angle)
+ save_dict["use_filter"] = self.use_filter
+ save_dict["filter_kwargs"] = self.filter_kwargs
+ return save_dict
+
+
+@register_processor
+class ExampleProcessorSocketFilterKeypoints(BaseProcessorSocket): # pragma: no cover
+ PROCESSOR_NAME = "Mouse Pose with less keypoints"
+ PROCESSOR_DESCRIPTION = "Calculates mouse center, heading, and head angle with optional One-Euro filtering"
+ PROCESSOR_PARAMS = {
+ "bind": {
+ "type": "tuple",
+ "default": ("127.0.0.1", 6000),
+ "description": "Server address (host, port)",
+ },
+ "authkey": {
+ "type": "bytes",
+ "default": b"secret password",
+ "description": "Authentication key for clients",
+ },
+ "use_perf_counter": {
+ "type": "bool",
+ "default": False,
+ "description": "Use time.perf_counter() instead of time.time()",
+ },
+ "use_filter": {
+ "type": "bool",
+ "default": False,
+ "description": "Apply One-Euro filter to calculated values",
+ },
+ "filter_kwargs": {
+ "type": "dict",
+ "default": {"min_cutoff": 1.0, "beta": 0.02, "d_cutoff": 1.0},
+ "description": "One-Euro filter parameters (min_cutoff, beta, d_cutoff)",
+ },
+ "save_original": {
+ "type": "bool",
+ "default": True,
+ "description": "Save raw pose arrays for analysis",
+ },
+ }
+
+ def __init__(
+ self,
+ bind=("127.0.0.1", 6000),
+ authkey=b"secret password",
+ use_perf_counter=False,
+ use_filter=False,
+ filter_kwargs: dict | None = None,
+ save_original=True,
+ p_cutoff=0.4,
+ ):
+ super().__init__(
+ bind=bind,
+ authkey=authkey,
+ use_perf_counter=use_perf_counter,
+ save_original=save_original,
+ )
+
+ self.center_x = deque()
+ self.center_y = deque()
+ self.heading_direction = deque()
+ self.head_angle = deque()
+
+ self.p_cutoff = p_cutoff
+
+ self.use_filter = use_filter
+ self.filter_kwargs = filter_kwargs if filter_kwargs is not None else {}
+ self.filters = None
+
+ def _clear_data_queues(self):
+ super()._clear_data_queues()
+ self.center_x.clear()
+ self.center_y.clear()
+ self.heading_direction.clear()
+ self.head_angle.clear()
+
+ def _initialize_filters(self, vals):
+ t0 = self.timing_func()
+ self.filters = {
+ "center_x": OneEuroFilter(t0, vals[0], **self.filter_kwargs),
+ "center_y": OneEuroFilter(t0, vals[1], **self.filter_kwargs),
+ "heading": OneEuroFilter(t0, vals[2], **self.filter_kwargs),
+ "head_angle": OneEuroFilter(t0, vals[3], **self.filter_kwargs),
+ }
+ logger.debug(f"Initialized One-Euro filters with parameters: {self.filter_kwargs}")
+
+ def process(self, pose, **kwargs):
+ # Extract keypoints and confidence
+ xy = pose[:, :2]
+ conf = pose[:, 2]
+
+ # Calculate weighted center from head keypoints
+ head_xy = xy[[0, 1, 2, 3, 5, 6, 7], :]
+ head_conf = conf[[0, 1, 2, 3, 5, 6, 7]]
+ # set low confidence keypoints to zero weight
+ head_conf = np.where(head_conf < self.p_cutoff, 0, head_conf)
+ try:
+ center = np.average(head_xy, axis=0, weights=head_conf)
+ except ZeroDivisionError:
+ # If all keypoints have zero weight, return without processing
+ return pose
+
+ neck = np.average(xy[[2, 3, 6, 7], :], axis=0, weights=conf[[2, 3, 6, 7]])
+
+ # Calculate body axis (tail_base -> neck)
+ body_axis = neck - xy[9]
+ body_axis /= sqrt(np.sum(body_axis**2))
+
+ # Calculate head axis (neck -> nose)
+ head_axis = xy[0] - neck
+ head_axis /= sqrt(np.sum(head_axis**2))
+
+ # 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:
+ head_angle = 0
+
+ # Calculate heading (body orientation)
+ heading = degrees(atan2(body_axis[1], body_axis[0]))
+ vals = [center[0], center[1], heading, head_angle]
+
+ curr_time = self.timing_func()
+ if self.use_filter:
+ if self.filters is None:
+ self._initialize_filters(vals)
+
+ vals = [
+ self.filters["center_x"](curr_time, vals[0]),
+ self.filters["center_y"](curr_time, vals[1]),
+ self.filters["heading"](curr_time, vals[2]),
+ self.filters["head_angle"](curr_time, vals[3]),
+ ]
+
+ # Wrap heading to [0, 360) after filtering
+ vals[2] = vals[2] % 360
+ # Update step counter
+ self.curr_step = self.curr_step + 1
+
+ # Store processed data (only if recording)
+ if self.recording:
+ if self.save_original and self.original_pose is not None:
+ self.original_pose.append(pose.copy())
+ self.center_x.append(vals[0])
+ self.center_y.append(vals[1])
+ self.heading_direction.append(vals[2])
+ self.head_angle.append(vals[3])
+ self.time_stamp.append(curr_time)
+ self.step.append(self.curr_step)
+ self.frame_time.append(kwargs.get("frame_time", -1))
+ if "pose_time" in kwargs:
+ self.pose_time.append(kwargs["pose_time"])
+
+ payload = [curr_time, vals[0], vals[1], vals[2], vals[3]]
+ self.broadcast(payload)
+ return pose
+
+ def get_data(self):
+ save_dict = super().get_data()
+ save_dict["x_pos"] = np.array(self.center_x)
+ save_dict["y_pos"] = np.array(self.center_y)
+ save_dict["heading_direction"] = np.array(self.heading_direction)
+ save_dict["head_angle"] = np.array(self.head_angle)
+ save_dict["use_filter"] = self.use_filter
+ save_dict["filter_kwargs"] = self.filter_kwargs
+ return save_dict
diff --git a/dlclivegui/processors/processor_utils.py b/dlclivegui/processors/processor_utils.py
index b32445c38..e47dbe2f8 100644
--- a/dlclivegui/processors/processor_utils.py
+++ b/dlclivegui/processors/processor_utils.py
@@ -17,7 +17,104 @@ def default_processors_dir() -> str:
return str(path)
-def scan_processor_folder(folder_path):
+def _processor_base_class():
+ from dlclive.processor import Processor
+
+ return Processor
+
+
+def _is_processor_subclass(
+ obj,
+ *,
+ include_base: bool = False,
+) -> bool:
+ """Return whether obj is a selectable Processor subclass."""
+ if not inspect.isclass(obj):
+ return False
+
+ try:
+ processor_base = _processor_base_class()
+ except Exception:
+ logger.exception("Could not import dlclive.Processor")
+ return False
+
+ try:
+ if obj is processor_base:
+ return bool(include_base)
+
+ if not issubclass(obj, processor_base):
+ return False
+
+ # Check only the class itself, not inherited values. This lets concrete
+ # subclasses of a non-discoverable base remain discoverable by default.
+ # getattr would return the inherited value.
+ if obj.__dict__.get("PROCESSOR_DISCOVERABLE", True) is False:
+ return False
+
+ return True
+ except Exception:
+ logger.exception(
+ "Error checking whether %r is a Processor subclass",
+ obj,
+ )
+ return False
+
+
+def _add_processor_results(
+ target: dict[str, dict],
+ processors: dict[str, dict],
+ *,
+ file_name: str,
+ file_path: str,
+) -> None:
+ """Normalize discovered processors and add them to a scan result."""
+ for class_name, processor_info in processors.items():
+ key = f"{file_name}::{class_name}"
+ info = dict(processor_info)
+ info.update(
+ {
+ "file": file_name,
+ "class_name": class_name,
+ "file_path": file_path,
+ }
+ )
+ target[key] = info
+
+
+def _processor_info_from_class(cls, fallback_name: str) -> dict:
+ return {
+ "class": cls,
+ "name": getattr(cls, "PROCESSOR_NAME", fallback_name),
+ "description": getattr(cls, "PROCESSOR_DESCRIPTION", ""),
+ "params": getattr(cls, "PROCESSOR_PARAMS", {}),
+ }
+
+
+def discover_processor_classes(module, *, only_defined_in_module: bool = True) -> dict[str, dict]:
+ """Discover dlclive.Processor subclasses in a module.
+
+ Includes indirect subclasses of Processor.
+
+ Args:
+ module: Imported Python module.
+ only_defined_in_module: If True, ignore Processor subclasses imported
+ from other modules to avoid duplicate registry entries.
+ """
+ processors: dict[str, dict] = {}
+
+ for name, obj in inspect.getmembers(module, inspect.isclass):
+ if only_defined_in_module and getattr(obj, "__module__", None) != module.__name__:
+ continue
+
+ if not _is_processor_subclass(obj):
+ continue
+
+ processors[name] = _processor_info_from_class(obj, name)
+
+ return processors
+
+
+def scan_processor_folder(folder_path: str | Path) -> dict[str, dict]:
all_processors = {}
folder = Path(folder_path)
@@ -27,23 +124,21 @@ def scan_processor_folder(folder_path):
try:
processors = load_processors_from_file(py_file)
- for class_or_id, processor_info in processors.items():
- key = f"{py_file.name}::{class_or_id}"
- processor_info["file"] = py_file.name
- processor_info["class_name"] = class_or_id
- processor_info["file_path"] = str(py_file)
- all_processors[key] = processor_info
+ _add_processor_results(
+ all_processors,
+ processors,
+ file_name=py_file.name,
+ file_path=str(py_file),
+ )
except Exception:
logger.exception(f"Error loading {py_file}")
return all_processors
-def scan_processor_package(package_name: str = "dlclivegui.processors") -> dict[str | dict]:
+def scan_processor_package(package_name: str = "dlclivegui.processors") -> dict[str, dict]:
"""
Discover and load processor classes from a package namespace.
- Returns a dict keyed as 'module.py::ClassName' with the same
- structure you use today.
"""
all_processors: dict[str, dict] = {}
@@ -59,38 +154,13 @@ def scan_processor_package(package_name: str = "dlclivegui.processors") -> dict[
continue
try:
mod = import_module(mod_name)
-
- # Prefer module-level registry function if present
- if hasattr(mod, "get_available_processors"):
- processors = mod.get_available_processors()
- else:
- # Fallback: scan for dlclive.Processor subclasses
- from dlclive import Processor
-
- processors = {}
- for attr_name in dir(mod):
- obj = getattr(mod, attr_name)
- try:
- if isinstance(obj, type) and obj is not Processor and issubclass(obj, Processor):
- processors[attr_name] = {
- "class": obj,
- "name": getattr(obj, "PROCESSOR_NAME", attr_name),
- "description": getattr(obj, "PROCESSOR_DESCRIPTION", ""),
- "params": getattr(obj, "PROCESSOR_PARAMS", {}),
- }
- except Exception:
- # Non-class or weird metaclass; ignore
- pass
-
- # Normalize into your “file::class” shape
- module_file = mod.__name__.split(".")[-1] + ".py"
- for class_name, info in processors.items():
- key = f"{module_file}::{class_name}"
- info = dict(info) # copy
- info["file"] = module_file
- info["class_name"] = class_name
- info["file_path"] = mod.__file__ or ""
- all_processors[key] = info
+ processors = discover_processor_classes(mod)
+ _add_processor_results(
+ all_processors,
+ processors,
+ file_name=mod_name.split(".")[-1] + ".py",
+ file_path=getattr(mod, "__file__", ""),
+ )
except Exception:
logger.exception(f"Error importing processor module '{mod_name}'")
@@ -98,7 +168,7 @@ def scan_processor_package(package_name: str = "dlclivegui.processors") -> dict[
return all_processors
-def load_processors_from_file(file_path: str | Path):
+def load_processors_from_file(file_path: str | Path) -> dict[str, dict]:
"""
Load all processor classes from a Python file.
@@ -123,34 +193,8 @@ def load_processors_from_file(file_path: str | Path):
sys.modules[module_name] = module # Make visible during import for intra-module imports
spec.loader.exec_module(module)
- # Preferred path: the module exposes get_available_processors()
- if hasattr(module, "get_available_processors"):
- processors = module.get_available_processors()
- if not isinstance(processors, dict):
- raise TypeError(f"{file_path}: get_available_processors() must return a dict, got {type(processors)}")
- return processors
-
# Fallback path: discover subclasses of dlclive.Processor
- from dlclive import Processor
-
- processors: dict[str, dict] = {}
- for name, obj in inspect.getmembers(module, inspect.isclass):
- if obj is Processor:
- continue
- # Guard: module might define other classes; only include Processor subclasses
- try:
- if issubclass(obj, Processor):
- processors[name] = {
- "class": obj,
- "name": getattr(obj, "PROCESSOR_NAME", name),
- "description": getattr(obj, "PROCESSOR_DESCRIPTION", ""),
- "params": getattr(obj, "PROCESSOR_PARAMS", {}),
- }
- except Exception:
- # Some "classes" can fail issubclass checks; ignore safely
- continue
-
- return processors
+ return discover_processor_classes(module)
except Exception:
# Full traceback helps a ton when a plugin fails to import
@@ -158,7 +202,7 @@ def load_processors_from_file(file_path: str | Path):
return {}
-def instantiate_from_scan(processors_dict, processor_key, **kwargs):
+def instantiate_from_scan(processors_dict: dict[str, dict], processor_key: str, **kwargs):
"""
Instantiate a processor from scan_processor_folder results.
diff --git a/dlclivegui/processors/registry.py b/dlclivegui/processors/registry.py
new file mode 100644
index 000000000..38a11e4d7
--- /dev/null
+++ b/dlclivegui/processors/registry.py
@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+import logging
+import warnings
+
+logger = logging.getLogger(__name__)
+
+# Legacy compatibility registry.
+# GUI discovery no longer depends on this registry.
+PROCESSOR_REGISTRY: dict[str, type] = {}
+
+
+def register_processor(cls):
+ """Register a processor for backward compatibility.
+
+ New processor modules do not need this decorator. Processor discovery now
+ finds eligible dlclive.Processor subclasses directly.
+ """
+ warnings.warn(
+ "@register_processor is deprecated and no longer required for GUI "
+ "discovery. Define a discoverable Processor subclass instead.",
+ DeprecationWarning,
+ stacklevel=2,
+ )
+
+ registry_key = str(getattr(cls, "PROCESSOR_ID", cls.__name__))
+
+ existing = PROCESSOR_REGISTRY.get(registry_key)
+ if existing is not None and existing is not cls:
+ logger.warning(
+ "Duplicate legacy processor registration key %r: %s vs %s",
+ registry_key,
+ existing.__name__,
+ cls.__name__,
+ )
+
+ PROCESSOR_REGISTRY[registry_key] = cls
+ return cls
+
+
+def get_available_processors() -> dict[str, dict]:
+ """Return processors registered through the legacy decorator.
+
+ Deprecated:
+ GUI discovery now inspects Processor subclasses directly.
+ """
+ warnings.warn(
+ "get_available_processors() is deprecated. Use "
+ "discover_processor_classes(), scan_processor_package(), or "
+ "scan_processor_folder() instead.",
+ DeprecationWarning,
+ stacklevel=2,
+ )
+
+ 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: str,
+ **kwargs,
+):
+ """Instantiate a processor from the legacy registry.
+
+ Deprecated:
+ Use instantiate_from_scan() with scanner output instead.
+ """
+ warnings.warn(
+ "instantiate_processor() is deprecated. Use instantiate_from_scan() instead.",
+ DeprecationWarning,
+ stacklevel=2,
+ )
+
+ if class_name not in PROCESSOR_REGISTRY:
+ available = ", ".join(sorted(PROCESSOR_REGISTRY))
+ raise ValueError(f"Unknown processor {class_name!r}. Available legacy registrations: {available}")
+
+ return PROCESSOR_REGISTRY[class_name](**kwargs)
diff --git a/dlclivegui/services/video_recorder.py b/dlclivegui/services/video_recorder.py
index 4eee5ff8c..4d43a2c70 100644
--- a/dlclivegui/services/video_recorder.py
+++ b/dlclivegui/services/video_recorder.py
@@ -312,15 +312,16 @@ def write(
expected_h, expected_w = self._frame_size
actual_h, actual_w = frame.shape[:2]
if (actual_h, actual_w) != (expected_h, expected_w):
- logger.warning(
- f"Frame size mismatch: expected (h={expected_h}, w={expected_w}), "
- f"got (h={actual_h}, w={actual_w}). "
- "Stopping recorder to prevent encoding errors."
+ message = (
+ f"Frame size mismatch for recorder {self._output.name}: "
+ f"expected_hw=({expected_h}, {expected_w}) "
+ f"actual_hw=({actual_h}, {actual_w}) "
+ f"{self._describe_frame(frame)}. "
+ "Stopping recorder to prevent FFmpeg pipe errors."
)
- with self._stats_lock:
- self._encode_error = ValueError(
- f"Frame size changed from (h={expected_h}, w={expected_w}) to (h={actual_h}, w={actual_w})"
- )
+
+ logger.warning(message)
+ self._set_encode_error(message)
self._process_timing.note_error()
self._process_timing.maybe_log()
return False
@@ -460,9 +461,12 @@ def _writer_loop(self) -> None:
break
continue
except Exception as exc:
- with self._stats_lock:
- self._encode_error = exc
- logger.exception("Could not retrieve item from queue", exc_info=exc)
+ message = (
+ f"Could not retrieve frame from recorder queue for {self._output.name}: "
+ f"{type(exc).__name__}: {exc!s}"
+ )
+ self._set_encode_error(message, exc)
+ logger.exception(message)
self._stop_event.set()
break
@@ -507,9 +511,28 @@ def _writer_loop(self) -> None:
self._frame_timestamps.append(record)
except Exception as exc:
+ queue_size = q.qsize() if q is not None else -1
+
with self._stats_lock:
- self._encode_error = exc
- logger.exception("Video encoding failed while writing frame", exc_info=exc)
+ frames_enqueued = self._frames_enqueued
+ frames_written = self._frames_written
+ dropped_frames = self._dropped_frames
+
+ message = (
+ f"Video encoding failed for recorder {self._output.name}: "
+ f"{type(exc).__name__}: {exc!s}. "
+ f"{self._describe_frame(frame)} "
+ f"expected_frame_size={self._frame_size} "
+ f"frames_written={frames_written} "
+ f"frames_enqueued={frames_enqueued} "
+ f"dropped={dropped_frames} "
+ f"queue_size={queue_size}. "
+ "The FFmpeg/WriteGear pipe is no longer usable; stopping this recorder."
+ )
+
+ self._set_encode_error(message, exc)
+
+ logger.exception(message)
self._stop_event.set()
self._writer_timing.note_error()
self._writer_timing.maybe_log()
@@ -581,10 +604,34 @@ def _compute_write_fps_locked(self) -> float:
return 0.0
return (len(self._written_times) - 1) / duration
+ def _describe_frame(self, frame: np.ndarray | None) -> str:
+ if frame is None:
+ return "frame=None"
+
+ try:
+ return (
+ f"shape={frame.shape} "
+ f"dtype={frame.dtype} "
+ f"contiguous={frame.flags.c_contiguous} "
+ f"nbytes={frame.nbytes / (1024 * 1024):.2f}MB"
+ )
+ except Exception:
+ return f"frame="
+
def _current_error(self) -> Exception | None:
with self._stats_lock:
return self._encode_error
+ def _set_encode_error(self, message: str, exc: Exception | None = None) -> Exception:
+ error = RuntimeError(message)
+ if exc is not None:
+ error.__cause__ = exc
+
+ with self._stats_lock:
+ self._encode_error = error
+
+ return error
+
def _save_timestamps(self) -> None:
"""Save frame timestamps to a JSON file alongside the video."""
if not self._frame_timestamps:
diff --git a/dlclivegui/temp/engine.py b/dlclivegui/temp/engine.py
index 22138ede9..e75701783 100644
--- a/dlclivegui/temp/engine.py
+++ b/dlclivegui/temp/engine.py
@@ -6,7 +6,7 @@
# or if we update dlclive.Engine to have these methods and use that instead of a separate enum here.
# The latter would be more cohesive but also creates a dependency from utils to dlclive,
# pending release of dlclive
-class Engine(Enum):
+class Engine(str, Enum):
TENSORFLOW = "tensorflow"
PYTORCH = "pytorch"
diff --git a/dlclivegui/utils/settings_store.py b/dlclivegui/utils/settings_store.py
index a0c5677f4..0107afb1c 100644
--- a/dlclivegui/utils/settings_store.py
+++ b/dlclivegui/utils/settings_store.py
@@ -57,6 +57,42 @@ def get_fast_encoding(self, default: bool = False) -> bool:
return value
return str(value).strip().lower() in {"1", "true", "yes", "on"}
+ def get_processor_folder(self, default: str = "") -> str:
+ """
+ Return the persisted processor folder if it still exists and is a directory.
+ Otherwise return default.
+ """
+ value = self._s.value("dlc/processor_folder", default)
+ value = str(value).strip() if value is not None else ""
+
+ if not value:
+ return default
+
+ try:
+ path = Path(value).expanduser()
+ if path.is_dir():
+ return str(path.resolve())
+ except Exception:
+ logger.debug("Persisted processor folder is invalid: %s", value, exc_info=True)
+
+ return default
+
+ def set_processor_folder(self, folder: str) -> None:
+ """
+ Persist processor folder only if it exists and is a directory.
+ Invalid folders are ignored.
+ """
+ folder = str(folder).strip() if folder is not None else ""
+ if not folder:
+ return
+
+ try:
+ path = Path(folder).expanduser()
+ if path.is_dir():
+ self._s.setValue("dlc/processor_folder", str(path.resolve()))
+ except Exception:
+ logger.debug("Failed to persist processor folder: %s", folder, exc_info=True)
+
def set_fast_encoding(self, enabled: bool) -> None:
self._s.setValue("recording/fast_encoding", bool(enabled))
diff --git a/tests/custom_processors/test_base_processor.py b/tests/custom_processors/test_base_processor.py
index d38749b34..8711eec11 100644
--- a/tests/custom_processors/test_base_processor.py
+++ b/tests/custom_processors/test_base_processor.py
@@ -3,8 +3,6 @@
import importlib
import pickle
-import sys
-import types
from pathlib import Path
import numpy as np
@@ -12,29 +10,20 @@
import pytest
-def _mock_dlclive(monkeypatch):
- """Provide a dummy dlclive.Processor so the module can import in tests."""
- fake = types.ModuleType("dlclive")
-
- class Processor:
- def __init__(self, *args, **kwargs):
- pass
+@pytest.fixture
+def socket_mod():
+ """Import the socket processor using the installed DLCLive package."""
+ pytest.importorskip("dlclive.processor")
- fake.Processor = Processor
- monkeypatch.setitem(sys.modules, "dlclive", fake)
+ return importlib.import_module("dlclivegui.processors.dlc_processor_socket")
@pytest.fixture
-def socket_mod(monkeypatch):
- """
- Import the processor module with dlclive mocked.
- Adjust module name if your file lives elsewhere.
- """
- _mock_dlclive(monkeypatch)
- mod_name = "dlclivegui.processors.dlc_processor_socket"
- if mod_name in sys.modules:
- del sys.modules[mod_name]
- return importlib.import_module(mod_name)
+def example_processor_mod():
+ """Import the built-in example processors normally."""
+ pytest.importorskip("dlclive.processor")
+
+ return importlib.import_module("dlclivegui.processors.examples")
def _module_data_dir(socket_mod) -> Path:
@@ -233,12 +222,14 @@ def test_save_ignores_pre_recording_original_pose_frames(socket_mod):
("ExampleProcessorSocketFilterKeypoints", 10),
],
)
-def test_subclass_save_ignores_pre_recording_original_pose_frames(socket_mod, class_name, n_keypoints):
+def test_subclass_save_ignores_pre_recording_original_pose_frames(
+ socket_mod, example_processor_mod, class_name, n_keypoints
+):
"""
Concrete processors must keep original_pose aligned with recorded metadata
even when process() is called before recording starts.
"""
- processor_class = getattr(socket_mod, class_name)
+ processor_class = getattr(example_processor_mod, class_name)
proc = processor_class(bind=("127.0.0.1", 0), save_original=True)
try:
diff --git a/tests/custom_processors/test_builtin_discovery_utils.py b/tests/custom_processors/test_builtin_discovery_utils.py
index d91caae0a..f5041f7d9 100644
--- a/tests/custom_processors/test_builtin_discovery_utils.py
+++ b/tests/custom_processors/test_builtin_discovery_utils.py
@@ -2,13 +2,14 @@
from __future__ import annotations
import importlib
-import uuid
from pathlib import Path
import pytest
from dlclivegui.processors.processor_utils import (
+ _is_processor_subclass,
default_processors_dir,
+ discover_processor_classes,
display_processor_info,
instantiate_from_scan,
load_processors_from_file,
@@ -21,40 +22,41 @@
# ---------------------------------------------------------------------------
-def _write_temp_processor_file(tmp_path: Path, stem: str | None = None) -> Path:
- """
- Create a temporary processor module that exposes get_available_processors()
- so we don't depend on dlclive.Processor being importable.
-
- The dummy processor has safe __init__ and no side-effects.
- """
- stem = stem or f"tmp_proc_{uuid.uuid4().hex}"
+def _write_temp_processor_file(
+ tmp_path: Path,
+ *,
+ stem: str = "dummy_proc",
+) -> Path:
py_file = tmp_path / f"{stem}.py"
-
py_file.write_text(
- # Use get_available_processors to bypass dlclive import in loader.
"""
-class DummyProc:
+from dlclive.processor import Processor
+
+
+class DummyProc(Processor):
PROCESSOR_NAME = "Dummy Processor"
- PROCESSOR_DESCRIPTION = "A safe, dummy processor for tests"
+ PROCESSOR_DESCRIPTION = "Test processor"
PROCESSOR_PARAMS = {
- "foo": {"type": "int", "default": 1, "description": "dummy param"}
+ "foo": {
+ "type": "int",
+ "default": 0,
+ "description": "Test integer parameter",
+ },
+ "bar": {
+ "type": "str",
+ "default": "",
+ "description": "Test string parameter",
+ },
}
def __init__(self, **kwargs):
- self.kwargs = kwargs
-
-def get_available_processors():
- # Return the normalized mapping the loader expects
- return {
- "DummyProc": {
- "class": DummyProc,
- "name": DummyProc.PROCESSOR_NAME,
- "description": DummyProc.PROCESSOR_DESCRIPTION,
- "params": DummyProc.PROCESSOR_PARAMS,
- }
- }
-"""
+ super().__init__()
+ self.kwargs = dict(kwargs)
+
+ def process(self, pose, **kwargs):
+ return pose
+""",
+ encoding="utf-8",
)
return py_file
@@ -87,6 +89,35 @@ def test_default_processors_dir_exists():
# ---------------------------------------------------------------------------
+def test_builtin_examples_module_has_discoverable_processors():
+ from dlclivegui.processors import examples
+
+ processors = discover_processor_classes(examples)
+
+ assert processors, "No discoverable Processor subclasses found in dlclivegui.processors.examples"
+
+
+def test_builtin_example_processor_is_selectable():
+ from dlclive.processor import Processor
+
+ from dlclivegui.processors.examples import (
+ ExampleProcessorSocketCalculateMousePose,
+ )
+
+ cls = ExampleProcessorSocketCalculateMousePose
+
+ assert issubclass(cls, Processor)
+ assert cls.__module__ == "dlclivegui.processors.examples"
+ assert (
+ cls.__dict__.get(
+ "PROCESSOR_DISCOVERABLE",
+ True,
+ )
+ is not False
+ )
+ assert _is_processor_subclass(cls)
+
+
@pytest.mark.skipif(
importlib.util.find_spec("dlclivegui.processors") is None,
reason="dlclivegui.processors package not importable in this test environment",
@@ -109,16 +140,14 @@ def test_scan_processor_package_populates_and_has_valid_shape():
# ---------------------------------------------------------------------------
-def test_load_processors_from_file_prefers_registry(tmp_path: Path):
+def test_load_processors_from_file_discovers_subclass(tmp_path: Path):
py_file = _write_temp_processor_file(tmp_path)
result = load_processors_from_file(py_file)
assert isinstance(result, dict)
assert "DummyProc" in result
info = result["DummyProc"]
- # For load_processors_from_file (registry path), the minimal fields are present:
assert "class" in info and info["class"].__name__ == "DummyProc"
assert info["name"] == "Dummy Processor"
- assert "params" in info and "foo" in info["params"]
def test_scan_processor_folder_discovers_files_and_normalizes_shape(tmp_path: Path):
@@ -165,3 +194,23 @@ def test_display_processor_info_prints(capsys, tmp_path: Path):
assert "Dummy Processor" in captured
assert "Parameters:" in captured
assert "- foo (int)" in captured or "foo" in captured # depends on your formatter
+
+
+def test_legacy_register_processor_remains_import_compatible():
+ from dlclive.processor import Processor
+
+ from dlclivegui.processors import (
+ PROCESSOR_REGISTRY,
+ register_processor,
+ )
+
+ PROCESSOR_REGISTRY.pop("LegacyProc", None)
+
+ with pytest.warns(DeprecationWarning):
+
+ @register_processor
+ class LegacyProc(Processor):
+ def process(self, pose, **kwargs):
+ return pose
+
+ assert PROCESSOR_REGISTRY["LegacyProc"] is LegacyProc
diff --git a/tests/gui/test_main.py b/tests/gui/test_main.py
index df320bce3..ca177149f 100644
--- a/tests/gui/test_main.py
+++ b/tests/gui/test_main.py
@@ -175,3 +175,25 @@ def test_dlc_settings_from_ui_validates_detected_model_type(
assert settings.model_type == "pytorch"
assert isinstance(settings.model_type, str)
+
+
+def test_processor_controls_reenabled_after_inference_stops(
+ window,
+):
+ window._dlc_active = True
+ window._update_dlc_controls_enabled()
+
+ assert not window.processor_folder_edit.isEnabled()
+ assert not window.browse_processor_folder_button.isEnabled()
+ assert not window.refresh_processors_button.isEnabled()
+ assert not window.processor_combo.isEnabled()
+ assert not window.use_custom_proc_checkbox.isEnabled()
+
+ window._dlc_active = False
+ window._update_dlc_controls_enabled()
+
+ assert window.processor_folder_edit.isEnabled()
+ assert window.browse_processor_folder_button.isEnabled()
+ assert window.refresh_processors_button.isEnabled()
+ assert window.processor_combo.isEnabled()
+ assert window.use_custom_proc_checkbox.isEnabled()
diff --git a/tests/gui/test_recording_paths_ui.py b/tests/gui/test_recording_paths_ui.py
index 234c3133a..c6561d698 100644
--- a/tests/gui/test_recording_paths_ui.py
+++ b/tests/gui/test_recording_paths_ui.py
@@ -135,26 +135,35 @@ def test_start_recording_passes_session_and_timestamp(window, start_all_spy, qtb
assert recording.filename == window.filename_edit.text()
-def test_processor_overrides_session_name_and_persists(window, start_all_spy, monkeypatch, fake_processor):
- # Arrange window state so processor status logic runs
+def test_processor_overrides_session_name_and_persists(
+ window,
+ start_all_spy,
+ monkeypatch,
+ fake_processor,
+):
window._dlc_active = True
window._dlc_initialized = True
- window.allow_processor_ctrl_checkbox.setChecked(True)
+
+ window.processor_combo.addItem(
+ "Fake Processor",
+ "fake_processor",
+ )
+ window.processor_combo.setCurrentIndex(window.processor_combo.count() - 1)
+ window.use_custom_proc_checkbox.setChecked(True)
# Patch start_recording to avoid preview start/timers
- monkeypatch.setattr(window, "_start_recording", lambda: window._start_multi_camera_recording())
+ monkeypatch.setattr(
+ window,
+ "_start_recording",
+ lambda: window._start_multi_camera_recording(),
+ )
- # Install fake processor
window._dlc._processor = fake_processor
- window._last_processor_vid_recording = False # ensure it sees a "change"
+ window._last_processor_vid_recording = False
- # Act
window._update_processor_status()
- # Assert UI updated
assert window.session_name_edit.text() == "auto_ABC"
assert window.filename_edit.text() == "auto_ABC"
-
- # Assert recording call used overridden session name
kwargs = start_all_spy["kwargs"]
assert kwargs["session_name"] == "auto_ABC"