#!/usr/bin/env python3 """ BBD202 Stage Controller Test Application PyQt6 GUI for jogging the stage and configuring trigger outputs. Thomas Ales | Mar 2026 """ import sys import queue import time from PyQt6.QtWidgets import ( QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QGroupBox, QLabel, QLineEdit, QPushButton, QComboBox, QDoubleSpinBox, QStatusBar, QMessageBox, QGridLayout ) from PyQt6.QtCore import Qt, QThread, pyqtSignal, QObject, QTimer from PyQt6.QtGui import QFont from hardware.pybbd202 import ThorlabsServoDriver, AXIS_X, AXIS_Y from hardware.pybbd202.apt_constants import TriggerBitsServo # ── Worker thread ───────────────────────────────────────────────────────────── class StageCommand: def __init__(self, cmd, **kwargs): self.cmd = cmd self.params = kwargs class StageWorker(QObject): connected = pyqtSignal() disconnected = pyqtSignal() conn_failed = pyqtSignal(str) position_updated = pyqtSignal(float, float) # x_mm, y_mm status_updated = pyqtSignal(bool, bool, bool, bool) # x_homed, y_homed, x_moving, y_moving trigger_read = pyqtSignal(int, int) # x_mode, y_mode error_occurred = pyqtSignal(str) home_done = pyqtSignal(str) # 'x', 'y', or 'both' def __init__(self): super().__init__() self._driver = None self._queue = queue.Queue() self._running = True self._last_x = None self._last_y = None def enqueue(self, cmd, **kwargs): self._queue.put(StageCommand(cmd, **kwargs)) def run(self): self._poll_timer = QTimer() self._poll_timer.setInterval(200) self._poll_timer.timeout.connect(self._poll_status) while self._running: try: cmd = self._queue.get(timeout=0.05) self._dispatch(cmd) except queue.Empty: pass def _dispatch(self, cmd): try: if cmd.cmd == 'connect': self._do_connect(cmd.params['port']) elif cmd.cmd == 'disconnect': self._do_disconnect() elif cmd.cmd == 'home_x': self._do_home(AXIS_X, 'x') elif cmd.cmd == 'home_y': self._do_home(AXIS_Y, 'y') elif cmd.cmd == 'jog': self._do_jog(cmd.params['axis'], cmd.params['distance_mm']) elif cmd.cmd == 'move_abs': self._do_move_abs(cmd.params['axis'], cmd.params['pos_mm']) elif cmd.cmd == 'set_velocity': self._do_set_velocity(cmd.params['axis'], cmd.params['vel'], cmd.params['accel']) elif cmd.cmd == 'set_trigger': self._do_set_trigger(cmd.params['axis'], cmd.params['mode']) elif cmd.cmd == 'get_trigger': self._do_get_trigger() elif cmd.cmd == 'stop': pass # TODO: add stop message if needed except TimeoutError as e: self.error_occurred.emit(f"Timeout: {e}") except ValueError as e: self.error_occurred.emit(f"Value error: {e}") except Exception as e: self.error_occurred.emit(f"Error: {e}") def _do_connect(self, port): try: self._driver = ThorlabsServoDriver() self._driver.connect(port=port) self._driver.enable_axis(AXIS_X) self._driver.enable_axis(AXIS_Y) self._driver.start_polling(interval=0.2) time.sleep(0.5) # let first polls come in self.connected.emit() except Exception as e: self._driver = None self.conn_failed.emit(str(e)) def _do_disconnect(self): if self._driver: try: self._driver.disconnect() except Exception: pass self._driver = None self.disconnected.emit() def _do_home(self, axis, label): self._driver.home_axis(axis, timeout=60.0) self.home_done.emit(label) def _do_jog(self, axis, distance_mm): self._driver.move_axis_relative(axis, distance_mm) def _do_move_abs(self, axis, pos_mm): self._driver.move_axis_absolute(axis, pos_mm) def _do_set_velocity(self, axis, vel, accel): self._driver.set_velocity_params(axis, max_velocity=vel, acceleration=accel) def _do_set_trigger(self, axis, mode): self._driver.set_trigger(axis, mode) def _do_get_trigger(self): x_mode = int(self._driver.get_trigger(AXIS_X)) y_mode = int(self._driver.get_trigger(AXIS_Y)) self.trigger_read.emit(x_mode, y_mode) def _poll_status(self): if not self._driver: return x = self._driver.positions[0] y = self._driver.positions[1] if x != self._last_x or y != self._last_y: self._last_x = x self._last_y = y self.position_updated.emit(x, y) self.status_updated.emit( self._driver.am_homed[0], self._driver.am_homed[1], self._driver.am_moving[0], self._driver.am_moving[1] ) def stop(self): self._running = False # ── Main window ─────────────────────────────────────────────────────────────── TRIG_OPTIONS = [ ("Disabled", 0x00), ("In: Logic High", TriggerBitsServo.TRIGIN_HIGH), ("In: Relative Move", TriggerBitsServo.TRIGIN_RELMOVE), ("In: Absolute Move", TriggerBitsServo.TRIGIN_ABSMOVE), ("In: Home Move", TriggerBitsServo.TRIGIN_HOMEMOVE), ("Out: Logic High", TriggerBitsServo.TRIGOUT_HIGH), ("Out: In Motion", TriggerBitsServo.TRIGOUT_INMOTION), ("Out: Motion Complete", TriggerBitsServo.TRIGOUT_MOTIONCOMPLETE), ("Out: At Max Velocity", TriggerBitsServo.TRIGOUT_MAXVELOCITY), ("Out: High + Max Vel", TriggerBitsServo.TRIGOUT_MAXV), ] class BBD202TestApp(QMainWindow): def __init__(self): super().__init__() self.setWindowTitle("BBD202 Stage Controller Test") self.resize(700, 620) self._worker = StageWorker() self._thread = QThread() self._worker.moveToThread(self._thread) self._thread.started.connect(self._worker.run) self._worker.connected.connect(self._on_connected) self._worker.disconnected.connect(self._on_disconnected) self._worker.conn_failed.connect(self._on_conn_failed) self._worker.position_updated.connect(self._on_position_updated) self._worker.status_updated.connect(self._on_status_updated) self._worker.trigger_read.connect(self._on_trigger_read) self._worker.error_occurred.connect(self._on_error) self._worker.home_done.connect(self._on_home_done) self._thread.start() # Periodic status poll from worker — drive via a QTimer in main thread self._status_timer = QTimer() self._status_timer.setInterval(200) self._status_timer.timeout.connect(self._poll_worker_status) self._build_ui() self._set_connected(False) # ── UI construction ─────────────────────────────────────────────────────── def _build_ui(self): central = QWidget() self.setCentralWidget(central) root = QVBoxLayout(central) root.setSpacing(8) root.addWidget(self._build_connection_group()) root.addWidget(self._build_status_group()) root.addWidget(self._build_jog_group()) root.addWidget(self._build_velocity_group()) root.addWidget(self._build_trigger_group()) self.statusbar = QStatusBar() self.setStatusBar(self.statusbar) self.statusbar.showMessage("Not connected.") def _build_connection_group(self): grp = QGroupBox("Connection") lay = QHBoxLayout(grp) lay.addWidget(QLabel("Serial Port:")) self.le_port = QLineEdit("/dev/ttyUSB1") self.le_port.setMaximumWidth(150) lay.addWidget(self.le_port) self.btn_connect = QPushButton("Connect") self.btn_connect.clicked.connect(self._on_connect_clicked) lay.addWidget(self.btn_connect) self.btn_disconnect = QPushButton("Disconnect") self.btn_disconnect.clicked.connect(self._on_disconnect_clicked) lay.addWidget(self.btn_disconnect) lay.addStretch() return grp def _build_status_group(self): grp = QGroupBox("Status") grid = QGridLayout(grp) bold = QFont() bold.setBold(True) grid.addWidget(QLabel(""), 0, 0) lbl_x = QLabel("X"); lbl_x.setFont(bold) lbl_y = QLabel("Y"); lbl_y.setFont(bold) grid.addWidget(lbl_x, 0, 1, Qt.AlignmentFlag.AlignCenter) grid.addWidget(lbl_y, 0, 2, Qt.AlignmentFlag.AlignCenter) grid.addWidget(QLabel("Position (mm):"), 1, 0) self.lbl_x_pos = QLabel("---") self.lbl_y_pos = QLabel("---") grid.addWidget(self.lbl_x_pos, 1, 1, Qt.AlignmentFlag.AlignCenter) grid.addWidget(self.lbl_y_pos, 1, 2, Qt.AlignmentFlag.AlignCenter) grid.addWidget(QLabel("Homed:"), 2, 0) self.lbl_x_homed = QLabel("No") self.lbl_y_homed = QLabel("No") grid.addWidget(self.lbl_x_homed, 2, 1, Qt.AlignmentFlag.AlignCenter) grid.addWidget(self.lbl_y_homed, 2, 2, Qt.AlignmentFlag.AlignCenter) grid.addWidget(QLabel("Moving:"), 3, 0) self.lbl_x_moving = QLabel("No") self.lbl_y_moving = QLabel("No") grid.addWidget(self.lbl_x_moving, 3, 1, Qt.AlignmentFlag.AlignCenter) grid.addWidget(self.lbl_y_moving, 3, 2, Qt.AlignmentFlag.AlignCenter) # Home buttons self.btn_home_x = QPushButton("Home X") self.btn_home_y = QPushButton("Home Y") self.btn_home_x.clicked.connect(lambda: self._worker.enqueue('home_x')) self.btn_home_y.clicked.connect(lambda: self._worker.enqueue('home_y')) grid.addWidget(self.btn_home_x, 4, 1) grid.addWidget(self.btn_home_y, 4, 2) return grp def _build_jog_group(self): grp = QGroupBox("Jog / Manual Move") lay = QVBoxLayout(grp) # Step size step_row = QHBoxLayout() step_row.addWidget(QLabel("Step size (mm):")) self.dsb_step = QDoubleSpinBox() self.dsb_step.setRange(0.001, 50.0) self.dsb_step.setValue(1.0) self.dsb_step.setDecimals(3) self.dsb_step.setSingleStep(0.5) self.dsb_step.setMaximumWidth(100) step_row.addWidget(self.dsb_step) step_row.addStretch() lay.addLayout(step_row) # Jog buttons — arrow-style grid jog_grid = QGridLayout() jog_grid.setSpacing(4) self.btn_y_pos = QPushButton("Y +") self.btn_y_neg = QPushButton("Y −") self.btn_x_neg = QPushButton("← X −") self.btn_x_pos = QPushButton("X + →") for btn in (self.btn_y_pos, self.btn_y_neg, self.btn_x_neg, self.btn_x_pos): btn.setMinimumWidth(80) jog_grid.addWidget(self.btn_y_pos, 0, 1) jog_grid.addWidget(self.btn_x_neg, 1, 0) jog_grid.addWidget(self.btn_x_pos, 1, 2) jog_grid.addWidget(self.btn_y_neg, 2, 1) self.btn_y_pos.clicked.connect( lambda: self._worker.enqueue('jog', axis=AXIS_Y, distance_mm=self.dsb_step.value())) self.btn_y_neg.clicked.connect( lambda: self._worker.enqueue('jog', axis=AXIS_Y, distance_mm=-self.dsb_step.value())) self.btn_x_pos.clicked.connect( lambda: self._worker.enqueue('jog', axis=AXIS_X, distance_mm=self.dsb_step.value())) self.btn_x_neg.clicked.connect( lambda: self._worker.enqueue('jog', axis=AXIS_X, distance_mm=-self.dsb_step.value())) lay.addLayout(jog_grid) # Absolute move row abs_row = QHBoxLayout() abs_row.addWidget(QLabel("Go to X (mm):")) self.dsb_abs_x = QDoubleSpinBox() self.dsb_abs_x.setRange(0.0, 110.0) self.dsb_abs_x.setDecimals(3) self.dsb_abs_x.setMaximumWidth(100) abs_row.addWidget(self.dsb_abs_x) abs_row.addWidget(QLabel("Y (mm):")) self.dsb_abs_y = QDoubleSpinBox() self.dsb_abs_y.setRange(0.0, 75.0) self.dsb_abs_y.setDecimals(3) self.dsb_abs_y.setMaximumWidth(100) abs_row.addWidget(self.dsb_abs_y) btn_go = QPushButton("Move") btn_go.clicked.connect(self._on_abs_move_clicked) abs_row.addWidget(btn_go) abs_row.addStretch() lay.addLayout(abs_row) return grp def _build_velocity_group(self): grp = QGroupBox("Velocity Parameters") lay = QHBoxLayout(grp) lay.addWidget(QLabel("Max Vel (mm/s):")) self.dsb_vel = QDoubleSpinBox() self.dsb_vel.setRange(0.1, 300.0) self.dsb_vel.setValue(20.0) self.dsb_vel.setDecimals(1) self.dsb_vel.setMaximumWidth(90) lay.addWidget(self.dsb_vel) lay.addWidget(QLabel("Accel (mm/s²):")) self.dsb_accel = QDoubleSpinBox() self.dsb_accel.setRange(1.0, 2000.0) self.dsb_accel.setValue(100.0) self.dsb_accel.setDecimals(1) self.dsb_accel.setMaximumWidth(90) lay.addWidget(self.dsb_accel) lay.addWidget(QLabel("Axis:")) self.cmb_vel_axis = QComboBox() self.cmb_vel_axis.addItems(["X", "Y"]) lay.addWidget(self.cmb_vel_axis) btn_set_vel = QPushButton("Apply") btn_set_vel.clicked.connect(self._on_set_velocity_clicked) lay.addWidget(btn_set_vel) btn_read_vel = QPushButton("Read") btn_read_vel.clicked.connect(self._on_read_velocity_clicked) lay.addWidget(btn_read_vel) lay.addStretch() return grp def _build_trigger_group(self): grp = QGroupBox("Trigger Configuration") lay = QVBoxLayout(grp) grid = QGridLayout() bold = QFont(); bold.setBold(True) lbl_x = QLabel("X Axis"); lbl_x.setFont(bold) lbl_y = QLabel("Y Axis"); lbl_y.setFont(bold) grid.addWidget(lbl_x, 0, 1, Qt.AlignmentFlag.AlignCenter) grid.addWidget(lbl_y, 0, 2, Qt.AlignmentFlag.AlignCenter) grid.addWidget(QLabel("Trigger Mode:"), 1, 0) self.cmb_trig_x = QComboBox() self.cmb_trig_y = QComboBox() for name, _ in TRIG_OPTIONS: self.cmb_trig_x.addItem(name) self.cmb_trig_y.addItem(name) grid.addWidget(self.cmb_trig_x, 1, 1) grid.addWidget(self.cmb_trig_y, 1, 2) lay.addLayout(grid) btn_row = QHBoxLayout() btn_apply = QPushButton("Apply Trigger Settings") btn_apply.clicked.connect(self._on_apply_trigger_clicked) btn_row.addWidget(btn_apply) btn_read = QPushButton("Read from Controller") btn_read.clicked.connect(lambda: self._worker.enqueue('get_trigger')) btn_row.addWidget(btn_read) btn_row.addStretch() lay.addLayout(btn_row) return grp # ── UI state helpers ────────────────────────────────────────────────────── def _set_connected(self, connected): self.btn_connect.setEnabled(not connected) self.btn_disconnect.setEnabled(connected) self.le_port.setEnabled(not connected) for w in (self.btn_home_x, self.btn_home_y, self.btn_x_pos, self.btn_x_neg, self.btn_y_pos, self.btn_y_neg, self.dsb_step, self.dsb_abs_x, self.dsb_abs_y, self.dsb_vel, self.dsb_accel, self.cmb_vel_axis, self.cmb_trig_x, self.cmb_trig_y): w.setEnabled(connected) # The Apply / Read / Move buttons — find them by iterating children for btn in self.findChildren(QPushButton): if btn not in (self.btn_connect, self.btn_disconnect): btn.setEnabled(connected) # Keep connect/disconnect right self.btn_connect.setEnabled(not connected) self.btn_disconnect.setEnabled(connected) def _poll_worker_status(self): """Drive the worker's status poll from main thread timer.""" if self._worker._driver: self._worker._poll_status() # ── Slots ───────────────────────────────────────────────────────────────── def _on_connect_clicked(self): port = self.le_port.text().strip() if not port: QMessageBox.warning(self, "Input Error", "Enter a serial port.") return self.btn_connect.setEnabled(False) self.statusbar.showMessage(f"Connecting to {port}…") self._worker.enqueue('connect', port=port) def _on_disconnect_clicked(self): self._status_timer.stop() self._worker.enqueue('disconnect') def _on_connected(self): self._set_connected(True) self._status_timer.start() self.statusbar.showMessage("Connected.") def _on_disconnected(self): self._set_connected(False) self._status_timer.stop() self.lbl_x_pos.setText("---") self.lbl_y_pos.setText("---") self.lbl_x_homed.setText("No") self.lbl_y_homed.setText("No") self.lbl_x_moving.setText("No") self.lbl_y_moving.setText("No") self.statusbar.showMessage("Disconnected.") def _on_conn_failed(self, msg): self._set_connected(False) self.statusbar.showMessage(f"Connection failed: {msg}") QMessageBox.critical(self, "Connection Failed", msg) def _on_position_updated(self, x, y): self.lbl_x_pos.setText(f"{x:.3f}") self.lbl_y_pos.setText(f"{y:.3f}") def _on_status_updated(self, x_homed, y_homed, x_moving, y_moving): self.lbl_x_homed.setText("Yes" if x_homed else "No") self.lbl_y_homed.setText("Yes" if y_homed else "No") self.lbl_x_moving.setText("Yes" if x_moving else "No") self.lbl_y_moving.setText("Yes" if y_moving else "No") def _on_home_done(self, axis): self.statusbar.showMessage(f"{axis.upper()} homing complete.") def _on_abs_move_clicked(self): self._worker.enqueue('move_abs', axis=AXIS_X, pos_mm=self.dsb_abs_x.value()) self._worker.enqueue('move_abs', axis=AXIS_Y, pos_mm=self.dsb_abs_y.value()) def _on_set_velocity_clicked(self): axis = AXIS_X if self.cmb_vel_axis.currentText() == "X" else AXIS_Y self._worker.enqueue('set_velocity', axis=axis, vel=self.dsb_vel.value(), accel=self.dsb_accel.value()) self.statusbar.showMessage("Velocity parameters applied.") def _on_read_velocity_clicked(self): axis = AXIS_X if self.cmb_vel_axis.currentText() == "X" else AXIS_Y if not self._worker._driver: return try: params = self._worker._driver.get_velocity_params(axis) self.dsb_vel.setValue(params['max_velocity']) self.dsb_accel.setValue(params['acceleration']) self.statusbar.showMessage( f"Read: vel={params['max_velocity']:.1f} mm/s, " f"accel={params['acceleration']:.1f} mm/s²") except Exception as e: self._on_error(str(e)) def _on_apply_trigger_clicked(self): x_mode = TRIG_OPTIONS[self.cmb_trig_x.currentIndex()][1] y_mode = TRIG_OPTIONS[self.cmb_trig_y.currentIndex()][1] self._worker.enqueue('set_trigger', axis=AXIS_X, mode=int(x_mode)) self._worker.enqueue('set_trigger', axis=AXIS_Y, mode=int(y_mode)) self.statusbar.showMessage("Trigger settings applied.") def _on_trigger_read(self, x_mode, y_mode): def _find_idx(mode_val): for i, (_, v) in enumerate(TRIG_OPTIONS): if int(v) == mode_val: return i return 0 self.cmb_trig_x.setCurrentIndex(_find_idx(x_mode)) self.cmb_trig_y.setCurrentIndex(_find_idx(y_mode)) self.statusbar.showMessage( f"Trigger read: X=0x{x_mode:02X}, Y=0x{y_mode:02X}") def _on_error(self, msg): self.statusbar.showMessage(f"Error: {msg}") QMessageBox.warning(self, "Error", msg) # ── Cleanup ─────────────────────────────────────────────────────────────── def closeEvent(self, event): self._status_timer.stop() if self._worker._driver: self._worker._do_disconnect() self._worker.stop() self._thread.quit() self._thread.wait(3000) event.accept() # ── Entry point ─────────────────────────────────────────────────────────────── if __name__ == '__main__': app = QApplication(sys.argv) app.setStyle('Fusion') win = BBD202TestApp() win.show() sys.exit(app.exec())