Merge dev-camera-jog: jog controls in the camera window

README.md

@@ -84,6 +84,7 @@ scanengine-3/
 │   ├── inspect_bridge.py          # QtAngleInspector over core.angle_inspect
 │   ├── qt_t3r.py                  # Qt adapter over the T3R driver
 │   ├── qt_workers.py              # QueueWorker / PollingQueueWorker bases
+│   ├── jog_panel.py               # T3R + BBD202 jog controls (camera window)
 │   └── widgets.py                 # ConnectionBar, LogConsole, PortSelector…
 │
 ├── sc3_aui_app.py                 # Main acquisition application

gui/jog_panel.py

@@ -0,0 +1,392 @@
+"""Jog controls for the T3R axes and the BBD202 stage.
+
+These sit beside the camera image.  Focusing the T-axis and framing the
+sample on the XY stage are both done by eye, so the controls have to be
+reachable without looking away from the video.  Both panels drive the same
+driver/worker the main window uses, so a jog here is the same command as a
+jog there.
+
+The T3R panel jogs while the button is held (the controller's JOG command is
+a continuous velocity move, ended by STOP).  The BBD202 has no such command,
+so a held button repeats a short relative move, the way the main window
+already does it.
+"""
+
+from __future__ import annotations
+
+from PyQt6.QtCore import Qt, QTimer
+from PyQt6.QtGui import QFont
+from PyQt6.QtWidgets import (
+    QCheckBox, QComboBox, QDoubleSpinBox, QFrame, QGridLayout, QGroupBox,
+    QLabel, QPushButton, QSpinBox,
+)
+
+import hardware.t3r_protocol as proto
+from hardware.t3r_driver import T3RDriver
+
+# BBD202 jog defaults, shared with the main window's worker.
+BBD_JOG_STEP_MM     = 0.5       # default jog step
+BBD_JOG_SPEED_MM_S  = 10.0
+BBD_JOG_ACCEL_MM_S2 = 50.0
+
+# T3R jog defaults, matching the T3R control panel's own spin boxes.
+T3R_JOG_VELOCITY = 8000         # steps/s
+T3R_JOG_ACCEL    = 4000         # steps/s²
+T3R_MICROSTEPS   = 16           # shown until the device reports its own
+
+# A held BBD button repeats a relative move at this interval; the move itself
+# is short, so a faster repeat would just queue up moves the stage can't
+# finish (the main window uses the same 200 ms).
+BBD_JOG_REPEAT_MS = 200
+
+# Applying velocity on every spin-box click would flood the command queue, so
+# the write is debounced until the operator stops adjusting.
+BBD_VELOCITY_DEBOUNCE_MS = 300
+
+
+def _mono(size: int = 11) -> QFont:
+    f = QFont("Menlo")
+    f.setStyleHint(QFont.StyleHint.Monospace)
+    f.setPointSize(size)
+    return f
+
+
+def _hline() -> QFrame:
+    line = QFrame()
+    line.setFrameShape(QFrame.Shape.HLine)
+    line.setFrameShadow(QFrame.Shadow.Sunken)
+    return line
+
+
+def _jog_button(text: str, tip: str) -> QPushButton:
+    btn = QPushButton(text)
+    btn.setToolTip(tip)
+    btn.setFixedWidth(38)
+    btn.setAutoRepeat(False)     # the repeat is ours, on a timer
+    return btn
+
+
+# ── T3R ───────────────────────────────────────────────────────────────────────
+
+class T3RJogPanel(QGroupBox):
+    """Enable, microstep and jog controls for the four T3R axes.
+
+    Jog velocity and acceleration are shared by every axis; microstepping is
+    per-channel, because that is how the controller stores it.
+    """
+
+    def __init__(self, driver, parent=None):
+        super().__init__("T3R Axes", parent)
+        self._driver = driver
+        self._jogging: set[int] = set()
+        # Microsteps the operator picked but the device hasn't confirmed yet.
+        # Without this, an info poll already in flight carrying the old value
+        # would snap the combo back and look like the click was ignored.
+        self._pending_micro: dict[int, int] = {}
+        self._enable_chks: dict[int, QCheckBox] = {}
+        self._micro_combos: dict[int, QComboBox] = {}
+        self._pos_lbls: dict[int, QLabel] = {}
+        self._jog_btns: dict[tuple[int, int], QPushButton] = {}
+
+        self._build()
+
+        driver.handshake_ok.connect(lambda *_: self._set_online(True))
+        driver.disconnected.connect(lambda *_: self._set_online(False))
+        driver.info_updated.connect(self._on_info)
+        self._set_online(driver.is_open)
+
+    # ── Construction ──────────────────────────────────────────────────────────
+
+    def _build(self):
+        grid = QGridLayout(self)
+        grid.setVerticalSpacing(4)
+        grid.setHorizontalSpacing(6)
+        row = 0
+
+        grid.addWidget(QLabel("Velocity"), row, 0)
+        self.vel_spin = QSpinBox()
+        self.vel_spin.setRange(1, proto.MAX_VELOCITY)
+        self.vel_spin.setValue(T3R_JOG_VELOCITY)
+        self.vel_spin.setGroupSeparatorShown(True)
+        self.vel_spin.setSuffix(" steps/s")
+        grid.addWidget(self.vel_spin, row, 1, 1, 3)
+        row += 1
+
+        grid.addWidget(QLabel("Accel"), row, 0)
+        self.accel_spin = QSpinBox()
+        self.accel_spin.setRange(0, proto.MAX_ACCEL)
+        self.accel_spin.setValue(T3R_JOG_ACCEL)
+        self.accel_spin.setGroupSeparatorShown(True)
+        self.accel_spin.setSuffix(" steps/s²")
+        grid.addWidget(self.accel_spin, row, 1, 1, 3)
+        row += 1
+
+        grid.addWidget(_hline(), row, 0, 1, 5)
+        row += 1
+
+        hdr = QLabel("hold a jog button to move")
+        hdr.setStyleSheet("color: gray;")
+        grid.addWidget(hdr, row, 0, 1, 3)
+        grid.addWidget(QLabel("µsteps"), row, 3)
+        grid.addWidget(QLabel("pos"), row, 4)
+        row += 1
+
+        for ch, name in enumerate(T3RDriver.CHANNEL_NAMES):
+            chk = QCheckBox(name)
+            chk.setToolTip(f"Energise axis {ch} — a disabled axis ignores jogs")
+            chk.toggled.connect(lambda on, c=ch: self._on_enable_toggled(c, on))
+            grid.addWidget(chk, row, 0)
+            self._enable_chks[ch] = chk
+
+            for col, (text, direction, way) in enumerate(
+                    (("◀", -1, "negative"), ("▶", +1, "positive")), start=1):
+                btn = _jog_button(text, f"Jog {way} while held")
+                btn.pressed.connect(
+                    lambda c=ch, d=direction: self._start_jog(c, d))
+                btn.released.connect(lambda c=ch: self._stop_jog(c))
+                grid.addWidget(btn, row, col)
+                self._jog_btns[(ch, direction)] = btn
+
+            combo = QComboBox()
+            for m in proto.MICROSTEPS:
+                combo.addItem(str(m), m)
+            combo.setCurrentIndex(combo.findData(T3R_MICROSTEPS))
+            combo.setToolTip("SET_MICROSTEP — applied immediately, axis must be idle")
+            combo.activated.connect(lambda _i, c=ch: self._on_micro_selected(c))
+            grid.addWidget(combo, row, 3)
+            self._micro_combos[ch] = combo
+
+            pos_lbl = QLabel("—")
+            pos_lbl.setFont(_mono(11))
+            pos_lbl.setMinimumWidth(76)
+            pos_lbl.setAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter)
+            grid.addWidget(pos_lbl, row, 4)
+            self._pos_lbls[ch] = pos_lbl
+            row += 1
+
+        # Recovery for a jog whose button-release never arrived (window hidden
+        # or focus stolen mid-press).  It only stops axes this panel started,
+        # so it can never cut a scan's rotation short.
+        self.stop_btn = QPushButton("Stop jogging")
+        self.stop_btn.clicked.connect(self.stop_jogs)
+        grid.addWidget(self.stop_btn, row, 0, 1, 5)
+
+        grid.setColumnStretch(4, 1)
+
+    # ── Commands ──────────────────────────────────────────────────────────────
+
+    def _on_enable_toggled(self, ch: int, on: bool):
+        if on:
+            self._driver.enable(ch)
+        else:
+            self._driver.disable(ch)
+
+    def _on_micro_selected(self, ch: int):
+        micro = self._micro_combos[ch].currentData()
+        self._pending_micro[ch] = micro
+        self._driver.set_microstep(ch, micro)
+
+    def _start_jog(self, ch: int, direction: int):
+        self._jogging.add(ch)
+        self._driver.jog(ch, direction * self.vel_spin.value(),
+                         self.accel_spin.value())
+
+    def _stop_jog(self, ch: int):
+        if ch in self._jogging:
+            self._jogging.discard(ch)
+            self._driver.stop(ch, False)
+
+    def stop_jogs(self):
+        """Stop every axis this panel is jogging.  Safe to call when idle."""
+        for ch in sorted(self._jogging):
+            self._driver.stop(ch, False)
+        self._jogging.clear()
+
+    # ── Device updates ────────────────────────────────────────────────────────
+
+    def _on_info(self, ch: int, info):
+        lbl = self._pos_lbls.get(ch)
+        if lbl is None:
+            return
+        lbl.setText(f"{info.position}")
+
+        chk = self._enable_chks[ch]
+        if chk.isChecked() != info.enabled:
+            chk.blockSignals(True)
+            chk.setChecked(info.enabled)
+            chk.blockSignals(False)
+
+        pending = self._pending_micro.get(ch)
+        if pending is not None:
+            if info.microsteps != pending:
+                return                      # change still in flight
+            del self._pending_micro[ch]
+
+        combo = self._micro_combos[ch]
+        idx = combo.findData(info.microsteps)
+        if idx >= 0 and idx != combo.currentIndex():
+            combo.blockSignals(True)
+            combo.setCurrentIndex(idx)
+            combo.blockSignals(False)
+
+    def _set_online(self, on: bool):
+        if not on:
+            self._jogging.clear()
+            self._pending_micro.clear()
+            for lbl in self._pos_lbls.values():
+                lbl.setText("—")
+        self.setEnabled(on)
+
+
+# ── BBD202 ────────────────────────────────────────────────────────────────────
+
+class BBDJogPanel(QGroupBox):
+    """X/Y jog controls for the BBD202 stage.
+
+    ``worker`` is the main window's BBD202Worker; it is duck-typed here to
+    keep this module free of an import cycle with the app.  The stage runs
+    closed-loop brushless servos, so there is no microstepping to set —
+    velocity and acceleration are the equivalent knobs.
+    """
+
+    def __init__(self, worker, parent=None):
+        super().__init__("BBD202 XY Stage", parent)
+        self._worker = worker
+        self._active: tuple[str, int] | None = None
+
+        self._repeat = QTimer(self)
+        self._repeat.setInterval(BBD_JOG_REPEAT_MS)
+        self._repeat.timeout.connect(self._jog_tick)
+
+        self._vel_debounce = QTimer(self)
+        self._vel_debounce.setSingleShot(True)
+        self._vel_debounce.setInterval(BBD_VELOCITY_DEBOUNCE_MS)
+        self._vel_debounce.timeout.connect(self.apply_velocity)
+
+        self._build()
+
+        worker.connected.connect(self._on_connected)
+        worker.disconnected.connect(lambda: self._set_online(False))
+        worker.position_updated.connect(self._on_position)
+        self._set_online(worker.is_connected)
+
+    # ── Construction ──────────────────────────────────────────────────────────
+
+    def _build(self):
+        grid = QGridLayout(self)
+        grid.setVerticalSpacing(4)
+        grid.setHorizontalSpacing(6)
+        row = 0
+
+        grid.addWidget(QLabel("X"), row, 0)
+        self.x_pos_lbl = QLabel("---.---")
+        self.x_pos_lbl.setFont(_mono(11))
+        grid.addWidget(self.x_pos_lbl, row, 1)
+        grid.addWidget(QLabel("Y"), row, 2)
+        self.y_pos_lbl = QLabel("---.---")
+        self.y_pos_lbl.setFont(_mono(11))
+        grid.addWidget(self.y_pos_lbl, row, 3)
+        row += 1
+
+        grid.addWidget(_hline(), row, 0, 1, 4)
+        row += 1
+
+        # Jog pad: Y over X, laid out the way the stage moves.
+        self.y_pos_btn = _jog_button("▲", "Jog +Y while held")
+        grid.addWidget(self.y_pos_btn, row, 1, 1, 2, Qt.AlignmentFlag.AlignHCenter)
+        row += 1
+
+        self.x_neg_btn = _jog_button("◀", "Jog −X while held")
+        grid.addWidget(self.x_neg_btn, row, 1, Qt.AlignmentFlag.AlignRight)
+        self.x_pos_btn = _jog_button("▶", "Jog +X while held")
+        grid.addWidget(self.x_pos_btn, row, 2, Qt.AlignmentFlag.AlignLeft)
+        row += 1
+
+        self.y_neg_btn = _jog_button("▼", "Jog −Y while held")
+        grid.addWidget(self.y_neg_btn, row, 1, 1, 2, Qt.AlignmentFlag.AlignHCenter)
+        row += 1
+
+        for btn, axis, direction in (
+            (self.x_pos_btn, "x", +1), (self.x_neg_btn, "x", -1),
+            (self.y_pos_btn, "y", +1), (self.y_neg_btn, "y", -1),
+        ):
+            btn.pressed.connect(lambda a=axis, d=direction: self._start_jog(a, d))
+            btn.released.connect(self.stop_jogs)
+
+        grid.addWidget(QLabel("Step"), row, 0)
+        self.step_spin = QDoubleSpinBox()
+        self.step_spin.setRange(0.001, 25.0)
+        self.step_spin.setDecimals(3)
+        self.step_spin.setSingleStep(0.1)
+        self.step_spin.setValue(BBD_JOG_STEP_MM)
+        self.step_spin.setSuffix(" mm")
+        grid.addWidget(self.step_spin, row, 1, 1, 3)
+        row += 1
+
+        grid.addWidget(QLabel("Velocity"), row, 0)
+        self.vel_spin = QDoubleSpinBox()
+        self.vel_spin.setRange(0.1, 100.0)
+        self.vel_spin.setDecimals(1)
+        self.vel_spin.setValue(BBD_JOG_SPEED_MM_S)
+        self.vel_spin.setSuffix(" mm/s")
+        self.vel_spin.valueChanged.connect(lambda _v: self._vel_debounce.start())
+        grid.addWidget(self.vel_spin, row, 1, 1, 3)
+        row += 1
+
+        grid.addWidget(QLabel("Accel"), row, 0)
+        self.accel_spin = QDoubleSpinBox()
+        self.accel_spin.setRange(0.1, 500.0)
+        self.accel_spin.setDecimals(1)
+        self.accel_spin.setValue(BBD_JOG_ACCEL_MM_S2)
+        self.accel_spin.setSuffix(" mm/s²")
+        self.accel_spin.valueChanged.connect(lambda _v: self._vel_debounce.start())
+        grid.addWidget(self.accel_spin, row, 1, 1, 3)
+        row += 1
+
+        note = QLabel("Microstepping: n/a — closed-loop servo")
+        note.setStyleSheet("color: gray;")
+        note.setWordWrap(True)
+        grid.addWidget(note, row, 0, 1, 4)
+
+        grid.setColumnStretch(3, 1)
+
+    # ── Commands ──────────────────────────────────────────────────────────────
+
+    def _start_jog(self, axis: str, direction: int):
+        self._active = (axis, direction)
+        self._jog_tick()
+        self._repeat.start()
+
+    def _jog_tick(self):
+        if self._active is None:
+            return
+        axis, direction = self._active
+        self._worker.queue_jog(axis, direction, step_mm=self.step_spin.value())
+
+    def stop_jogs(self):
+        """Stop the repeat.  A move already sent runs to its (short) end."""
+        self._repeat.stop()
+        self._active = None
+
+    def apply_velocity(self):
+        """Push the panel's velocity/acceleration to both axes."""
+        if self._worker.is_connected:
+            self._worker.queue_set_velocity(self.vel_spin.value(),
+                                            self.accel_spin.value())
+
+    # ── Device updates ────────────────────────────────────────────────────────
+
+    def _on_connected(self):
+        self._set_online(True)
+        self.apply_velocity()
+
+    def _on_position(self, x_mm: float, y_mm: float):
+        self.x_pos_lbl.setText(f"{x_mm:07.3f}")
+        self.y_pos_lbl.setText(f"{y_mm:07.3f}")
+
+    def _set_online(self, on: bool):
+        if not on:
+            self.stop_jogs()
+            self.x_pos_lbl.setText("---.---")
+            self.y_pos_lbl.setText("---.---")
+        self.setEnabled(on)

