Thomas Ales
afe33249d1
The headline of the refactor. Scan orchestration no longer lives inside a QObject that reaches through Qt workers for its hardware handles. core/scan_engine.py — ScanEngine(stage, scope, rotator, plan, out_path, resume, callbacks). Takes the concrete drivers, blocks in run(), reports via plain callables, and prompts through an injected blocking callable. No Qt import anywhere in the path (test_engine_imports_without_qt proves it), so a simpler GUI or a CLI can drive the identical acquisition. Supporting extractions, all Qt-free: - core/scope_sras.py — SCPI policy: channel profiles, trigger programming, background average, per-row FastFrame transfer - core/rotation.py — RotationAxis + RotationSettings (the GR_* constants) - core/scan_resume.py — frontier contiguity rule + settings compatibility - gui/scan_bridge.py — QtScanController, exposing exactly the signal surface the old ScanWorker had, so MainWindow's connections are unchanged hardware/t3r_driver.py is now Qt-free: a plain Signal class, a threading reader, and a polling thread instead of QObject/QThread/QTimer. gui/qt_t3r.py re-emits its callbacks as queued Qt signals for the panels. Fixes carried by the extraction: - rotation waits on the driver's MOTION_DONE event instead of time.sleep(estimate + 0.5) - abort during an operator prompt now takes effect; the old _prompt_event.wait() had no timeout and could not be interrupted - the poll timer is a thread, so an I/O error tearing down the driver no longer calls QTimer.stop() from the wrong thread - T3RDriver.disconnect() renamed close(); it shadowed QObject.disconnect() - per-frame DC means use np.frombuffer over the joined block instead of struct.unpack per frame (~16k tuple allocations per row) tests/fakes.py + test_scan_engine.py (14 tests) assert the exact command sequence, file layout, resume seeking, abort/pause, and geometry rejection before any hardware call; test_scan_resume.py covers the frontier rule. 58 passing. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
716 lines
28 KiB
Python
Executable File
716 lines
28 KiB
Python
Executable File
"""T3R Focusing & Rotation Control Panel.
|
||
|
||
A user-hidable QDialog that provides full control over the T3R four-channel
|
||
stepper controller. Mirrors the functionality of /opt/t3r-firmware/tester.py
|
||
and adds a Stage Rotation section that computes the correct GR-axis step count
|
||
from the physical gear train.
|
||
|
||
Gear train (ch3 = GR-axis):
|
||
Motor → 10T pinion → 30T idler → 125T index gear (stage)
|
||
Motor:stage ratio = 125/10 = 12.5
|
||
|
||
Usage::
|
||
|
||
from t3r_control_panel import T3RControlPanel
|
||
panel = T3RControlPanel(driver)
|
||
panel.show()
|
||
# toggle with panel.setVisible(not panel.isVisible())
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from PyQt6.QtCore import Qt
|
||
from PyQt6.QtGui import QFont
|
||
from PyQt6.QtWidgets import (
|
||
QCheckBox, QComboBox, QDialog, QDoubleSpinBox, QFrame, QGridLayout,
|
||
QGroupBox, QHBoxLayout, QLabel, QPlainTextEdit, QPushButton, QScrollArea,
|
||
QSpinBox, QSplitter, QVBoxLayout, QWidget,
|
||
)
|
||
|
||
from gui.qt_t3r import QtT3RAdapter
|
||
from hardware.serial_util import scored_ports
|
||
from hardware.t3r_driver import T3RDriver # class constants (gear train, channel names)
|
||
import hardware.t3r_protocol as proto
|
||
|
||
|
||
# ── Utilities ─────────────────────────────────────────────────────────────────
|
||
|
||
def _hline() -> QFrame:
|
||
line = QFrame()
|
||
line.setFrameShape(QFrame.Shape.HLine)
|
||
line.setFrameShadow(QFrame.Shadow.Sunken)
|
||
return line
|
||
|
||
|
||
def _mono_font(size: int = 11) -> QFont:
|
||
f = QFont("Menlo")
|
||
f.setStyleHint(QFont.StyleHint.Monospace)
|
||
f.setPointSize(size)
|
||
return f
|
||
|
||
|
||
def _spin(lo: int, hi: int, val: int) -> QSpinBox:
|
||
s = QSpinBox()
|
||
s.setRange(lo, hi)
|
||
s.setValue(val)
|
||
s.setGroupSeparatorShown(True)
|
||
s.setMaximumWidth(130)
|
||
return s
|
||
|
||
|
||
# ── Per-channel panel ─────────────────────────────────────────────────────────
|
||
|
||
class ChannelPanel(QGroupBox):
|
||
"""Controls and live readouts for one T3R axis."""
|
||
|
||
def __init__(self, ch: int, driver: QtT3RAdapter):
|
||
label = f"Axis {ch} — {T3RDriver.CHANNEL_NAMES[ch]}"
|
||
super().__init__(label)
|
||
self.ch = ch
|
||
self._driver = driver
|
||
self._build()
|
||
|
||
driver.info_updated.connect(self._on_info)
|
||
driver.drv_status_updated.connect(self._on_drv_status)
|
||
driver.motion_done.connect(self._on_event_pos)
|
||
driver.stopped.connect(self._on_event_pos)
|
||
driver.fault_occurred.connect(self._on_fault_event)
|
||
|
||
def _build(self):
|
||
grid = QGridLayout(self)
|
||
grid.setVerticalSpacing(4)
|
||
grid.setHorizontalSpacing(8)
|
||
row = 0
|
||
|
||
# Enable + state + position
|
||
self.enable_chk = QCheckBox("Enabled")
|
||
self.enable_chk.toggled.connect(self._on_enable_toggled)
|
||
grid.addWidget(self.enable_chk, row, 0)
|
||
|
||
self.state_lbl = QLabel("—")
|
||
self.state_lbl.setMinimumWidth(72)
|
||
grid.addWidget(self.state_lbl, row, 1)
|
||
|
||
lbl = QLabel("pos:")
|
||
lbl.setAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter)
|
||
grid.addWidget(lbl, row, 2)
|
||
self.pos_lbl = QLabel("0")
|
||
self.pos_lbl.setFont(_mono_font(13))
|
||
grid.addWidget(self.pos_lbl, row, 3, 1, 2)
|
||
row += 1
|
||
|
||
self.fault_lbl = QLabel("faults: none")
|
||
self.fault_lbl.setStyleSheet("color: #2e7d32;")
|
||
grid.addWidget(self.fault_lbl, row, 0, 1, 3)
|
||
self.comms_lbl = QLabel("comms: —")
|
||
grid.addWidget(self.comms_lbl, row, 3, 1, 2)
|
||
row += 1
|
||
|
||
grid.addWidget(_hline(), row, 0, 1, 5); row += 1
|
||
|
||
# Configuration
|
||
grid.addWidget(QLabel("Microsteps"), row, 0)
|
||
self.micro_combo = QComboBox()
|
||
for m in proto.MICROSTEPS:
|
||
self.micro_combo.addItem(str(m), m)
|
||
self.micro_combo.setCurrentText("16")
|
||
grid.addWidget(self.micro_combo, row, 1)
|
||
|
||
grid.addWidget(QLabel("Run mA"), row, 2)
|
||
self.run_spin = _spin(0, 3000, 800)
|
||
grid.addWidget(self.run_spin, row, 3)
|
||
row += 1
|
||
|
||
grid.addWidget(QLabel("ihold delay"), row, 0)
|
||
self.ihold_spin = _spin(0, 15, 6)
|
||
grid.addWidget(self.ihold_spin, row, 1)
|
||
|
||
grid.addWidget(QLabel("Hold mA"), row, 2)
|
||
self.hold_spin = _spin(0, 3000, 400)
|
||
grid.addWidget(self.hold_spin, row, 3)
|
||
|
||
apply_btn = QPushButton("Apply config")
|
||
apply_btn.setToolTip("Send SET_MICROSTEP + SET_CURRENT (channel must be idle)")
|
||
apply_btn.clicked.connect(self._on_apply_config)
|
||
grid.addWidget(apply_btn, row, 4)
|
||
row += 1
|
||
|
||
self.achieved_lbl = QLabel("achieved: run —/hold — mA, — µsteps")
|
||
self.achieved_lbl.setStyleSheet("color: gray;")
|
||
grid.addWidget(self.achieved_lbl, row, 0, 1, 5)
|
||
row += 1
|
||
|
||
self.drv_lbl = QLabel("driver status: —")
|
||
self.drv_lbl.setStyleSheet("color: gray;")
|
||
grid.addWidget(self.drv_lbl, row, 0, 1, 5)
|
||
row += 1
|
||
|
||
grid.addWidget(_hline(), row, 0, 1, 5); row += 1
|
||
|
||
# Motion parameters
|
||
grid.addWidget(QLabel("Max vel (steps/s)"), row, 0)
|
||
self.vel_spin = _spin(1, proto.MAX_VELOCITY, 8000)
|
||
grid.addWidget(self.vel_spin, row, 1)
|
||
|
||
grid.addWidget(QLabel("Accel (steps/s²)"), row, 2)
|
||
self.accel_spin = _spin(0, proto.MAX_ACCEL, 4000)
|
||
grid.addWidget(self.accel_spin, row, 3)
|
||
row += 1
|
||
|
||
# Move
|
||
grid.addWidget(QLabel("Steps (signed)"), row, 0)
|
||
self.steps_spin = _spin(-100_000_000, 100_000_000, 3200)
|
||
grid.addWidget(self.steps_spin, row, 1)
|
||
move_btn = QPushButton("Move")
|
||
move_btn.clicked.connect(self._on_move)
|
||
grid.addWidget(move_btn, row, 2)
|
||
neg_btn = QPushButton("Move −")
|
||
neg_btn.clicked.connect(lambda: self._on_move(-1))
|
||
grid.addWidget(neg_btn, row, 3)
|
||
pos_btn = QPushButton("Move +")
|
||
pos_btn.clicked.connect(lambda: self._on_move(+1))
|
||
grid.addWidget(pos_btn, row, 4)
|
||
row += 1
|
||
|
||
# Jog / stop
|
||
jogneg = QPushButton("◀ Jog −")
|
||
jogneg.clicked.connect(lambda: self._on_jog(-1))
|
||
grid.addWidget(jogneg, row, 0)
|
||
jogpos = QPushButton("Jog + ▶")
|
||
jogpos.clicked.connect(lambda: self._on_jog(+1))
|
||
grid.addWidget(jogpos, row, 1)
|
||
stop_btn = QPushButton("Stop")
|
||
stop_btn.clicked.connect(self._on_stop)
|
||
grid.addWidget(stop_btn, row, 2)
|
||
self.hard_chk = QCheckBox("hard stop")
|
||
grid.addWidget(self.hard_chk, row, 3, 1, 2)
|
||
row += 1
|
||
|
||
grid.addWidget(_hline(), row, 0, 1, 5); row += 1
|
||
|
||
# Position utilities
|
||
grid.addWidget(QLabel("Set pos"), row, 0)
|
||
self.setpos_spin = _spin(-100_000_000, 100_000_000, 0)
|
||
grid.addWidget(self.setpos_spin, row, 1)
|
||
setpos_btn = QPushButton("Set")
|
||
setpos_btn.clicked.connect(self._on_set_position)
|
||
grid.addWidget(setpos_btn, row, 2)
|
||
zero_btn = QPushButton("Zero")
|
||
zero_btn.clicked.connect(lambda: self._driver.set_position(self.ch, 0))
|
||
grid.addWidget(zero_btn, row, 3)
|
||
drvst_btn = QPushButton("Driver status")
|
||
drvst_btn.setToolTip("GET_DRV_STATUS — SPI read, only works while idle")
|
||
drvst_btn.clicked.connect(lambda: self._driver.get_drv_status(self.ch))
|
||
grid.addWidget(drvst_btn, row, 4)
|
||
|
||
grid.setColumnStretch(4, 1)
|
||
|
||
# ── Commands ──────────────────────────────────────────────────────────────
|
||
|
||
def _on_enable_toggled(self, checked: bool):
|
||
if checked:
|
||
self._driver.enable(self.ch)
|
||
else:
|
||
self._driver.disable(self.ch)
|
||
|
||
def _on_apply_config(self):
|
||
micro = self.micro_combo.currentData()
|
||
self._driver.set_microstep(self.ch, micro)
|
||
self._driver.set_current(self.ch, self.run_spin.value(),
|
||
self.hold_spin.value(), self.ihold_spin.value())
|
||
|
||
def _on_move(self, force_sign: int = 0):
|
||
steps = self.steps_spin.value()
|
||
if force_sign:
|
||
steps = force_sign * abs(steps)
|
||
self._driver.move(self.ch, steps, self.vel_spin.value(), self.accel_spin.value())
|
||
|
||
def _on_jog(self, direction: int):
|
||
self._driver.jog(self.ch, direction * self.vel_spin.value(), self.accel_spin.value())
|
||
|
||
def _on_stop(self):
|
||
self._driver.stop(self.ch, self.hard_chk.isChecked())
|
||
|
||
def _on_set_position(self):
|
||
self._driver.set_position(self.ch, self.setpos_spin.value())
|
||
|
||
# ── Incoming updates ──────────────────────────────────────────────────────
|
||
|
||
def _on_info(self, ch: int, info):
|
||
if ch != self.ch:
|
||
return
|
||
self.state_lbl.setText(proto.STATE_NAMES.get(info.state, "?"))
|
||
self.pos_lbl.setText(f"{info.position:,}")
|
||
self.achieved_lbl.setText(
|
||
f"achieved: run {info.run_ma}/hold {info.hold_ma} mA, {info.microsteps} µsteps")
|
||
self.comms_lbl.setText("comms: OK" if info.comms_ok else "comms: FAIL")
|
||
self.comms_lbl.setStyleSheet("" if info.comms_ok else "color: #c0392b;")
|
||
self._set_fault(info.fault_mask)
|
||
if self.enable_chk.isChecked() != info.enabled:
|
||
self.enable_chk.blockSignals(True)
|
||
self.enable_chk.setChecked(info.enabled)
|
||
self.enable_chk.blockSignals(False)
|
||
|
||
def _on_drv_status(self, ch: int, st):
|
||
if ch != self.ch:
|
||
return
|
||
self._set_fault(st.fault_mask)
|
||
self.drv_lbl.setText(
|
||
f"driver status: CS_ACTUAL={st.cs_actual} SG_RESULT={st.sg_result} "
|
||
f"{'standstill' if st.standstill else 'moving'} raw=0x{st.raw:08X}")
|
||
|
||
def _on_event_pos(self, ch: int, position: int):
|
||
if ch == self.ch:
|
||
self.pos_lbl.setText(f"{position:,}")
|
||
|
||
def _on_fault_event(self, ch: int, mask: int):
|
||
if ch == self.ch:
|
||
self._set_fault(mask)
|
||
|
||
def _set_fault(self, mask: int):
|
||
names = proto.fault_names(mask)
|
||
self.fault_lbl.setText(f"faults: {names}")
|
||
self.fault_lbl.setStyleSheet(
|
||
"color: #c0392b; font-weight: bold;" if mask else "color: #2e7d32;")
|
||
|
||
|
||
# ── Ganged motion panel ───────────────────────────────────────────────────────
|
||
|
||
class GroupPanel(QGroupBox):
|
||
"""Ganged motion — selected axes step in lockstep."""
|
||
|
||
def __init__(self, driver: QtT3RAdapter, log_fn):
|
||
super().__init__("Ganged / synchronised motion — selected axes move in lockstep")
|
||
self._driver = driver
|
||
self._log = log_fn
|
||
self._axis_chks: list[QCheckBox] = []
|
||
self._build()
|
||
|
||
def _build(self):
|
||
grid = QGridLayout(self)
|
||
grid.setVerticalSpacing(4)
|
||
grid.setHorizontalSpacing(8)
|
||
|
||
grid.addWidget(QLabel("Axes:"), 0, 0)
|
||
axis_box = QHBoxLayout()
|
||
for i in range(proto.NUM_CHANNELS):
|
||
chk = QCheckBox(str(i))
|
||
self._axis_chks.append(chk)
|
||
axis_box.addWidget(chk)
|
||
axis_box.addStretch(1)
|
||
holder = QWidget()
|
||
holder.setLayout(axis_box)
|
||
grid.addWidget(holder, 0, 1, 1, 3)
|
||
|
||
hold_btn = QPushButton("Energise selected")
|
||
hold_btn.clicked.connect(self._on_energise)
|
||
grid.addWidget(hold_btn, 0, 4)
|
||
rel_btn = QPushButton("Release all")
|
||
rel_btn.clicked.connect(lambda: self._driver.enable_mask(0))
|
||
grid.addWidget(rel_btn, 0, 5)
|
||
|
||
grid.addWidget(QLabel("Max vel (steps/s)"), 1, 0)
|
||
self.vel_spin = _spin(1, proto.MAX_VELOCITY, 8000)
|
||
grid.addWidget(self.vel_spin, 1, 1)
|
||
grid.addWidget(QLabel("Accel (steps/s²)"), 1, 2)
|
||
self.accel_spin = _spin(0, proto.MAX_ACCEL, 4000)
|
||
grid.addWidget(self.accel_spin, 1, 3)
|
||
grid.addWidget(QLabel("Steps (signed)"), 1, 4)
|
||
self.steps_spin = _spin(-100_000_000, 100_000_000, 3200)
|
||
grid.addWidget(self.steps_spin, 1, 5)
|
||
|
||
move_btn = QPushButton("Move group")
|
||
move_btn.clicked.connect(lambda: self._on_move())
|
||
grid.addWidget(move_btn, 2, 0)
|
||
neg_btn = QPushButton("Move −")
|
||
neg_btn.clicked.connect(lambda: self._on_move(-1))
|
||
grid.addWidget(neg_btn, 2, 1)
|
||
pos_btn = QPushButton("Move +")
|
||
pos_btn.clicked.connect(lambda: self._on_move(+1))
|
||
grid.addWidget(pos_btn, 2, 2)
|
||
jogneg = QPushButton("◀ Jog −")
|
||
jogneg.clicked.connect(lambda: self._on_jog(-1))
|
||
grid.addWidget(jogneg, 2, 3)
|
||
jogpos = QPushButton("Jog + ▶")
|
||
jogpos.clicked.connect(lambda: self._on_jog(+1))
|
||
grid.addWidget(jogpos, 2, 4)
|
||
stop_btn = QPushButton("Stop group")
|
||
stop_btn.clicked.connect(self._on_stop)
|
||
grid.addWidget(stop_btn, 2, 5)
|
||
|
||
self.hard_chk = QCheckBox("hard stop")
|
||
grid.addWidget(self.hard_chk, 3, 5)
|
||
grid.setColumnStretch(3, 1)
|
||
|
||
def _mask(self) -> int:
|
||
return sum((1 << i) for i, chk in enumerate(self._axis_chks) if chk.isChecked())
|
||
|
||
def _require_mask(self) -> int | None:
|
||
mask = self._mask()
|
||
if not mask:
|
||
self._log("No axes selected for ganged motion", "err")
|
||
return None
|
||
return mask
|
||
|
||
def _on_energise(self):
|
||
mask = self._require_mask()
|
||
if mask is not None:
|
||
self._driver.enable_mask(mask)
|
||
|
||
def _on_move(self, force_sign: int = 0):
|
||
mask = self._require_mask()
|
||
if mask is None:
|
||
return
|
||
steps = self.steps_spin.value()
|
||
if force_sign:
|
||
steps = force_sign * abs(steps)
|
||
self._driver.move_group(mask, steps, self.vel_spin.value(), self.accel_spin.value())
|
||
|
||
def _on_jog(self, direction: int):
|
||
mask = self._require_mask()
|
||
if mask is None:
|
||
return
|
||
self._driver.jog_group(mask, direction * self.vel_spin.value(), self.accel_spin.value())
|
||
|
||
def _on_stop(self):
|
||
mask = self._require_mask()
|
||
if mask is None:
|
||
return
|
||
ch = (mask & -mask).bit_length() - 1
|
||
self._driver.stop(ch, self.hard_chk.isChecked())
|
||
|
||
|
||
# ── Rotation panel ────────────────────────────────────────────────────────────
|
||
|
||
class RotationPanel(QGroupBox):
|
||
"""Stage rotation via GR-axis (ch3).
|
||
|
||
Computes the GR-axis step count from the physical gear train:
|
||
Motor → 10T pinion → 30T idler → 125T index gear (stage)
|
||
Ratio = 125/10 = 12.5
|
||
"""
|
||
|
||
def __init__(self, driver: QtT3RAdapter):
|
||
super().__init__(
|
||
f"Stage Rotation (GR-axis ch{T3RDriver.GR_AXIS_CH}) — "
|
||
f"gear: {T3RDriver.GEAR_TEETH_MOTOR}T motor → 30T idler → "
|
||
f"{T3RDriver.GEAR_TEETH_STAGE}T stage "
|
||
f"= {T3RDriver.GEAR_TEETH_STAGE}/{T3RDriver.GEAR_TEETH_MOTOR} ratio"
|
||
)
|
||
self._driver = driver
|
||
self._gr_microsteps = 16 # updated from info_updated signal
|
||
self._build()
|
||
driver.info_updated.connect(self._on_info)
|
||
|
||
def _build(self):
|
||
grid = QGridLayout(self)
|
||
grid.setVerticalSpacing(4)
|
||
grid.setHorizontalSpacing(8)
|
||
|
||
# Microsteps (read-only, tracked from driver)
|
||
grid.addWidget(QLabel("GR-axis µsteps:"), 0, 0)
|
||
self.microstep_lbl = QLabel("16 (live from device)")
|
||
self.microstep_lbl.setStyleSheet("color: gray;")
|
||
grid.addWidget(self.microstep_lbl, 0, 1)
|
||
|
||
# Angle input
|
||
grid.addWidget(QLabel("Rotate by (°):"), 0, 2)
|
||
self.angle_spin = QDoubleSpinBox()
|
||
self.angle_spin.setRange(-360.0, 360.0)
|
||
self.angle_spin.setSingleStep(1.0)
|
||
self.angle_spin.setDecimals(3)
|
||
self.angle_spin.setValue(90.0)
|
||
self.angle_spin.valueChanged.connect(self._update_steps_display)
|
||
grid.addWidget(self.angle_spin, 0, 3)
|
||
|
||
self.steps_lbl = QLabel("= — steps")
|
||
self.steps_lbl.setFont(_mono_font(11))
|
||
grid.addWidget(self.steps_lbl, 0, 4)
|
||
|
||
# Velocity / accel
|
||
grid.addWidget(QLabel("Velocity (steps/s):"), 1, 0)
|
||
self.vel_spin = _spin(1, proto.MAX_VELOCITY, 8000)
|
||
grid.addWidget(self.vel_spin, 1, 1)
|
||
|
||
grid.addWidget(QLabel("Accel (steps/s²):"), 1, 2)
|
||
self.accel_spin = _spin(0, proto.MAX_ACCEL, 4000)
|
||
grid.addWidget(self.accel_spin, 1, 3)
|
||
|
||
# Preset angles for N-angle scans
|
||
grid.addWidget(QLabel("Quick presets:"), 2, 0)
|
||
presets = QHBoxLayout()
|
||
for n in (2, 4, 6, 8, 12):
|
||
btn = QPushButton(f"360/{n}")
|
||
btn.setToolTip(f"{360/n:.3f}° — rotate for {n}-angle scan")
|
||
btn.clicked.connect(lambda _, ang=360.0/n: self.angle_spin.setValue(ang))
|
||
presets.addWidget(btn)
|
||
presets.addStretch(1)
|
||
presets_w = QWidget()
|
||
presets_w.setLayout(presets)
|
||
grid.addWidget(presets_w, 2, 1, 1, 3)
|
||
|
||
# Execute
|
||
rotate_btn = QPushButton("Rotate Stage")
|
||
rotate_btn.setStyleSheet(
|
||
"QPushButton { background: #1565c0; color: white; font-weight: bold; padding: 4px 14px; }"
|
||
"QPushButton:disabled { background: #90a4ae; }")
|
||
rotate_btn.clicked.connect(self._on_rotate)
|
||
grid.addWidget(rotate_btn, 2, 4)
|
||
|
||
grid.setColumnStretch(4, 1)
|
||
self._update_steps_display()
|
||
|
||
def _on_info(self, ch: int, info):
|
||
if ch != T3RDriver.GR_AXIS_CH:
|
||
return
|
||
self._gr_microsteps = info.microsteps
|
||
self.microstep_lbl.setText(f"{info.microsteps} µsteps (from device)")
|
||
self._update_steps_display()
|
||
|
||
def _update_steps_display(self):
|
||
steps = self._driver.steps_for_angle(self.angle_spin.value(), self._gr_microsteps)
|
||
self.steps_lbl.setText(f"= {steps:,} steps")
|
||
|
||
def _on_rotate(self):
|
||
self._driver.rotate_stage(
|
||
self.angle_spin.value(), self._gr_microsteps,
|
||
self.vel_spin.value(), self.accel_spin.value())
|
||
|
||
|
||
# ── Main dialog ───────────────────────────────────────────────────────────────
|
||
|
||
class T3RControlPanel(QDialog):
|
||
"""User-hidable T3R control window.
|
||
|
||
Pass a QtT3RAdapter instance. The panel connects to its signals and forwards
|
||
commands via its API. Connection management (port open/close) is handled
|
||
inside the panel itself.
|
||
"""
|
||
|
||
def __init__(self, driver: QtT3RAdapter, parent=None):
|
||
super().__init__(parent)
|
||
self.setWindowTitle("T3R Stepper Controller")
|
||
self.setWindowFlags(
|
||
Qt.WindowType.Window
|
||
| Qt.WindowType.WindowCloseButtonHint
|
||
| Qt.WindowType.WindowMinimizeButtonHint
|
||
)
|
||
self._driver = driver
|
||
self._build()
|
||
self._connect_driver_signals()
|
||
self._set_controls_enabled(False)
|
||
|
||
# ── Construction ──────────────────────────────────────────────────────────
|
||
|
||
def _build(self):
|
||
outer = QVBoxLayout(self)
|
||
|
||
outer.addLayout(self._build_connection_bar())
|
||
|
||
splitter = QSplitter(Qt.Orientation.Vertical)
|
||
|
||
panels_host = QWidget()
|
||
vbox = QVBoxLayout(panels_host)
|
||
self.group_panel = GroupPanel(self._driver, self._log)
|
||
vbox.addWidget(self.group_panel)
|
||
|
||
grid_holder = QWidget()
|
||
grid = QGridLayout(grid_holder)
|
||
grid.setContentsMargins(0, 0, 0, 0)
|
||
self.channel_panels: list[ChannelPanel] = []
|
||
for ch in range(proto.NUM_CHANNELS):
|
||
p = ChannelPanel(ch, self._driver)
|
||
self.channel_panels.append(p)
|
||
grid.addWidget(p, ch // 2, ch % 2)
|
||
vbox.addWidget(grid_holder)
|
||
|
||
self.rotation_panel = RotationPanel(self._driver)
|
||
vbox.addWidget(self.rotation_panel)
|
||
|
||
scroll = QScrollArea()
|
||
scroll.setWidgetResizable(True)
|
||
scroll.setWidget(panels_host)
|
||
splitter.addWidget(scroll)
|
||
|
||
splitter.addWidget(self._build_log())
|
||
splitter.setStretchFactor(0, 3)
|
||
splitter.setStretchFactor(1, 1)
|
||
outer.addWidget(splitter, 1)
|
||
|
||
self.resize(1100, 950)
|
||
|
||
def _build_connection_bar(self) -> QHBoxLayout:
|
||
bar = QHBoxLayout()
|
||
bar.addWidget(QLabel("Port:"))
|
||
self.port_combo = QComboBox()
|
||
self.port_combo.setMinimumWidth(280)
|
||
bar.addWidget(self.port_combo)
|
||
|
||
refresh_btn = QPushButton("⟳")
|
||
refresh_btn.setMaximumWidth(36)
|
||
refresh_btn.setToolTip("Rescan serial ports")
|
||
refresh_btn.clicked.connect(self._refresh_ports)
|
||
bar.addWidget(refresh_btn)
|
||
|
||
self.connect_btn = QPushButton("Connect")
|
||
self.connect_btn.clicked.connect(self._toggle_connect)
|
||
bar.addWidget(self.connect_btn)
|
||
|
||
self.conn_lbl = QLabel("disconnected")
|
||
bar.addWidget(self.conn_lbl)
|
||
bar.addStretch(1)
|
||
|
||
self.ping_btn = QPushButton("Ping")
|
||
self.ping_btn.setEnabled(False)
|
||
self.ping_btn.clicked.connect(self._driver.ping)
|
||
bar.addWidget(self.ping_btn)
|
||
|
||
self.fw_lbl = QLabel("")
|
||
bar.addWidget(self.fw_lbl)
|
||
|
||
self.stopall_btn = QPushButton("STOP ALL")
|
||
self.stopall_btn.setEnabled(False)
|
||
self.stopall_btn.setStyleSheet(
|
||
"QPushButton { background:#c0392b; color:white; font-weight:bold; padding:4px 14px; }"
|
||
"QPushButton:disabled { background:#e8a39b; }")
|
||
self.stopall_btn.clicked.connect(self._driver.stop_all)
|
||
bar.addWidget(self.stopall_btn)
|
||
return bar
|
||
|
||
def _build_log(self) -> QWidget:
|
||
box = QWidget()
|
||
v = QVBoxLayout(box)
|
||
v.setContentsMargins(0, 0, 0, 0)
|
||
head = QHBoxLayout()
|
||
head.addWidget(QLabel("Log"))
|
||
head.addStretch(1)
|
||
self.rawlog_chk = QCheckBox("Log raw frames")
|
||
head.addWidget(self.rawlog_chk)
|
||
clear_btn = QPushButton("Clear")
|
||
clear_btn.clicked.connect(lambda: self.log_view.clear())
|
||
head.addWidget(clear_btn)
|
||
v.addLayout(head)
|
||
self.log_view = QPlainTextEdit()
|
||
self.log_view.setReadOnly(True)
|
||
self.log_view.setMaximumBlockCount(2000)
|
||
self.log_view.setFont(_mono_font(11))
|
||
v.addWidget(self.log_view)
|
||
return box
|
||
|
||
# ── Driver signal wiring ──────────────────────────────────────────────────
|
||
|
||
def _connect_driver_signals(self):
|
||
self._driver.port_opened.connect(self._on_port_opened)
|
||
self._driver.handshake_ok.connect(self._on_handshake_ok)
|
||
self._driver.disconnected.connect(self._on_disconnected)
|
||
self._driver.ack_received.connect(self._on_ack)
|
||
self._driver.motion_done.connect(self._on_motion_done)
|
||
self._driver.stopped.connect(self._on_stopped)
|
||
self._driver.fault_occurred.connect(self._on_fault)
|
||
self._driver.frame_received.connect(self._on_raw_frame)
|
||
|
||
# ── Connection control ────────────────────────────────────────────────────
|
||
|
||
def _refresh_ports(self):
|
||
current = self.port_combo.currentText()
|
||
self.port_combo.clear()
|
||
for device, label in scored_ports():
|
||
self.port_combo.addItem(label, device)
|
||
if self.port_combo.count() == 0:
|
||
self.port_combo.addItem("(no serial ports found)", None)
|
||
elif current:
|
||
idx = self.port_combo.findText(current, Qt.MatchFlag.MatchStartsWith)
|
||
if idx >= 0:
|
||
self.port_combo.setCurrentIndex(idx)
|
||
|
||
def _toggle_connect(self):
|
||
if self._driver.is_open:
|
||
self._driver.close()
|
||
return
|
||
port = self.port_combo.currentData()
|
||
if not port:
|
||
self._log("No serial port selected", "err")
|
||
return
|
||
try:
|
||
self._driver.open(port)
|
||
except Exception as exc:
|
||
self._log(f"Connect failed: {exc}", "err")
|
||
self.conn_lbl.setText("connect failed")
|
||
|
||
# ── Driver event handlers ─────────────────────────────────────────────────
|
||
|
||
def _on_port_opened(self):
|
||
self.conn_lbl.setText("opening…")
|
||
self.connect_btn.setText("Disconnect")
|
||
self.port_combo.setEnabled(False)
|
||
self._log(f"Port opened, sending PING…", "evt")
|
||
|
||
def _on_handshake_ok(self, proto_ver: int, fw_ver: int, num_ch: int):
|
||
self.conn_lbl.setText("connected")
|
||
self.fw_lbl.setText(
|
||
f"proto v{proto_ver}, fw {fw_ver >> 8}.{fw_ver & 0xFF}, {num_ch} ch")
|
||
self._log(f"PONG: proto v{proto_ver}, fw 0x{fw_ver:04X}, {num_ch} ch", "rx")
|
||
self._set_controls_enabled(True)
|
||
|
||
def _on_disconnected(self, reason: str):
|
||
self.conn_lbl.setText(f"disconnected ({reason})" if reason else "disconnected")
|
||
self.connect_btn.setText("Connect")
|
||
self.port_combo.setEnabled(True)
|
||
self.fw_lbl.setText("")
|
||
self._set_controls_enabled(False)
|
||
if reason:
|
||
self._log(f"Disconnected: {reason}", "err")
|
||
else:
|
||
self._log("Disconnected", "evt")
|
||
|
||
def _on_ack(self, req_cmd: int, status: int):
|
||
name = proto.CMD_NAMES.get(req_cmd, f"0x{req_cmd:02X}")
|
||
status_name = proto.STATUS_NAMES.get(status, f"0x{status:02X}")
|
||
if status != 0:
|
||
self._log(f"ACK {name} → {status_name}", "rx")
|
||
elif self.rawlog_chk.isChecked():
|
||
self._log(f"ACK {name} → OK", "rx")
|
||
|
||
def _on_motion_done(self, ch: int, position: int):
|
||
name = T3RDriver.CHANNEL_NAMES[ch] if ch < len(T3RDriver.CHANNEL_NAMES) else f"ch{ch}"
|
||
self._log(f"MOTION_DONE {name} @ {position:,}", "evt")
|
||
|
||
def _on_stopped(self, ch: int, position: int):
|
||
name = T3RDriver.CHANNEL_NAMES[ch] if ch < len(T3RDriver.CHANNEL_NAMES) else f"ch{ch}"
|
||
self._log(f"STOPPED {name} @ {position:,}", "evt")
|
||
|
||
def _on_fault(self, ch: int, mask: int):
|
||
name = T3RDriver.CHANNEL_NAMES[ch] if ch < len(T3RDriver.CHANNEL_NAMES) else f"ch{ch}"
|
||
self._log(f"FAULT {name}: {proto.fault_names(mask)}", "err")
|
||
|
||
def _on_raw_frame(self, cmd: int, payload: bytes):
|
||
if self.rawlog_chk.isChecked():
|
||
frame = proto.build_frame(cmd, payload)
|
||
self._log(frame.hex(" "), "rx")
|
||
|
||
def _set_controls_enabled(self, on: bool):
|
||
self.ping_btn.setEnabled(on)
|
||
self.stopall_btn.setEnabled(on)
|
||
self.group_panel.setEnabled(on)
|
||
self.rotation_panel.setEnabled(on)
|
||
for p in self.channel_panels:
|
||
p.setEnabled(on)
|
||
|
||
# ── Logging ───────────────────────────────────────────────────────────────
|
||
|
||
def _log(self, msg: str, kind: str = ""):
|
||
prefix = {"tx": "→ ", "rx": "← ", "evt": "● ", "err": "! "}.get(kind, " ")
|
||
self.log_view.appendPlainText(prefix + msg)
|
||
|
||
# ── Lifecycle ─────────────────────────────────────────────────────────────
|
||
|
||
def showEvent(self, event):
|
||
super().showEvent(event)
|
||
if self.port_combo.count() == 0:
|
||
self._refresh_ports()
|
||
|
||
def closeEvent(self, event):
|
||
# Hide instead of destroy so the window can be re-shown.
|
||
event.ignore()
|
||
self.hide()
|