tka/scanengine-3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

precommit

Browse Source
This commit is contained in:
Thomas Ales [MSE]
2026-07-21 20:10:06 -05:00
parent 66e72884fe
commit 1014533626
3 changed files with 1383 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hardware/t3r_driver.py
Executable
+287
View File
@@ -0,0 +1,287 @@
"""T3R Stepper Controller driver for ScanEngine-3.
Qt-based driver that owns the serial connection and an internal reader QThread.
All events arrive as Qt signals; all commands are fire-and-forget writes.
Create in the main (GUI) thread; no additional thread management required.
Gear train (stage rotation via GR-axis, ch3):
Motor → 10T pinion → 30T idler → 125T index gear (stage)
Ratio motor:stage = 125/10 = 12.5
"""
from __future__ import annotations
import threading
import serial
from PyQt6.QtCore import QObject, QThread, QTimer, pyqtSignal
from . import t3r_protocol as proto
class _T3RReader(QThread):
"""Blocking read loop — runs on its own QThread."""
frame = pyqtSignal(int, bytes) # (cmd, payload) for each valid frame
finished_reason = pyqtSignal(str) # "" = clean stop, else I/O error string
def __init__(self, ser: serial.Serial):
super().__init__()
self._ser = ser
self._running = True
self._parser = proto.FrameParser()
def stop(self):
self._running = False
def run(self):
reason = ""
while self._running:
try:
data = self._ser.read(256)
except Exception as exc:
reason = str(exc) or exc.__class__.__name__
break
if data:
for cmd, payload in self._parser.feed(data):
self.frame.emit(cmd, payload)
self.finished_reason.emit(reason)
class T3RDriver(QObject):
"""Qt-based driver for the T3R four-channel stepper controller.
Usage::
driver = T3RDriver()
driver.handshake_ok.connect(lambda pv, fw, nc: print("connected"))
driver.info_updated.connect(on_info)
driver.connect("/dev/ttyUSB0")
driver.move(0, steps=3200, velocity=8000, accel=4000)
"""
CHANNEL_NAMES = ["T-axis (focus)", "Axis 1", "Axis 2", "GR-axis"]
GR_AXIS_CH = 3
# Gear train constants
MOTOR_FULL_STEPS_PER_REV = 200
GEAR_TEETH_MOTOR = 10
GEAR_TEETH_STAGE = 125 # idler is 30T but does not change ratio
# ── Signals ───────────────────────────────────────────────────────────────
port_opened = pyqtSignal() # serial port open; PING sent
handshake_ok = pyqtSignal(int, int, int) # proto_ver, fw_ver, num_channels
disconnected = pyqtSignal(str) # reason ("" = user-initiated)
info_updated = pyqtSignal(int, object) # ch, proto.Info
drv_status_updated = pyqtSignal(int, object) # ch, proto.DrvStatus
position_updated = pyqtSignal(int, int) # ch, position (microsteps)
motion_done = pyqtSignal(int, int) # ch, final_position
stopped = pyqtSignal(int, int) # ch, final_position
fault_occurred = pyqtSignal(int, int) # ch, fault_mask
ack_received = pyqtSignal(int, int) # req_cmd, status (0=OK)
frame_received = pyqtSignal(int, bytes) # raw (cmd, payload) for log
def __init__(self, parent=None):
super().__init__(parent)
self._ser: serial.Serial | None = None
self._reader: _T3RReader | None = None
self._write_lock = threading.Lock()
self._tearing_down = False
self._is_open = False
self._poll_timer = QTimer(self)
self._poll_timer.setInterval(250)
self._poll_timer.timeout.connect(self._poll)
# ── Connection ────────────────────────────────────────────────────────────
@property
def is_open(self) -> bool:
return self._is_open
def connect(self, port: str, baud: int = 115200) -> None:
"""Open the serial port and start the reader. Emits port_opened on success."""
if self._is_open:
self.disconnect()
try:
self._ser = serial.Serial(port, baudrate=baud, timeout=0.05)
except Exception as exc:
raise RuntimeError(f"Cannot open {port}: {exc}") from exc
self._tearing_down = False
self._is_open = True
self._reader = _T3RReader(self._ser)
self._reader.frame.connect(self._on_frame)
self._reader.finished_reason.connect(self._on_reader_finished)
self._reader.start()
self.port_opened.emit()
self.send_frame(proto.ping()) # handshake; polling starts on PONG
def disconnect(self) -> None:
"""Close port and stop polling."""
self._teardown("")
def _on_reader_finished(self, reason: str):
if reason:
self._teardown(reason)
def _teardown(self, reason: str):
if self._tearing_down or not self._is_open:
return
self._tearing_down = True
self._poll_timer.stop()
self._is_open = False
reader, self._reader = self._reader, None
ser, self._ser = self._ser, None
if reader is not None:
reader.stop()
if QThread.currentThread() is not reader:
reader.wait(1000)
if ser is not None:
try:
ser.close()
except Exception:
pass
self.disconnected.emit(reason)
# ── Sending ───────────────────────────────────────────────────────────────
def send_frame(self, frame: bytes) -> bool:
if not self._is_open or self._ser is None:
return False
try:
with self._write_lock:
self._ser.write(frame)
return True
except Exception as exc:
self._teardown(str(exc) or exc.__class__.__name__)
return False
# ── Command API ───────────────────────────────────────────────────────────
def ping(self):
self.send_frame(proto.ping())
def stop_all(self):
self.send_frame(proto.stop_all())
def enable(self, ch: int):
self.send_frame(proto.enable(ch))
def disable(self, ch: int):
self.send_frame(proto.disable(ch))
def enable_mask(self, mask: int):
self.send_frame(proto.enable_mask(mask))
def set_microstep(self, ch: int, microsteps: int):
self.send_frame(proto.set_microstep(ch, microsteps))
def set_current(self, ch: int, run_ma: int, hold_ma: int, ihold_delay: int):
self.send_frame(proto.set_current(ch, run_ma, hold_ma, ihold_delay))
def move(self, ch: int, steps: int, velocity: int, accel: int):
self.send_frame(proto.move(ch, steps, velocity, accel))
def jog(self, ch: int, velocity: int, accel: int):
self.send_frame(proto.jog(ch, velocity, accel))
def stop(self, ch: int, hard: bool = False):
self.send_frame(proto.stop(ch, hard))
def move_group(self, mask: int, steps: int, velocity: int, accel: int):
self.send_frame(proto.move_group(mask, steps, velocity, accel))
def jog_group(self, mask: int, velocity: int, accel: int):
self.send_frame(proto.jog_group(mask, velocity, accel))
def get_info(self, ch: int):
self.send_frame(proto.get_info(ch))
def get_drv_status(self, ch: int):
self.send_frame(proto.get_drv_status(ch))
def get_position(self, ch: int):
self.send_frame(proto.get_position(ch))
def set_position(self, ch: int, position: int):
self.send_frame(proto.set_position(ch, position))
# ── Rotation helpers ──────────────────────────────────────────────────────
def steps_for_angle(self, angle_deg: float, microsteps: int) -> int:
"""Compute GR-axis microsteps needed to rotate the stage by angle_deg."""
gear_ratio = self.GEAR_TEETH_STAGE / self.GEAR_TEETH_MOTOR
steps_per_stage_rev = self.MOTOR_FULL_STEPS_PER_REV * microsteps * gear_ratio
return round(steps_per_stage_rev * angle_deg / 360.0)
def rotate_stage(self, angle_deg: float, microsteps: int,
velocity: int = 8000, accel: int = 4000):
"""Move GR-axis by the number of steps that rotate the stage by angle_deg."""
steps = self.steps_for_angle(angle_deg, microsteps)
self.move(self.GR_AXIS_CH, steps, velocity, accel)
# ── Polling ───────────────────────────────────────────────────────────────
def start_polling(self):
self._poll_timer.start()
def stop_polling(self):
self._poll_timer.stop()
def _poll(self):
if not self._is_open:
return
for ch in range(proto.NUM_CHANNELS):
self.send_frame(proto.get_info(ch))
# ── Frame dispatcher ──────────────────────────────────────────────────────
def _on_frame(self, cmd: int, payload: bytes):
self.frame_received.emit(cmd, payload)
if cmd == proto.RSP_PONG:
p = proto.decode_pong(payload)
if p:
self.handshake_ok.emit(p.proto_ver, p.fw_version, p.num_channels)
self.start_polling()
elif cmd == proto.RSP_ACK:
a = proto.decode_ack(payload)
if a:
self.ack_received.emit(a.req_cmd, a.status)
elif cmd == proto.RSP_INFO:
info = proto.decode_info(payload)
if info and 0 <= info.ch < proto.NUM_CHANNELS:
self.info_updated.emit(info.ch, info)
elif cmd == proto.RSP_DRV_STATUS:
st = proto.decode_drv_status(payload)
if st and 0 <= st.ch < proto.NUM_CHANNELS:
self.drv_status_updated.emit(st.ch, st)
elif cmd == proto.RSP_POSITION:
pos = proto.decode_position(payload)
if pos and 0 <= pos.ch < proto.NUM_CHANNELS:
self.position_updated.emit(pos.ch, pos.position)
elif cmd == proto.EVT_MOTION_DONE:
ev = proto.decode_event_position(payload)
if ev and 0 <= ev.ch < proto.NUM_CHANNELS:
self.motion_done.emit(ev.ch, ev.position)
elif cmd == proto.EVT_STOPPED:
ev = proto.decode_event_position(payload)
if ev and 0 <= ev.ch < proto.NUM_CHANNELS:
self.stopped.emit(ev.ch, ev.position)
elif cmd == proto.EVT_FAULT:
ev = proto.decode_fault(payload)
if ev and 0 <= ev.ch < proto.NUM_CHANNELS:
self.fault_occurred.emit(ev.ch, ev.position)
hardware/t3r_protocol.py
Executable
+374
View File
@@ -0,0 +1,374 @@
"""T3R binary serial protocol — host-side codec.
A faithful Python port of the wire protocol implemented in ``main/protocol.c``.
Pure standard library (no PyQt / pyserial), so it can be unit-tested on its own.
Frame layout (all multi-byte fields little-endian):
+------+------+------+----------------+------+
| SOF | CMD | LEN | PAYLOAD[LEN] | CRC8 |
+------+------+------+----------------+------+
0xA5 1 B 1 B LEN bytes 1 B
CRC8 is CRC-8/SMBUS (poly 0x07, init 0x00) over CMD, LEN and PAYLOAD.
See PROTOCOL.md for the full catalogue.
"""
from __future__ import annotations
import struct
from dataclasses import dataclass
SOF = 0xA5
PROTO_VERSION = 1
# ---- Requests (host -> device) --------------------------------------------
CMD_PING = 0x01
CMD_SET_MICROSTEP = 0x10
CMD_SET_CURRENT = 0x11
CMD_ENABLE = 0x12
CMD_DISABLE = 0x13
CMD_ENABLE_MASK = 0x14
CMD_MOVE = 0x20
CMD_JOG = 0x21
CMD_STOP = 0x22
CMD_STOP_ALL = 0x23
CMD_MOVE_GROUP = 0x24
CMD_JOG_GROUP = 0x25
CMD_GET_INFO = 0x30
CMD_GET_DRV_STATUS = 0x31
CMD_GET_POSITION = 0x32
CMD_SET_POSITION = 0x33
CMD_READ_REG = 0x40
CMD_WRITE_REG = 0x41
# ---- Responses (device -> host) -------------------------------------------
RSP_ACK = 0x81
RSP_PONG = 0x82
RSP_INFO = 0x83
RSP_DRV_STATUS = 0x84
RSP_POSITION = 0x85
RSP_REG = 0x86
# ---- Asynchronous events (device -> host) ---------------------------------
EVT_MOTION_DONE = 0xE0
EVT_STOPPED = 0xE1
EVT_FAULT = 0xE2
# Human-readable names, used by the log/console.
CMD_NAMES = {
CMD_PING: "PING", CMD_SET_MICROSTEP: "SET_MICROSTEP",
CMD_SET_CURRENT: "SET_CURRENT", CMD_ENABLE: "ENABLE", CMD_DISABLE: "DISABLE",
CMD_ENABLE_MASK: "ENABLE_MASK",
CMD_MOVE: "MOVE", CMD_JOG: "JOG", CMD_STOP: "STOP", CMD_STOP_ALL: "STOP_ALL",
CMD_MOVE_GROUP: "MOVE_GROUP", CMD_JOG_GROUP: "JOG_GROUP",
CMD_GET_INFO: "GET_INFO", CMD_GET_DRV_STATUS: "GET_DRV_STATUS",
CMD_GET_POSITION: "GET_POSITION", CMD_SET_POSITION: "SET_POSITION",
CMD_READ_REG: "READ_REG", CMD_WRITE_REG: "WRITE_REG",
RSP_ACK: "ACK", RSP_PONG: "PONG", RSP_INFO: "INFO",
RSP_DRV_STATUS: "DRV_STATUS", RSP_POSITION: "POSITION", RSP_REG: "REG",
EVT_MOTION_DONE: "MOTION_DONE", EVT_STOPPED: "STOPPED", EVT_FAULT: "FAULT",
}
# ---- ACK status codes ------------------------------------------------------
STATUS_NAMES = {
0x00: "OK", 0x02: "bad length", 0x03: "bad channel", 0x04: "bad parameter",
0x05: "busy", 0x06: "SPI comms fault", 0x07: "unknown command",
}
# ---- INFO.state ------------------------------------------------------------
STATE_NAMES = {0: "IDLE", 1: "MOVING", 2: "JOGGING", 3: "STOPPING"}
# ---- Fault mask bits -------------------------------------------------------
FAULT_BITS = [
(1 << 0, "SHORT_GND_A"),
(1 << 1, "SHORT_GND_B"),
(1 << 2, "OVERTEMP"),
(1 << 3, "OVERTEMP_WARN"),
(1 << 4, "OPEN_LOAD_A"),
(1 << 5, "OPEN_LOAD_B"),
]
MICROSTEPS = [1, 2, 4, 8, 16, 32, 64, 128, 256]
MAX_VELOCITY = 200_000 # steps/s (MOTOR_MAX_VELOCITY)
MAX_ACCEL = 2_000_000 # steps/s2 (MOTOR_MAX_ACCEL)
NUM_CHANNELS = 4
def fault_names(mask: int) -> str:
"""Render a fault mask as a comma-separated list, or 'none'."""
names = [name for bit, name in FAULT_BITS if mask & bit]
return ", ".join(names) if names else "none"
# ---------------------------------------------------------------------------
# CRC-8 (poly 0x07, init 0x00) — matches the firmware reference.
# ---------------------------------------------------------------------------
def crc8(data: bytes) -> int:
crc = 0
for byte in data:
crc ^= byte
for _ in range(8):
crc = ((crc << 1) ^ 0x07) & 0xFF if crc & 0x80 else (crc << 1) & 0xFF
return crc
def build_frame(cmd: int, payload: bytes = b"") -> bytes:
body = bytes([cmd, len(payload)]) + payload
return bytes([SOF]) + body + bytes([crc8(body)])
# ---------------------------------------------------------------------------
# Request encoders — each returns a complete frame ready to write.
# ---------------------------------------------------------------------------
def ping() -> bytes:
return build_frame(CMD_PING)
def set_microstep(ch: int, microsteps: int) -> bytes:
return build_frame(CMD_SET_MICROSTEP, struct.pack("<BH", ch, microsteps))
def set_current(ch: int, run_ma: int, hold_ma: int, ihold_delay: int) -> bytes:
return build_frame(CMD_SET_CURRENT,
struct.pack("<BHHB", ch, run_ma, hold_ma, ihold_delay))
def enable(ch: int) -> bytes:
return build_frame(CMD_ENABLE, bytes([ch]))
def disable(ch: int) -> bytes:
return build_frame(CMD_DISABLE, bytes([ch]))
def enable_mask(mask: int) -> bytes:
"""Energise exactly the channels in `mask` (bit i = channel i); 0 = all off."""
return build_frame(CMD_ENABLE_MASK, bytes([mask & 0x0F]))
def move(ch: int, steps: int, velocity: int, accel: int) -> bytes:
return build_frame(CMD_MOVE, struct.pack("<BiII", ch, steps, velocity, accel))
def jog(ch: int, velocity: int, accel: int) -> bytes:
return build_frame(CMD_JOG, struct.pack("<BiI", ch, velocity, accel))
def move_group(mask: int, steps: int, velocity: int, accel: int) -> bytes:
"""Ganged move: every channel in `mask` steps together in lockstep."""
return build_frame(CMD_MOVE_GROUP,
struct.pack("<BiII", mask & 0x0F, steps, velocity, accel))
def jog_group(mask: int, velocity: int, accel: int) -> bytes:
"""Ganged jog: every channel in `mask` runs together (velocity 0 = stop)."""
return build_frame(CMD_JOG_GROUP,
struct.pack("<BiI", mask & 0x0F, velocity, accel))
def stop(ch: int, hard: bool) -> bytes:
return build_frame(CMD_STOP, struct.pack("<BB", ch, 1 if hard else 0))
def stop_all() -> bytes:
return build_frame(CMD_STOP_ALL)
def get_info(ch: int) -> bytes:
return build_frame(CMD_GET_INFO, bytes([ch]))
def get_drv_status(ch: int) -> bytes:
return build_frame(CMD_GET_DRV_STATUS, bytes([ch]))
def get_position(ch: int) -> bytes:
return build_frame(CMD_GET_POSITION, bytes([ch]))
def set_position(ch: int, position: int) -> bytes:
return build_frame(CMD_SET_POSITION, struct.pack("<Bi", ch, position))
def read_reg(ch: int, reg: int) -> bytes:
return build_frame(CMD_READ_REG, struct.pack("<BB", ch, reg))
def write_reg(ch: int, reg: int, value: int) -> bytes:
return build_frame(CMD_WRITE_REG, struct.pack("<BBI", ch, reg, value))
# ---------------------------------------------------------------------------
# Response / event decoders. Each returns a dataclass (or None on bad length).
# ---------------------------------------------------------------------------
@dataclass
class Pong:
proto_ver: int
fw_version: int
num_channels: int
@dataclass
class Ack:
req_cmd: int
status: int
@property
def ok(self) -> bool:
return self.status == 0x00
@dataclass
class Info:
ch: int
state: int
position: int
velocity: int
microsteps: int
run_ma: int
hold_ma: int
enabled: bool
comms_ok: bool
fault_mask: int
@dataclass
class DrvStatus:
ch: int
raw: int
fault_mask: int
@property
def standstill(self) -> bool:
return bool(self.raw & (1 << 31))
@property
def cs_actual(self) -> int:
return (self.raw >> 16) & 0x1F
@property
def sg_result(self) -> int:
return self.raw & 0x3FF
@dataclass
class Position:
ch: int
position: int
@dataclass
class Reg:
ch: int
reg: int
value: int
def decode_pong(p: bytes):
if len(p) < 4:
return None
proto, fw, nch = struct.unpack_from("<BHB", p, 0)
return Pong(proto, fw, nch)
def decode_ack(p: bytes):
if len(p) < 2:
return None
return Ack(p[0], p[1])
def decode_info(p: bytes):
if len(p) < 18:
return None
(ch, state, pos, vel, micro, run_ma, hold_ma, flags, fault) = \
struct.unpack_from("<BBiIHHHBB", p, 0)
return Info(ch, state, pos, vel, micro, run_ma, hold_ma,
bool(flags & 0x01), bool(flags & 0x02), fault)
def decode_drv_status(p: bytes):
if len(p) < 6:
return None
ch, raw, fault = struct.unpack_from("<BIB", p, 0)
return DrvStatus(ch, raw, fault)
def decode_position(p: bytes):
if len(p) < 5:
return None
ch, pos = struct.unpack_from("<Bi", p, 0)
return Position(ch, pos)
def decode_reg(p: bytes):
if len(p) < 6:
return None
ch, reg, value = struct.unpack_from("<BBI", p, 0)
return Reg(ch, reg, value)
def decode_event_position(p: bytes):
"""MOTION_DONE / STOPPED share the (ch, position) layout."""
return decode_position(p)
def decode_fault(p: bytes):
if len(p) < 2:
return None
return Position(p[0], p[1]) # reuse (ch, value); value is the fault mask
# ---------------------------------------------------------------------------
# Incremental frame parser — mirrors the firmware byte-wise state machine.
# ---------------------------------------------------------------------------
class FrameParser:
"""Feed raw bytes, get back a list of (cmd, payload) for each valid frame.
Bad-CRC and oversized frames are dropped silently; the parser resyncs on
the next SOF, exactly like the device-side parser.
"""
MAX_PAYLOAD = 64
_SOF, _CMD, _LEN, _PAYLOAD, _CRC = range(5)
def __init__(self):
self._state = self._SOF
self._cmd = 0
self._len = 0
self._payload = bytearray()
def feed(self, data: bytes):
out = []
for b in data:
if self._state == self._SOF:
if b == SOF:
self._state = self._CMD
elif self._state == self._CMD:
self._cmd = b
self._state = self._LEN
elif self._state == self._LEN:
self._len = b
self._payload.clear()
if b > self.MAX_PAYLOAD:
self._state = self._SOF # too big: resync
elif b == 0:
self._state = self._CRC
else:
self._state = self._PAYLOAD
elif self._state == self._PAYLOAD:
self._payload.append(b)
if len(self._payload) >= self._len:
self._state = self._CRC
elif self._state == self._CRC:
body = bytes([self._cmd, self._len]) + bytes(self._payload)
if crc8(body) == b:
out.append((self._cmd, bytes(self._payload)))
# else bad CRC: drop, resync
self._state = self._SOF
return out
t3r_control_panel.py
Executable
+722
View File
@@ -0,0 +1,722 @@
"""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, QTimer
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 hardware.t3r_driver import T3RDriver
import hardware.t3r_protocol as proto
import serial.tools.list_ports
# ── 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: T3RDriver):
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: T3RDriver, 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: T3RDriver):
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 T3RDriver 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: T3RDriver, 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()
ports = list(serial.tools.list_ports.comports())
def score(p):
text = f"{p.description} {p.manufacturer or ''} {p.product or ''}".lower()
hints = ("esp32", "jtag", "espressif", "usb serial", "cp210", "ch340", "cdc")
return -sum(h in text for h in hints)
ports.sort(key=score)
for p in ports:
self.port_combo.addItem(f"{p.device} — {p.description or p.device}", p.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.disconnect()
return
port = self.port_combo.currentData()
if not port:
self._log("No serial port selected", "err")
return
try:
self._driver.connect(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()