tka/scanengine-3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2041fc8439b2cffb4daf77b9c5c9fc380260d90b
scanengine-3/bbd202_test_app.py
T
Thomas Ales 2041fc8439 pre merge cleanup commit
2026-07-28 09:20:58 -05:00

575 lines
22 KiB
Python
Executable File
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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, QCheckBox, QFrame
)
from PyQt6.QtCore import Qt, QThread, pyqtSignal, QObject, QTimer
from PyQt6.QtGui import QFont, QKeySequence, QShortcut
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())
Reference in New Issue View Git Blame Copy Permalink