sc3_aui_app.py

@@ -40,6 +40,10 @@ from core.scope_sras import SAMPLE_RATE_HZ, configure_channels
 from core.sras_format import (
     SCAN_CHANNELS, VERSION, VERSION_SAW_CHECK, SrasFile, plan_from_header,
 )
+from gui.jog_panel import (
+    BBD_JOG_ACCEL_MM_S2, BBD_JOG_SPEED_MM_S, BBD_JOG_STEP_MM,
+    BBDJogPanel, T3RJogPanel,
+)
 from gui.qt_t3r import QtT3RAdapter
 from gui.qt_workers import PollingQueueWorker, QueueWorker
 from gui.inspect_bridge import QtAngleInspector
@@ -54,7 +58,6 @@ from t3r_control_panel import T3RControlPanel
 
 DEFAULTS = ScanDefaults.load()
 
-BBD_DEFAULT_JOG_MM  = 0.5       # default jog step for BBD202
 # A SAW check is written beside the scan it belongs to, under the same prefix.
 # The suffix keeps it from overwriting the scan itself, which is the one file
 # in the directory that cost hours to acquire.
@@ -136,9 +139,6 @@ class DCBiasImageWidget(FigureCanvas):
         self.draw_idle()
 
 
-BBD_JOG_SPEED_MM_S  = 10.0
-BBD_JOG_ACCEL_MM_S2 = 50.0
-
 # ── BBD202 worker ─────────────────────────────────────────────────────────────
 
 class BBD202Worker(PollingQueueWorker):
