pre uc480 integration

This commit is contained in:
Thomas Ales [M S E]
2026-05-22 09:38:39 -05:00
parent 43e7b99512
commit a0e0151b5d
30 changed files with 12557 additions and 41 deletions
+574
View File
@@ -0,0 +1,574 @@
#!/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())