diff options
Diffstat (limited to 'plugins/motion_capture')
| -rw-r--r-- | plugins/motion_capture/camera_panel.py | 132 | ||||
| -rw-r--r-- | plugins/motion_capture/device.py | 124 | ||||
| -rw-r--r-- | plugins/motion_capture/plugin.py | 189 |
3 files changed, 249 insertions, 196 deletions
diff --git a/plugins/motion_capture/camera_panel.py b/plugins/motion_capture/camera_panel.py new file mode 100644 index 0000000..611c942 --- /dev/null +++ b/plugins/motion_capture/camera_panel.py @@ -0,0 +1,132 @@ +""" +motion_capture/camera_panel.py + +CameraPanel for AddDeviceDialog β scans for OpenCV cameras and builds +a CameraDevice when the user clicks "Add Device". +""" + +from __future__ import annotations + +from PyQt6.QtCore import Qt, QThread, pyqtSignal +from PyQt6.QtWidgets import ( + QCheckBox, QGroupBox, QHBoxLayout, QLabel, QListWidget, + QListWidgetItem, QPushButton, QVBoxLayout, QWidget, +) + + +class CameraScanThread(QThread): + cameras_found = pyqtSignal(list) # list of (index: int, label: str) + + def run(self): + results = [] + try: + import cv2 + for i in range(6): + cap = cv2.VideoCapture(i) + if cap.isOpened(): + ret, _ = cap.read() + if ret: + w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) + h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) + results.append((i, f"Camera {i} ({w}Γ{h})")) + cap.release() + except ImportError: + pass + self.cameras_found.emit(results) + + +class CameraPanel(QWidget): + """Config panel for Camera device in AddDeviceDialog.""" + + def __init__(self): + super().__init__() + self._cameras: list = [] + self._selected_index: int = 0 + self._scanner = None + + lay = QVBoxLayout(self) + lay.setContentsMargins(0, 4, 0, 4) + lay.setSpacing(8) + + self.sim_chk = QCheckBox("Simulation mode (Lissajous pattern, no camera needed)") + self.sim_chk.setChecked(False) + lay.addWidget(self.sim_chk) + + scan_grp = QGroupBox("Available Cameras") + sg = QVBoxLayout(scan_grp) + sg.setSpacing(4) + + top = QHBoxLayout() + self._scan_btn = QPushButton("π Scan Cameras") + self._scan_btn.setObjectName("addTraceBtn") + self._scan_btn.clicked.connect(self._scan) + self._status = QLabel("Click Scan to detect cameras") + self._status.setObjectName("traceSource") + top.addWidget(self._scan_btn) + top.addWidget(self._status, 1) + sg.addLayout(top) + + self._list = QListWidget() + self._list.setMaximumHeight(120) + self._list.setObjectName("portList") + self._list.itemClicked.connect(self._on_select) + self._list.setToolTip("Click a camera to select it") + sg.addWidget(self._list) + + hint = QLabel("β Click a camera above to select it") + hint.setObjectName("traceSource") + sg.addWidget(hint) + + lay.addWidget(scan_grp) + + note = QLabel( + "Requires OpenCV: pip install opencv-python\n" + "Use Simulation mode if no camera is available." + ) + note.setObjectName("traceSource") + note.setWordWrap(True) + lay.addWidget(note) + lay.addStretch() + + # ββ Scan βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ + + def _scan(self): + self._scan_btn.setEnabled(False) + self._status.setText("Scanningβ¦") + self._list.clear() + self._scanner = CameraScanThread() + self._scanner.cameras_found.connect(self._on_found) + self._scanner.start() + + def _on_found(self, cameras): + self._scan_btn.setEnabled(True) + self._cameras = cameras + self._list.clear() + if not cameras: + self._status.setText("No cameras found") + item = QListWidgetItem(" No cameras detected") + item.setFlags(item.flags() & ~Qt.ItemFlag.ItemIsSelectable) + self._list.addItem(item) + else: + self._status.setText(f"{len(cameras)} camera(s) found") + for idx, label in cameras: + item = QListWidgetItem(f" {label}") + item.setData(Qt.ItemDataRole.UserRole, idx) + self._list.addItem(item) + self._list.setCurrentRow(0) + self._selected_index = cameras[0][0] + + def _on_select(self, item: QListWidgetItem): + idx = item.data(Qt.ItemDataRole.UserRole) + if idx is not None: + self._selected_index = idx + + # ββ Build βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ + + def build_device(self, device_id: str): + from device import CameraDevice + return CameraDevice( + device_id=device_id, + camera_index=self._selected_index, + simulate=self.sim_chk.isChecked(), + ) diff --git a/plugins/motion_capture/device.py b/plugins/motion_capture/device.py index f0d082c..9806030 100644 --- a/plugins/motion_capture/device.py +++ b/plugins/motion_capture/device.py @@ -1,34 +1,39 @@ """ motion_capture/device.py -BaseDevice that reads x/y position from CameraTracker. +BaseDevice implementation for a camera-based motion tracker. + +Each CameraDevice owns a CameraTracker that runs its own background +capture thread. connect() opens the camera; disconnect() releases it. Channels: x_pos β horizontal position [px] (left=0, right=frame_width) y_pos β vertical position [px] (top=0, bottom=frame_height) -Apply the pixel_to_mm filter in Signals β Channels to convert to mm. +Apply the pixel_to_mm filter in Signals β Channels to convert to mm. """ from __future__ import annotations -import threading -from typing import Any, Dict +from typing import Any, Dict, Optional from devices.base_device import ( BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus, ) -class MotionCaptureDevice(BaseDevice): +class CameraDevice(BaseDevice): - def __init__(self, tracker, device_id: str = "motion_capture"): + def __init__(self, device_id: str = "cam_0", + camera_index: int = 0, + simulate: bool = False): + name = "Camera (Sim)" if simulate else f"Camera {camera_index}" super().__init__(DeviceInfo( device_id = device_id, - name = "Motion Capture", - device_type = "virtual", + name = name, + device_type = "camera", description = "Camera point-tracking β x/y position in pixels.", - icon = "π₯", + icon = "π·", channels = [ ChannelConfig( channel_id = "x_pos", @@ -48,38 +53,101 @@ class MotionCaptureDevice(BaseDevice): ), ], )) - self._tracker = tracker - self._lock = threading.Lock() - self._last = (0.0, 0.0) + self._camera_index = camera_index + self._simulate = simulate + self._tracker: Optional[Any] = None + self._tracking_win = None - # ββ BaseDevice interface ββββββββββββββββββββββββββββββββββββββββββββββ + # ββ BaseDevice ββββββββββββββββββββββββββββββββββββββββββββββββββββββββ def connect(self) -> bool: - self.status = DeviceStatus.SIMULATED + from tracker import CameraTracker + self._tracker = CameraTracker() + self._tracker.start(-1 if self._simulate else self._camera_index) + if self._tracker.simulated: + self.status = DeviceStatus.SIMULATED + else: + self.status = DeviceStatus.CONNECTED return True def disconnect(self) -> None: + if self._tracking_win is not None: + self._tracking_win.close() + self._tracking_win = None + if self._tracker is not None: + self._tracker.stop() + self._tracker = None self.status = DeviceStatus.DISCONNECTED def read_channels(self) -> Dict[str, float]: + if self._tracker is None: + return {"x_pos": 0.0, "y_pos": 0.0} pos = self._tracker.get_position() - if pos is not None: - with self._lock: - self._last = pos - with self._lock: - x, y = self._last - return {"x_pos": x, "y_pos": y} + if pos is None: + return {"x_pos": 0.0, "y_pos": 0.0} + x, y = pos + return {"x_pos": float(x), "y_pos": float(y)} def write_channel(self, channel_id: str, value: Any) -> bool: return False def get_config_widget(self): - from PyQt6.QtWidgets import QLabel - lbl = QLabel( - "Configure via the Motion Capture toolbar window.\n\n" - "Apply 'pixel_to_mm' filter in Signals β Channels\n" - "to convert pixel coordinates to millimetres." + from PyQt6.QtWidgets import ( + QFormLayout, QLabel, QPushButton, QVBoxLayout, QWidget, + ) + + w = QWidget() + lay = QVBoxLayout(w) + lay.setContentsMargins(8, 8, 8, 8) + lay.setSpacing(10) + + form = QFormLayout() + form.setContentsMargins(0, 0, 0, 0) + + src_lbl = QLabel( + "Simulation" if self._simulate else f"Camera {self._camera_index}" + ) + src_lbl.setObjectName("traceSource") + form.addRow("Source:", src_lbl) + + self._status_lbl = QLabel(self.status.name.title()) + self._status_lbl.setObjectName("traceSource") + form.addRow("Status:", self._status_lbl) + + lay.addLayout(form) + + open_btn = QPushButton("π₯ Open Camera View") + open_btn.setObjectName("addTraceBtn") + open_btn.clicked.connect(self._open_tracking_window) + lay.addWidget(open_btn) + + hint = QLabel( + "In the Camera View: click the frame to set the tracking\n" + "point. Apply the 'pixel_to_mm' filter in Signals β Channels\n" + "to convert pixel coordinates to real-world units." ) - lbl.setObjectName("traceSource") - lbl.setWordWrap(True) - return lbl + hint.setObjectName("traceSource") + hint.setWordWrap(True) + lay.addWidget(hint) + lay.addStretch() + + return w + + # ββ Internal ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ + + def _open_tracking_window(self): + if self._tracker is None: + return + from window import MotionCaptureWindow + if self._tracking_win is None: + self._tracking_win = MotionCaptureWindow(self._tracker) + self._tracking_win.closed.connect(self._on_tracking_win_closed) + self._tracking_win.show() + self._tracking_win.raise_() + + def _on_tracking_win_closed(self): + self._tracking_win = None + + +# Backward-compat alias for any profiles that reference the old class name +MotionCaptureDevice = CameraDevice diff --git a/plugins/motion_capture/plugin.py b/plugins/motion_capture/plugin.py index 1c655c5..6eb5508 100644 --- a/plugins/motion_capture/plugin.py +++ b/plugins/motion_capture/plugin.py @@ -3,30 +3,22 @@ motion_capture/plugin.py LabDAQ Motion Capture plugin. -Tracks a user-selected point via webcam (OpenCV CSRT tracker) and -streams x/y pixel coordinates as live channels into the acquisition -pipeline. Falls back to a Lissajous simulation when OpenCV is absent. - -Custom filter: - pixel_to_mm β convert pixel coords to millimetres using a - px_per_mm calibration value. Add it in Signals β Channels on - the x_pos or y_pos channel. - -Install OpenCV to use a real camera: - Arch Linux: sudo pacman -S python-opencv - Other: pip install opencv-python -""" +When enabled, adds a "Camera" device type to the Add Device dialog. +Each camera added by the user becomes a first-class device in the +acquisition pipeline β data flows through the standard +Device β Signal β Channel β Plot stream. -from __future__ import annotations +Custom filter registered globally: + pixel_to_mm β convert pixel coords to mm using a px_per_mm value. + Add it in Signals β Channels on x_pos or y_pos. -from typing import Any, Dict, List, Optional +Install OpenCV for real camera support: + pip install opencv-python +""" -from PyQt6.QtWidgets import ( - QComboBox, QDoubleSpinBox, QFormLayout, QHBoxLayout, - QLabel, QPushButton, QWidget, -) +from __future__ import annotations -from plugins.base_plugin import LabPlugin, PluginAction, PluginContext +from plugins.base_plugin import LabPlugin, PluginContext class MotionCapturePlugin(LabPlugin): @@ -44,8 +36,10 @@ class MotionCapturePlugin(LabPlugin): @property def description(self) -> str: - return ("Track a point via webcam and stream x/y position as live signals. " - "Apply pixel_to_mm filter to convert to real-world units.") + return ( + "Adds a Camera device type to the Add Device dialog. " + "Stream x/y tracking coordinates directly into the signal pipeline." + ) @property def author(self) -> str: return "LabDAQ" @@ -54,157 +48,16 @@ class MotionCapturePlugin(LabPlugin): def on_load(self, context: PluginContext) -> None: self._ctx = context - self._win: Optional[object] = None - self._toolbar_btn = None - - from tracker import CameraTracker - from device import MotionCaptureDevice - - self._tracker = CameraTracker() - self._device = MotionCaptureDevice(self._tracker) + from ui.add_device_dialog import _PANELS + from camera_panel import CameraPanel + _PANELS["Camera"] = (CameraPanel, "cam") def on_unload(self) -> None: - self._tracker.stop() - if self._win is not None: - self._win.close() - self._win = None + from ui.add_device_dialog import _PANELS + _PANELS.pop("Camera", None) # ββ Integration hooks βββββββββββββββββββββββββββββββββββββββββββββββββ - def get_devices(self) -> list: - return [self._device] - def get_filter_classes(self) -> dict: from filter import PixelToMMFilter return {"pixel_to_mm": PixelToMMFilter} - - def get_toolbar_actions(self) -> List[PluginAction]: - return [ - PluginAction( - label = "Motion Capture", - icon = "π₯", - tooltip = "Open motion capture window", - checkable = True, - callback = self._toggle_window, - button_ref_callback = self._store_btn, - ) - ] - - def get_settings_widget(self) -> Optional[QWidget]: - from tracker import list_cameras - w = QWidget() - lay = QFormLayout(w) - lay.setContentsMargins(0, 4, 0, 4) - - cam_row = QHBoxLayout() - self._s_cam = QComboBox() - self._s_cam.setObjectName("channelPickerCb") - self._s_cameras: list = [] - self._s_refresh_btn = QPushButton("β³") - self._s_refresh_btn.setObjectName("configButton") - self._s_refresh_btn.setFixedWidth(28) - self._s_refresh_btn.setToolTip("Rescan cameras") - self._s_refresh_btn.clicked.connect(lambda: self._populate_settings_cam(list_cameras)) - cam_row.addWidget(self._s_cam, 1) - cam_row.addWidget(self._s_refresh_btn) - lay.addRow("Camera:", cam_row) - - self._populate_settings_cam(list_cameras) - self._s_cam.currentIndexChanged.connect(self._sync_cam_to_window) - - self._s_pxmm = QDoubleSpinBox() - self._s_pxmm.setRange(0.01, 100_000) - self._s_pxmm.setDecimals(2) - self._s_pxmm.setSuffix(" px/mm") - self._s_pxmm.setValue( - self._win.get_px_per_mm() if self._win else 10.0) - self._s_pxmm.setObjectName("cfgGlobalSpin") - self._s_pxmm.valueChanged.connect(self._sync_pxmm_to_window) - lay.addRow("Calibration:", self._s_pxmm) - - hint = QLabel( - "Calibration: count how many pixels span a known real distance\n" - "in the camera frame, then divide pixels Γ· mm." - ) - hint.setObjectName("traceSource") - hint.setWordWrap(True) - lay.addRow(hint) - - return w - - # ββ Profile state βββββββββββββββββββββββββββββββββββββββββββββββββββββ - - def get_save_state(self) -> Dict[str, Any]: - return { - "camera_index": self._win.get_camera_index() if self._win else 0, - "px_per_mm": self._win.get_px_per_mm() if self._win else 10.0, - } - - def apply_save_state(self, state: Dict[str, Any]) -> None: - if self._win is not None: - self._win.set_camera_index(state.get("camera_index", 0)) - self._win.set_px_per_mm(state.get("px_per_mm", 10.0)) - # Store for when window is opened later - self._pending_state = state - - # ββ Internal ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ - - def _store_btn(self, btn) -> None: - self._toolbar_btn = btn - - def _toggle_window(self, checked: bool) -> None: - if self._win is None: - from window import MotionCaptureWindow - self._win = MotionCaptureWindow( - self._tracker, self._ctx.main_window) - self._win.closed.connect(self._on_win_closed) - - # Apply any state that arrived before window was created - state = getattr(self, "_pending_state", None) - if state: - self._win.set_camera_index(state.get("camera_index", 0)) - self._win.set_px_per_mm(state.get("px_per_mm", 10.0)) - - if checked: - if not self._win.isVisible(): - from PyQt6.QtWidgets import QApplication - mw = self._ctx.main_window - screen = QApplication.screenAt(mw.geometry().center()) or QApplication.primaryScreen() - avail = screen.availableGeometry() - geo = mw.normalGeometry() - self._win.adjustSize() - x = max(avail.left(), min(geo.right() + 8, avail.right() - self._win.width())) - y = max(avail.top(), min(geo.top() + 40, avail.bottom() - self._win.height())) - self._win.move(x, y) - self._win.show() - self._win.raise_() - else: - self._win.hide() - - def _on_win_closed(self) -> None: - if self._toolbar_btn is not None: - self._toolbar_btn.setChecked(False) - - def _populate_settings_cam(self, list_cameras_fn) -> None: - current = self._win.get_camera_index() if self._win else 0 - self._s_cam.blockSignals(True) - self._s_cam.clear() - self._s_cameras = list_cameras_fn() - for _, label in self._s_cameras: - self._s_cam.addItem(label) - for i, (idx, _) in enumerate(self._s_cameras): - if idx == current: - self._s_cam.setCurrentIndex(i) - break - self._s_cam.blockSignals(False) - - def _sync_cam_to_window(self, combo_i: int) -> None: - if not self._s_cameras: - return - idx = self._s_cameras[combo_i][0] - if self._win: - self._win.set_camera_index(idx) - - def _sync_pxmm_to_window(self, v: float) -> None: - if self._win: - self._win.set_px_per_mm(v) |