@@ -152,13 +152,14 @@ class BBD202Worker(PollingQueueWorker):
         super().__init__(poll_interval_s=self.POLL_INTERVAL_S)
         self.controller: ThorlabsServoDriver | None = None
         self.scanning_active = False   # pause polling during a scan
-        self._jog_step = BBD_DEFAULT_JOG_MM
+        self._jog_step = BBD_JOG_STEP_MM
         self._last_pos: tuple[float, float] | None = None
         self._last_homed: tuple[bool, bool] | None = None
         self._handlers.update({
             "connect": self._do_connect,
             "disconnect": self._do_disconnect,
             "jog": self._do_jog,
+            "set_velocity": self._do_set_velocity,
             "home_all": self._do_home_all,
             "enable_all": self._do_enable_all,
         })
@@ -200,15 +201,23 @@ class BBD202Worker(PollingQueueWorker):
         self.is_connected = False
         self.disconnected.emit()
 
-    def _do_jog(self, axis: str, direction: int):
+    def _do_jog(self, axis: str, direction: int, step_mm: float | None = None):
         if not self.controller:
             return
         dest = AXIS_X if axis == "x" else AXIS_Y
+        step = self._jog_step if step_mm is None else step_mm
         try:
-            self.controller.move_axis_relative(dest, self._jog_step * direction, timeout=2.0)
+            self.controller.move_axis_relative(dest, step * direction, timeout=2.0)
         except TimeoutError:
             pass
 
