diff --git a/README.md b/README.md index 3a3683d..f2738cc 100755 --- a/README.md +++ b/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 diff --git a/gui/jog_panel.py b/gui/jog_panel.py new file mode 100644 index 0000000..1e9b703 --- /dev/null +++ b/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) diff --git a/sc3_aui_app.py b/sc3_aui_app.py index 285faa2..c96c524 100755 --- a/sc3_aui_app.py +++ b/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() diff --git a/tests/test_jog_panel.py b/tests/test_jog_panel.py new file mode 100644 index 0000000..95aacad --- /dev/null +++ b/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() == "---.---"