+    def _do_set_velocity(self, max_velocity: float, acceleration: float):
+        if not self.controller:
+            return
+        for dest in (AXIS_X, AXIS_Y):
+            self.controller.set_velocity_params(dest, max_velocity=max_velocity,
+                                                acceleration=acceleration)
+
     def _do_home_all(self):
         if not self.controller:
             return
@@ -250,8 +259,12 @@ class BBD202Worker(PollingQueueWorker):
     def queue_disconnect(self):
         self._enqueue("disconnect")
 
-    def queue_jog(self, axis: str, direction: int):
-        self._enqueue("jog", axis=axis, direction=direction)
+    def queue_jog(self, axis: str, direction: int, step_mm: float | None = None):
+        self._enqueue("jog", axis=axis, direction=direction, step_mm=step_mm)
+
+    def queue_set_velocity(self, max_velocity: float, acceleration: float):
+        self._enqueue("set_velocity", max_velocity=max_velocity,
+                      acceleration=acceleration)
 
     def queue_home_all(self):
         self._enqueue("home_all")
@@ -426,11 +439,18 @@ class HeliosWorker(PollingQueueWorker):
 # ── Camera window popup ───────────────────────────────────────────────────────
 
 class CameraWindow(QWidget):
-    """Camera display popup — auto-connects on show, auto-disconnects on close."""
+    """Camera display popup — auto-connects on show, auto-disconnects on close.
+
+    Focusing and framing are done by eye, so the T3R and BBD202 jog controls
+    live beside the image.  Both take the objects the main window already
+    owns; passing neither leaves the window as a plain viewer.
+    """
 
     closed = pyqtSignal()
 
-    def __init__(self, parent: QWidget | None = None):
+    def __init__(self, t3r_driver: QtT3RAdapter | None = None,
+                 bbd_worker: BBD202Worker | None = None,
+                 parent: QWidget | None = None):
         super().__init__(parent, Qt.WindowType.Window)
         uic.loadUi(ROOT / "sc3-aui-camera.ui", self)
         self.setWindowTitle("uC480 Camera")
@@ -468,14 +488,37 @@ class CameraWindow(QWidget):
         self.uc480_stop_btn.clicked.connect(self._stop_stream)
         self.uc480_close_window_btn.clicked.connect(self.close)
 
+        self.t3r_jog_panel = self.bbd_jog_panel = None
+        self._build_jog_column(t3r_driver, bbd_worker)
+
         self._update_controls()
 
+    def _build_jog_column(self, t3r_driver, bbd_worker):
+        """Add the stage controls to the right of the image."""
+        if t3r_driver is None and bbd_worker is None:
+            return
+        column = QVBoxLayout()
+        column.setContentsMargins(0, 0, 0, 0)
+        if t3r_driver is not None:
+            self.t3r_jog_panel = T3RJogPanel(t3r_driver, self)
+            column.addWidget(self.t3r_jog_panel)
+        if bbd_worker is not None:
+            self.bbd_jog_panel = BBDJogPanel(bbd_worker, self)
+            column.addWidget(self.bbd_jog_panel)
+        column.addStretch()
+        self.horizontalLayout.addLayout(column)
+
     def showEvent(self, event):
         super().showEvent(event)
         self._connect_camera()
         self._start_stream()
 
     def closeEvent(self, event):
+        # A jog whose button-release lands after the window is gone would
+        # otherwise leave an axis running.
+        for panel in (self.t3r_jog_panel, self.bbd_jog_panel):
+            if panel is not None:
+                panel.stop_jogs()
         self._stop_stream()
         self._disconnect_camera()
         super().closeEvent(event)
@@ -980,7 +1023,7 @@ class MainWindow(QMainWindow):
         self._helios_thread.started.connect(self._helios_worker.run)
 
         # ── Popup windows ─────────────────────────────────────────────────────
-        self._camera_win    = CameraWindow()
+        self._camera_win    = CameraWindow(self._t3r_driver, self._bbd_worker)
         self._helios_win    = HeliosWindow(self._helios_worker)
         self._scan_progress = ScanProgressWindow()
 
@@ -1022,7 +1065,7 @@ class MainWindow(QMainWindow):
         self.bbd202_comport_edit.setText(DEFAULTS.bbd_port)
         self.oscope_ip_edit.setText(DEFAULTS.oscope_ip)
         self._helios_win.helios_port_edit.setText(DEFAULTS.helios_port)
-        self.lineEdit_10.setText(str(BBD_DEFAULT_JOG_MM))
+        self.lineEdit_10.setText(str(BBD_JOG_STEP_MM))
 
         self.x_start_edit.setText("10.000")
         self.y_start_edit.setText("10.000")
@@ -1188,10 +1231,12 @@ class MainWindow(QMainWindow):
             return
         axis, direction = self._bbd_active_jog
         try:
-            self._bbd_worker._jog_step = float(self.lineEdit_10.text())
+            step = float(self.lineEdit_10.text())
         except ValueError:
-            self._bbd_worker._jog_step = BBD_DEFAULT_JOG_MM
-        self._bbd_worker.queue_jog(axis, direction)
+            step = BBD_JOG_STEP_MM
+        # The step rides along with the command: writing it onto the worker
+        # from here would be a cross-thread poke at a field the worker reads.
+        self._bbd_worker.queue_jog(axis, direction, step_mm=step)
 
     def _stop_bbd_jog(self):
         self._bbd_jog_timer.stop()

tests/test_jog_panel.py

@@ -0,0 +1,216 @@
+"""gui.jog_panel: the camera window's T3R and BBD202 jog controls.
+
+The panels are the only place these devices are driven by a held button, so
+what matters here is that press/release map onto the right pair of commands
+and that the operator's velocity/microstep settings ride along.
+"""
+import pytest
+from PyQt6.QtCore import QObject, pyqtSignal
+from PyQt6.QtWidgets import QApplication
+
+import hardware.t3r_protocol as proto
+from gui.jog_panel import BBDJogPanel, T3RJogPanel
+
+
+@pytest.fixture(scope="module")
+def qapp():
+    yield QApplication.instance() or QApplication([])
+
+
+class FakeT3R(QObject):
+    """The slice of QtT3RAdapter the T3R panel touches."""
+    handshake_ok = pyqtSignal(int, int, int)
+    disconnected = pyqtSignal(str)
+    info_updated = pyqtSignal(int, object)
+
+    def __init__(self, is_open=True):
+        super().__init__()
+        self.is_open = is_open
+        self.calls = []
+
+    def enable(self, ch):
+        self.calls.append(("enable", ch))
+
+    def disable(self, ch):
+        self.calls.append(("disable", ch))
+
+    def set_microstep(self, ch, microsteps):
+        self.calls.append(("set_microstep", ch, microsteps))
+
+    def jog(self, ch, velocity, accel):
+        self.calls.append(("jog", ch, velocity, accel))
+
+    def stop(self, ch, hard):
+        self.calls.append(("stop", ch, hard))
+
+
+class FakeBBD(QObject):
+    """The slice of BBD202Worker the BBD panel touches."""
+    connected = pyqtSignal()
+    disconnected = pyqtSignal()
+    position_updated = pyqtSignal(float, float)
+
+    def __init__(self, is_connected=True):
+        super().__init__()
+        self.is_connected = is_connected
+        self.calls = []
+
+    def queue_jog(self, axis, direction, step_mm=None):
+        self.calls.append(("jog", axis, direction, step_mm))
+
+    def queue_set_velocity(self, max_velocity, acceleration):
+        self.calls.append(("velocity", max_velocity, acceleration))
+
+
+def _info(ch, position=0, microsteps=16, enabled=True):
+    return proto.Info(ch=ch, state=0, position=position, velocity=0,
+                      microsteps=microsteps, run_ma=800, hold_ma=400,
+                      enabled=enabled, comms_ok=True, fault_mask=0)
+
+
+# ── T3R ───────────────────────────────────────────────────────────────────────
+
+def test_t3r_jog_holds_then_stops(qapp):
+    drv = FakeT3R()
+    panel = T3RJogPanel(drv)
+    panel.vel_spin.setValue(1234)
+    panel.accel_spin.setValue(99)
+
+    btn = panel._jog_btns[(0, -1)]
+    btn.pressed.emit()
+    assert drv.calls == [("jog", 0, -1234, 99)]
+
+    btn.released.emit()
+    assert drv.calls[-1] == ("stop", 0, False)
+
+
+def test_t3r_release_without_press_sends_nothing(qapp):
+    """A stray release must not stop an axis a scan is driving."""
+    drv = FakeT3R()
+    panel = T3RJogPanel(drv)
+    panel._jog_btns[(3, 1)].released.emit()
+    assert drv.calls == []
+
+
+def test_t3r_stop_jogs_covers_a_lost_release(qapp):
+    drv = FakeT3R()
+    panel = T3RJogPanel(drv)
+    panel._jog_btns[(1, 1)].pressed.emit()
+    drv.calls.clear()
+
+    panel.stop_jogs()
+    assert drv.calls == [("stop", 1, False)]
+    panel.stop_jogs()                      # already stopped: no repeat command
+    assert drv.calls == [("stop", 1, False)]
+
+
+def test_t3r_microstep_applies_and_survives_a_stale_poll(qapp):
+    drv = FakeT3R()
+    panel = T3RJogPanel(drv)
+    combo = panel._micro_combos[2]
+
+    combo.setCurrentIndex(combo.findData(64))
+    combo.activated.emit(combo.currentIndex())
+    assert drv.calls == [("set_microstep", 2, 64)]
+
+    # An info frame already in flight still carries the old value.
+    drv.info_updated.emit(2, _info(2, microsteps=16))
+    assert combo.currentData() == 64
+
+    # Once the device confirms, the combo tracks it again.
+    drv.info_updated.emit(2, _info(2, microsteps=64))
+    assert combo.currentData() == 64
+    drv.info_updated.emit(2, _info(2, microsteps=8))
+    assert combo.currentData() == 8
+
+
+def test_t3r_enable_checkbox_follows_the_device(qapp):
+    drv = FakeT3R()
+    panel = T3RJogPanel(drv)
+
+    panel._enable_chks[0].setChecked(True)
+    assert drv.calls == [("enable", 0)]
+
+    # A device-side state change updates the box without echoing a command.
+    drv.info_updated.emit(0, _info(0, position=-42, enabled=False))
+    assert not panel._enable_chks[0].isChecked()
+    assert drv.calls == [("enable", 0)]
+    assert panel._pos_lbls[0].text() == "-42"
+
+
+def test_t3r_panel_tracks_connection(qapp):
+    drv = FakeT3R(is_open=False)
+    panel = T3RJogPanel(drv)
+    assert not panel.isEnabled()
+
+    drv.handshake_ok.emit(1, 1, 4)
+    assert panel.isEnabled()
+
+    drv.info_updated.emit(0, _info(0, position=7))
+    drv.disconnected.emit("cable")
+    assert not panel.isEnabled()
+    assert panel._pos_lbls[0].text() == "—"
+
+
+# ── BBD202 ────────────────────────────────────────────────────────────────────
+
+def test_bbd_jog_repeats_while_held(qapp):
+    worker = FakeBBD()
+    panel = BBDJogPanel(worker)
+    panel.step_spin.setValue(0.25)
+
+    panel.x_pos_btn.pressed.emit()
+    assert worker.calls == [("jog", "x", 1, 0.25)]
+    assert panel._repeat.isActive()
+
+    panel._jog_tick()                      # what the repeat timer fires
+    assert worker.calls[-1] == ("jog", "x", 1, 0.25)
+
+    panel.x_pos_btn.released.emit()
+    assert not panel._repeat.isActive()
+    panel._jog_tick()                      # a late tick moves nothing
+    assert len(worker.calls) == 2
+
+
+def test_bbd_jog_directions(qapp):
+    worker = FakeBBD()
+    panel = BBDJogPanel(worker)
+    for btn, expected in ((panel.x_neg_btn, ("jog", "x", -1, 0.5)),
+                          (panel.y_pos_btn, ("jog", "y", 1, 0.5)),
+                          (panel.y_neg_btn, ("jog", "y", -1, 0.5))):
+        btn.pressed.emit()
+        btn.released.emit()
+        assert worker.calls[-1] == expected
+
+
+def test_bbd_velocity_is_debounced_then_applied(qapp):
+    worker = FakeBBD()
+    panel = BBDJogPanel(worker)
+
+    panel.vel_spin.setValue(4.0)
+    panel.accel_spin.setValue(20.0)
+    assert worker.calls == []              # nothing sent mid-adjustment
+    assert panel._vel_debounce.isActive()
+
+    panel.apply_velocity()
+    assert worker.calls == [("velocity", 4.0, 20.0)]
+
+
+def test_bbd_panel_tracks_connection(qapp):
+    worker = FakeBBD(is_connected=False)
+    panel = BBDJogPanel(worker)
+    assert not panel.isEnabled()
+
+    worker.position_updated.emit(12.0, 34.5)
+    assert panel.x_pos_lbl.text() == "012.000"
+    assert panel.y_pos_lbl.text() == "034.500"
+
+    worker.is_connected = True
+    worker.connected.emit()
+    assert panel.isEnabled()
+    assert worker.calls == [("velocity", panel.vel_spin.value(),
+                             panel.accel_spin.value())]
+
+    worker.disconnected.emit()
+    assert not panel.isEnabled()
+    assert panel.x_pos_lbl.text() == "---.---"