Thomas Ales
44febe34b8
gui/qt_workers.py — one QueueWorker base replaces the per-device command queue + dispatch + signal boilerplate. The loop blocks on the queue instead of waking 10-20x/second forever (test_idle_worker_does_not_spin asserts an idle worker burns ~no CPU). PollingQueueWorker adds self-rescheduling polling: the next poll is queued only after the previous finishes, so a device slower than the interval can't accumulate a backlog (test_polling_never_overlaps_or_backs_up). Helios responsiveness — the concrete bug that motivated the above: a free running 1 s QTimer queued a status poll that took ~2 s, so the queue grew for as long as the panel stayed connected. - helios_laser._query reads until the CR terminator instead of sleeping a fixed 0.05 + 0.2 s per query - one _query_int() helper replaces five copies of parse-with-logging - polling is now driven by the worker; HeliosWindow's QTimer is gone - dropped __del__, which disabled the laser and wrote to the serial port from the garbage collector at an unpredictable time helios_test_app.py — the worker was moveToThread'd but every call site invoked its methods directly, so all serial I/O (including the sleeps) ran on the GUI thread; Query All froze the UI for ~2 s. Calls now go through a queued signal to a pyqtSlot. Also: connect/disconnect cycles leaked a QThread + worker + 9 connections each time; 16 copies of the not-connected guard collapse to _require_connection(); the Query Power button called a method that has never existed (AttributeError popup) and is now disabled and documented in KNOWN_ISSUES. DCBiasImageWidget preallocates its image and uses set_data/set_clim, so the live preview stops rebuilding the array and the whole artist tree per row (O(rows^2) over a scan). bbd20x: connect() now raises when no bays respond instead of reporting success on the wrong port; disconnect() joins with a timeout so a wedged reader can't hang shutdown; one _channel_for() helper replaces four copy-pasted axis mappings; hardcoded travel limits become TRAVEL_MM; the joke error strings are gone. gui/widgets.py adds the shared ConnectionBar / PortSelector / bounded LogConsole / StatusGrid for the test benches to adopt. 65 tests passing. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1006 lines
38 KiB
Python
Executable File
1006 lines
38 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Helios Laser Test Application
|
|
Simple PyQt6 GUI for testing and controlling the Helios laser.
|
|
"""
|
|
|
|
import sys
|
|
import logging
|
|
|
|
from PyQt6.QtWidgets import (
|
|
QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
|
|
QGroupBox, QLabel, QLineEdit, QPushButton, QComboBox, QSpinBox,
|
|
QMessageBox, QTabWidget, QTextEdit
|
|
)
|
|
from PyQt6.QtCore import QThread, pyqtSignal, pyqtSlot, QObject
|
|
from PyQt6.QtGui import QFont
|
|
|
|
from hardware.helios_laser import HeliosLaser, PulseMode
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Status register bit definitions (Tables 8-1, 8-2, 8-3 — Helios manual)
|
|
# Each entry: bit_number -> (severity, description, comment)
|
|
# severity: 'C' = critical error, 'S' = status, 'I' = input error, '' = none
|
|
# ---------------------------------------------------------------------------
|
|
_LER_FLAGS = {
|
|
0: ('C', 'Controller temperature failure (resonator/SHG/q-switch)',
|
|
'Check CCE register for details'),
|
|
1: ('S', 'Trigger input active',
|
|
'High when trigger signal applied or laser in continuous pulsing'),
|
|
2: ('I', 'Command error',
|
|
'Unknown command sent to controller'),
|
|
3: ('C', 'Laser disable pin open (utility connector)',
|
|
'Shuts down pump diodes; reset LER 0 required to restart'),
|
|
4: ('C', 'Internal hardware failure',
|
|
'Contact Coherent'),
|
|
5: ('C', 'Over voltage laser diode',
|
|
'Check for open circuit or voltage spikes'),
|
|
6: ('C', 'Internal hardware failure',
|
|
'Contact Coherent'),
|
|
7: ('C', 'Controller temperature failure at pump diodes',
|
|
'Check LCE register for details'),
|
|
8: ('S', 'Laser start delay (60 s warmup)',
|
|
'Laser cannot be started yet; status error LED flashing'),
|
|
9: ('C', 'Internal hardware failure',
|
|
'Check environment for strong EMI; contact Coherent'),
|
|
10: ('C', 'Internal hardware failure',
|
|
'Check environment for strong EMI; contact Coherent'),
|
|
11: ('C', 'Internal hardware failure',
|
|
'Check environment for strong EMI; contact Coherent'),
|
|
12: ('S', 'Slave controller error (remote input)',
|
|
'Valid only for master controller coupled with a slave'),
|
|
13: ('', 'Laserhead not found',
|
|
'Head not connected / not found; check EMI; ignore for double-electronic slave'),
|
|
14: ('', 'Laserhead I\u00b2C acknowledge error',
|
|
'Check environment for strong EMI; ignore for double-electronic slave'),
|
|
15: ('S', 'Range-Error (not critical)',
|
|
'Input value out of range'),
|
|
}
|
|
|
|
_LCE_FLAGS = {
|
|
0: ('C', 'Pump diode over/under temperature',
|
|
'Limit exceeded (<10\u00b0C or >60\u00b0C); check head cooling'),
|
|
1: ('C', 'Internal hardware failure',
|
|
'Contact Coherent'),
|
|
2: ('C', 'Pump diode temperature out of range',
|
|
'Actual temp >2\u00b0C off setpoint for >1 min'),
|
|
3: ('C', 'Pump diode current critical',
|
|
'Current set too close to current limit'),
|
|
4: ('C', 'Pump diode temperature out of limit',
|
|
'Pump diode temperature is out of limit'),
|
|
5: ('S', 'Door switch open',
|
|
'Close utility connector pin 2 permanently to pin 9 (GND)'),
|
|
7: ('C', 'Pump diode NTC error',
|
|
'Invalid temperature measured or NTC broken'),
|
|
8: ('C', 'Laser diode power stage over temperature',
|
|
'Temp <10\u00b0C or >65\u00b0C at controller; check controller cooling'),
|
|
9: ('C', 'Internal hardware failure',
|
|
'Check environment for strong EMI; contact Coherent'),
|
|
10: ('C', 'Internal hardware failure',
|
|
'Check environment for strong EMI; contact Coherent'),
|
|
11: ('C', 'Internal hardware failure',
|
|
'Check environment for strong EMI; contact Coherent'),
|
|
15: ('S', 'Range-Error (not critical)',
|
|
'Input value out of range'),
|
|
}
|
|
|
|
_CCE_FLAGS = {
|
|
0: ('C', 'Resonator/SHG under/over temperature',
|
|
'Limit exceeded (<10\u00b0C or >60\u00b0C); temperature controller deactivated'),
|
|
1: ('C', 'Resonator/SHG NTC failure',
|
|
'Temperature sensor broken or disconnected'),
|
|
2: ('C', 'Resonator/SHG temperature out of range',
|
|
'Actual temp >2\u00b0C off setpoint for >1 min'),
|
|
3: ('C', 'Q-switch ADC / temperature readout failure',
|
|
'Internal hardware error or no NTC connected'),
|
|
4: ('C', 'Q-switch temperature out of range',
|
|
'Actual temp >2\u00b0C off setpoint for >1 min'),
|
|
5: ('C', 'Q-switch under/over temperature',
|
|
'Limit exceeded (<10\u00b0C or >60\u00b0C); temperature controller deactivated'),
|
|
7: ('C', 'Q-switch NTC failure',
|
|
'Internal hardware error or no NTC connected'),
|
|
8: ('C', 'Internal hardware failure',
|
|
'Contact Coherent'),
|
|
15: ('S', 'Range-Error (not critical)',
|
|
'Input value out of range'),
|
|
}
|
|
|
|
_SEVERITY_LABEL = {'C': '[CRIT]', 'S': '[STAT]', 'I': '[INPT]', '': '[INFO]'}
|
|
|
|
|
|
def _decode_register(flags_dict: dict, value: int) -> list:
|
|
"""Return list of (bit, severity, description, comment) for each set bit."""
|
|
active = []
|
|
for bit, (sev, desc, comment) in flags_dict.items():
|
|
if value & (1 << bit):
|
|
active.append((bit, sev, desc, comment))
|
|
return active
|
|
|
|
|
|
# Configure logging
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class LaserWorker(QObject):
|
|
"""Worker thread for laser operations to prevent UI blocking."""
|
|
|
|
# Signals
|
|
operation_complete = pyqtSignal(bool, str) # success, message
|
|
frequency_updated = pyqtSignal(int)
|
|
current_updated = pyqtSignal(int)
|
|
power_updated = pyqtSignal(float)
|
|
enabled_updated = pyqtSignal(bool)
|
|
serial_updated = pyqtSignal(str, str) # controller_sn, head_sn
|
|
status_registers_updated = pyqtSignal(object, object, object) # ler, lce, cce (int or None)
|
|
remote_enable_updated = pyqtSignal(object) # bool or None
|
|
raw_response_received = pyqtSignal(str, str) # command, response
|
|
|
|
def __init__(self, laser: HeliosLaser):
|
|
super().__init__()
|
|
self.laser = laser
|
|
|
|
@pyqtSlot(str, object)
|
|
def invoke(self, method_name: str, args: tuple):
|
|
"""Run one of this worker's methods on the worker thread.
|
|
|
|
Reached through a queued signal connection, so the serial I/O (and
|
|
its blocking reads) stays off the GUI thread. Calling the methods
|
|
directly, as this app used to, executes them in the caller's thread
|
|
and freezes the UI for the duration.
|
|
"""
|
|
try:
|
|
getattr(self, method_name)(*args)
|
|
except Exception as exc:
|
|
self.operation_complete.emit(False, str(exc))
|
|
|
|
def set_frequency(self, freq: int):
|
|
try:
|
|
success = self.laser.set_frequency_hz(freq)
|
|
msg = f"Frequency set to {freq} Hz" if success else "Failed to set frequency"
|
|
self.operation_complete.emit(success, msg)
|
|
except Exception as e:
|
|
self.operation_complete.emit(False, str(e))
|
|
|
|
def set_current(self, current: int):
|
|
try:
|
|
success = self.laser.set_current_ma(current)
|
|
msg = f"Current set to {current} mA" if success else "Failed to set current"
|
|
self.operation_complete.emit(success, msg)
|
|
except Exception as e:
|
|
self.operation_complete.emit(False, str(e))
|
|
|
|
def set_pulse_mode(self, mode: int):
|
|
try:
|
|
pulse_mode = PulseMode(mode)
|
|
success = self.laser.set_pulse_mode(pulse_mode)
|
|
msg = f"Pulse mode set to {pulse_mode.name}" if success else "Failed to set pulse mode"
|
|
self.operation_complete.emit(success, msg)
|
|
except Exception as e:
|
|
self.operation_complete.emit(False, str(e))
|
|
|
|
def set_laser_enable(self, enable: bool):
|
|
try:
|
|
success = self.laser.set_laser_enable(enable)
|
|
state = "enabled" if enable else "disabled"
|
|
msg = f"Laser {state}" if success else f"Failed to {state} laser"
|
|
self.operation_complete.emit(success, msg)
|
|
except Exception as e:
|
|
self.operation_complete.emit(False, str(e))
|
|
|
|
def query_frequency(self):
|
|
try:
|
|
freq = self.laser.get_frequency_hz()
|
|
if freq is not None:
|
|
self.frequency_updated.emit(freq)
|
|
self.operation_complete.emit(True, f"Frequency: {freq} Hz")
|
|
else:
|
|
self.operation_complete.emit(False, "Failed to query frequency")
|
|
except Exception as e:
|
|
self.operation_complete.emit(False, str(e))
|
|
|
|
def query_current(self):
|
|
try:
|
|
current = self.laser.get_current_ma()
|
|
if current is not None:
|
|
self.current_updated.emit(current)
|
|
self.operation_complete.emit(True, f"Current: {current} mA")
|
|
else:
|
|
self.operation_complete.emit(False, "Failed to query current")
|
|
except Exception as e:
|
|
self.operation_complete.emit(False, str(e))
|
|
|
|
def query_power(self):
|
|
# HeliosLaser has no power query: the driver README advertises
|
|
# get_power_mw(), but no such method exists and the protocol
|
|
# mnemonic for an output-power read is not documented anywhere in
|
|
# this repo. This used to raise AttributeError into a popup.
|
|
# See KNOWN_ISSUES.md — needs the command from the Helios manual.
|
|
self.operation_complete.emit(
|
|
False, "Power query is not implemented (no known protocol command)")
|
|
|
|
def query_enabled(self):
|
|
try:
|
|
enabled = self.laser.is_laser_enabled()
|
|
self.enabled_updated.emit(enabled)
|
|
state = "enabled" if enabled else "disabled"
|
|
self.operation_complete.emit(True, f"Laser is {state}")
|
|
except Exception as e:
|
|
self.operation_complete.emit(False, str(e))
|
|
|
|
def query_serials(self):
|
|
try:
|
|
controller_sn = self.laser.get_controller_serial()
|
|
head_sn = self.laser.get_head_serial()
|
|
|
|
if controller_sn and head_sn:
|
|
self.serial_updated.emit(controller_sn, head_sn)
|
|
msg = f"Controller: {controller_sn}, Head: {head_sn}"
|
|
self.operation_complete.emit(True, msg)
|
|
else:
|
|
self.operation_complete.emit(False, "Failed to query serial numbers")
|
|
except Exception as e:
|
|
self.operation_complete.emit(False, str(e))
|
|
|
|
def query_status_registers(self):
|
|
try:
|
|
ler, lce, cce = self.laser.get_status_registers()
|
|
self.status_registers_updated.emit(ler, lce, cce)
|
|
self.operation_complete.emit(True, f"Status: LER={ler} LCE={lce} CCE={cce}")
|
|
except Exception as e:
|
|
self.operation_complete.emit(False, str(e))
|
|
|
|
def query_remote_enable(self):
|
|
try:
|
|
state = self.laser.get_remote_enable()
|
|
self.remote_enable_updated.emit(state)
|
|
if state is not None:
|
|
self.operation_complete.emit(True, f"Remote enable (LRE): {'ON' if state else 'OFF'}")
|
|
else:
|
|
self.operation_complete.emit(False, "Failed to query remote enable")
|
|
except Exception as e:
|
|
self.operation_complete.emit(False, str(e))
|
|
|
|
def set_remote_enable(self, enable: bool):
|
|
try:
|
|
success = self.laser.set_remote_enable(enable)
|
|
state = "ON" if enable else "OFF"
|
|
msg = f"Remote enable (LRE) set to {state}" if success else "Failed to set remote enable"
|
|
self.operation_complete.emit(success, msg)
|
|
except Exception as e:
|
|
self.operation_complete.emit(False, str(e))
|
|
|
|
def do_reset_faults(self):
|
|
try:
|
|
success = self.laser.reset_faults()
|
|
msg = "Fault reset sequence sent (CCE 0 → LCE 0 → LER 0)" if success else "Failed to send reset sequence"
|
|
self.operation_complete.emit(success, msg)
|
|
except Exception as e:
|
|
self.operation_complete.emit(False, str(e))
|
|
|
|
def do_ler_reset(self):
|
|
try:
|
|
success = self.laser._send_command("LER 0")
|
|
msg = "LER 0 sent" if success else "Failed to send LER 0"
|
|
self.operation_complete.emit(success, msg)
|
|
except Exception as e:
|
|
self.operation_complete.emit(False, str(e))
|
|
|
|
def send_raw(self, cmd: str):
|
|
try:
|
|
response = self.laser.send_raw_command(cmd)
|
|
if response is not None:
|
|
self.raw_response_received.emit(cmd, response)
|
|
self.operation_complete.emit(True, f"TX: {cmd!r} RX: {response!r}")
|
|
else:
|
|
self.operation_complete.emit(False, f"No response for: {cmd!r}")
|
|
except Exception as e:
|
|
self.operation_complete.emit(False, str(e))
|
|
|
|
|
|
class HeliosTestApp(QMainWindow):
|
|
"""Main application window for Helios laser testing."""
|
|
|
|
# Dispatches a worker method name + args across the thread boundary.
|
|
worker_call = pyqtSignal(str, object)
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.laser = HeliosLaser()
|
|
self.worker = None
|
|
self.worker_thread = None
|
|
|
|
self.init_ui()
|
|
self.update_port_list()
|
|
|
|
def init_ui(self):
|
|
"""Initialize the user interface."""
|
|
self.setWindowTitle("Helios Laser Test Application")
|
|
self.setGeometry(100, 100, 900, 700)
|
|
|
|
# Main widget and layout
|
|
main_widget = QWidget()
|
|
self.setCentralWidget(main_widget)
|
|
main_layout = QVBoxLayout(main_widget)
|
|
|
|
# Connection group
|
|
connection_group = self.create_connection_group()
|
|
main_layout.addWidget(connection_group)
|
|
|
|
# Tabs for control and monitoring
|
|
tabs = QTabWidget()
|
|
tabs.addTab(self.create_control_tab(), "Control")
|
|
tabs.addTab(self.create_monitor_tab(), "Monitor")
|
|
tabs.addTab(self.create_terminal_tab(), "Terminal")
|
|
main_layout.addWidget(tabs)
|
|
|
|
# Status bar
|
|
self.statusBar().showMessage("Disconnected")
|
|
|
|
def create_connection_group(self) -> QGroupBox:
|
|
"""Create the connection control group."""
|
|
group = QGroupBox("Connection")
|
|
layout = QHBoxLayout()
|
|
|
|
# Port selection
|
|
layout.addWidget(QLabel("Serial Port:"))
|
|
self.combo_port = QComboBox()
|
|
self.combo_port.setMinimumWidth(150)
|
|
layout.addWidget(self.combo_port)
|
|
|
|
# Refresh ports button
|
|
btn_refresh = QPushButton("Refresh Ports")
|
|
btn_refresh.clicked.connect(self.update_port_list)
|
|
layout.addWidget(btn_refresh)
|
|
|
|
# Connect button
|
|
self.btn_connect = QPushButton("Connect")
|
|
self.btn_connect.clicked.connect(self.toggle_connection)
|
|
self.btn_connect.setMinimumWidth(100)
|
|
layout.addWidget(self.btn_connect)
|
|
|
|
# Connection status
|
|
self.lbl_status = QLabel("Status: Disconnected")
|
|
font = self.lbl_status.font()
|
|
font.setBold(True)
|
|
self.lbl_status.setFont(font)
|
|
layout.addWidget(self.lbl_status)
|
|
|
|
layout.addStretch()
|
|
group.setLayout(layout)
|
|
return group
|
|
|
|
def create_control_tab(self) -> QWidget:
|
|
"""Create the control tab."""
|
|
widget = QWidget()
|
|
layout = QVBoxLayout(widget)
|
|
|
|
# Frequency control
|
|
freq_group = QGroupBox("Frequency Control")
|
|
freq_layout = QHBoxLayout()
|
|
freq_layout.addWidget(QLabel("Frequency (Hz):"))
|
|
self.spin_frequency = QSpinBox()
|
|
self.spin_frequency.setRange(16700, 125000)
|
|
self.spin_frequency.setValue(50000)
|
|
self.spin_frequency.setSingleStep(1000)
|
|
freq_layout.addWidget(self.spin_frequency)
|
|
|
|
btn_set_freq = QPushButton("Set Frequency")
|
|
btn_set_freq.clicked.connect(self.on_set_frequency)
|
|
btn_set_freq.setMaximumWidth(150)
|
|
freq_layout.addWidget(btn_set_freq)
|
|
|
|
btn_get_freq = QPushButton("Query")
|
|
btn_get_freq.clicked.connect(self.on_query_frequency)
|
|
btn_get_freq.setMaximumWidth(100)
|
|
freq_layout.addWidget(btn_get_freq)
|
|
freq_layout.addStretch()
|
|
freq_group.setLayout(freq_layout)
|
|
layout.addWidget(freq_group)
|
|
|
|
# Current control
|
|
current_group = QGroupBox("Current Control")
|
|
current_layout = QHBoxLayout()
|
|
current_layout.addWidget(QLabel("Current (mA):"))
|
|
self.spin_current = QSpinBox()
|
|
self.spin_current.setRange(0, 2000)
|
|
self.spin_current.setValue(1000)
|
|
self.spin_current.setSingleStep(100)
|
|
current_layout.addWidget(self.spin_current)
|
|
|
|
btn_set_current = QPushButton("Set Current")
|
|
btn_set_current.clicked.connect(self.on_set_current)
|
|
btn_set_current.setMaximumWidth(150)
|
|
current_layout.addWidget(btn_set_current)
|
|
|
|
btn_get_current = QPushButton("Query")
|
|
btn_get_current.clicked.connect(self.on_query_current)
|
|
btn_get_current.setMaximumWidth(100)
|
|
current_layout.addWidget(btn_get_current)
|
|
current_layout.addStretch()
|
|
current_group.setLayout(current_layout)
|
|
layout.addWidget(current_group)
|
|
|
|
# Pulse mode control
|
|
mode_group = QGroupBox("Pulse Mode Control")
|
|
mode_layout = QHBoxLayout()
|
|
mode_layout.addWidget(QLabel("Pulse Mode:"))
|
|
self.combo_mode = QComboBox()
|
|
self.combo_mode.addItem("Single Pulse", PulseMode.SINGLE_PULSE.value)
|
|
self.combo_mode.addItem("Continuous Gating", PulseMode.CONTINUOUS_GATING.value)
|
|
self.combo_mode.addItem("Continuous Pulsing", PulseMode.CONTINUOUS_PULSING.value)
|
|
mode_layout.addWidget(self.combo_mode)
|
|
|
|
btn_set_mode = QPushButton("Set Mode")
|
|
btn_set_mode.clicked.connect(self.on_set_mode)
|
|
btn_set_mode.setMaximumWidth(150)
|
|
mode_layout.addWidget(btn_set_mode)
|
|
mode_layout.addStretch()
|
|
mode_group.setLayout(mode_layout)
|
|
layout.addWidget(mode_group)
|
|
|
|
# Laser enable/disable
|
|
enable_group = QGroupBox("Laser Control")
|
|
enable_layout = QHBoxLayout()
|
|
|
|
self.btn_enable = QPushButton("Enable Laser")
|
|
self.btn_enable.setStyleSheet("background-color: lightgreen")
|
|
self.btn_enable.clicked.connect(self.on_enable_laser)
|
|
self.btn_enable.setMaximumWidth(150)
|
|
enable_layout.addWidget(self.btn_enable)
|
|
|
|
self.btn_disable = QPushButton("Disable Laser")
|
|
self.btn_disable.setStyleSheet("background-color: lightcoral")
|
|
self.btn_disable.clicked.connect(self.on_disable_laser)
|
|
self.btn_disable.setMaximumWidth(150)
|
|
enable_layout.addWidget(self.btn_disable)
|
|
|
|
btn_check_enabled = QPushButton("Query Status")
|
|
btn_check_enabled.clicked.connect(self.on_query_enabled)
|
|
btn_check_enabled.setMaximumWidth(150)
|
|
enable_layout.addWidget(btn_check_enabled)
|
|
|
|
self.lbl_enabled = QLabel("Status: Unknown")
|
|
font = self.lbl_enabled.font()
|
|
font.setBold(True)
|
|
self.lbl_enabled.setFont(font)
|
|
enable_layout.addWidget(self.lbl_enabled)
|
|
enable_layout.addStretch()
|
|
enable_group.setLayout(enable_layout)
|
|
layout.addWidget(enable_group)
|
|
|
|
# Status & Safety
|
|
safety_group = QGroupBox("Status && Safety")
|
|
safety_layout = QVBoxLayout()
|
|
|
|
# Row 1: status registers + query/reset buttons
|
|
reg_row = QHBoxLayout()
|
|
|
|
btn_query_status = QPushButton("Query Status")
|
|
btn_query_status.clicked.connect(self.on_query_status)
|
|
btn_query_status.setMaximumWidth(130)
|
|
reg_row.addWidget(btn_query_status)
|
|
|
|
btn_reset_faults = QPushButton("Reset Faults")
|
|
btn_reset_faults.setStyleSheet("background-color: #FFD700")
|
|
btn_reset_faults.clicked.connect(self.on_reset_faults)
|
|
btn_reset_faults.setMaximumWidth(130)
|
|
reg_row.addWidget(btn_reset_faults)
|
|
|
|
btn_ler_reset = QPushButton("LER 0 Reset")
|
|
btn_ler_reset.setStyleSheet("background-color: #FFA500")
|
|
btn_ler_reset.setToolTip("Send LER 0 only (clears controller error register)")
|
|
btn_ler_reset.clicked.connect(self.on_ler_reset)
|
|
btn_ler_reset.setMaximumWidth(130)
|
|
reg_row.addWidget(btn_ler_reset)
|
|
|
|
reg_row.addSpacing(20)
|
|
reg_row.addWidget(QLabel("LER:"))
|
|
self.lbl_ler = QLabel("—")
|
|
bold_font = self.lbl_ler.font()
|
|
bold_font.setBold(True)
|
|
self.lbl_ler.setFont(bold_font)
|
|
reg_row.addWidget(self.lbl_ler)
|
|
|
|
reg_row.addWidget(QLabel("LCE:"))
|
|
self.lbl_lce = QLabel("—")
|
|
self.lbl_lce.setFont(bold_font)
|
|
reg_row.addWidget(self.lbl_lce)
|
|
|
|
reg_row.addWidget(QLabel("CCE:"))
|
|
self.lbl_cce = QLabel("—")
|
|
self.lbl_cce.setFont(bold_font)
|
|
reg_row.addWidget(self.lbl_cce)
|
|
|
|
reg_row.addSpacing(20)
|
|
reg_row.addWidget(QLabel("Interlock:"))
|
|
self.lbl_interlock = QLabel("UNKNOWN")
|
|
self.lbl_interlock.setFont(bold_font)
|
|
self.lbl_interlock.setMinimumWidth(80)
|
|
reg_row.addWidget(self.lbl_interlock)
|
|
|
|
reg_row.addStretch()
|
|
safety_layout.addLayout(reg_row)
|
|
|
|
# Row 2: remote enable (LRE / pin 8)
|
|
lre_row = QHBoxLayout()
|
|
|
|
btn_lre_on = QPushButton("Remote Enable ON")
|
|
btn_lre_on.setStyleSheet("background-color: lightgreen")
|
|
btn_lre_on.clicked.connect(lambda: self.on_set_remote_enable(True))
|
|
btn_lre_on.setMaximumWidth(160)
|
|
lre_row.addWidget(btn_lre_on)
|
|
|
|
btn_lre_off = QPushButton("Remote Enable OFF")
|
|
btn_lre_off.setStyleSheet("background-color: lightcoral")
|
|
btn_lre_off.clicked.connect(lambda: self.on_set_remote_enable(False))
|
|
btn_lre_off.setMaximumWidth(160)
|
|
lre_row.addWidget(btn_lre_off)
|
|
|
|
btn_lre_query = QPushButton("Query LRE")
|
|
btn_lre_query.clicked.connect(self.on_query_remote_enable)
|
|
btn_lre_query.setMaximumWidth(120)
|
|
lre_row.addWidget(btn_lre_query)
|
|
|
|
lre_row.addSpacing(20)
|
|
lre_row.addWidget(QLabel("Enable Pin (LRE):"))
|
|
self.lbl_lre = QLabel("UNKNOWN")
|
|
self.lbl_lre.setFont(bold_font)
|
|
self.lbl_lre.setMinimumWidth(80)
|
|
lre_row.addWidget(self.lbl_lre)
|
|
|
|
lre_row.addStretch()
|
|
safety_layout.addLayout(lre_row)
|
|
|
|
# Row 3: decoded register flags
|
|
self.text_register_decode = QTextEdit()
|
|
self.text_register_decode.setReadOnly(True)
|
|
self.text_register_decode.setMinimumHeight(110)
|
|
self.text_register_decode.setMaximumHeight(160)
|
|
self.text_register_decode.setPlaceholderText(
|
|
"Register flags will appear here after querying status…"
|
|
)
|
|
mono_font = QFont("Monospace")
|
|
mono_font.setStyleHint(QFont.StyleHint.Monospace)
|
|
self.text_register_decode.setFont(mono_font)
|
|
safety_layout.addWidget(self.text_register_decode)
|
|
|
|
safety_group.setLayout(safety_layout)
|
|
layout.addWidget(safety_group)
|
|
|
|
layout.addStretch()
|
|
return widget
|
|
|
|
def create_monitor_tab(self) -> QWidget:
|
|
"""Create the monitoring/information tab."""
|
|
widget = QWidget()
|
|
layout = QVBoxLayout(widget)
|
|
|
|
# System information
|
|
info_group = QGroupBox("System Information")
|
|
info_layout = QHBoxLayout()
|
|
|
|
btn_query_all = QPushButton("Query All")
|
|
btn_query_all.clicked.connect(self.on_query_all)
|
|
info_layout.addWidget(btn_query_all)
|
|
|
|
info_layout.addWidget(QLabel("Controller SN:"))
|
|
self.lbl_controller_sn = QLineEdit()
|
|
self.lbl_controller_sn.setReadOnly(True)
|
|
info_layout.addWidget(self.lbl_controller_sn)
|
|
|
|
info_layout.addWidget(QLabel("Head SN:"))
|
|
self.lbl_head_sn = QLineEdit()
|
|
self.lbl_head_sn.setReadOnly(True)
|
|
info_layout.addWidget(self.lbl_head_sn)
|
|
info_group.setLayout(info_layout)
|
|
layout.addWidget(info_group)
|
|
|
|
# Power monitoring
|
|
power_group = QGroupBox("Power Monitoring")
|
|
power_layout = QHBoxLayout()
|
|
|
|
btn_get_power = QPushButton("Query Power")
|
|
btn_get_power.setEnabled(False)
|
|
btn_get_power.setToolTip(
|
|
"Not implemented: no documented Helios protocol command for "
|
|
"output power (see KNOWN_ISSUES.md).")
|
|
btn_get_power.clicked.connect(self.on_query_power)
|
|
power_layout.addWidget(btn_get_power)
|
|
|
|
power_layout.addWidget(QLabel("Output Power:"))
|
|
self.lbl_power = QLineEdit()
|
|
self.lbl_power.setReadOnly(True)
|
|
self.lbl_power.setMaximumWidth(150)
|
|
power_layout.addWidget(self.lbl_power)
|
|
|
|
power_layout.addWidget(QLabel("mW"))
|
|
power_layout.addStretch()
|
|
power_group.setLayout(power_layout)
|
|
layout.addWidget(power_group)
|
|
|
|
# Log/Message display
|
|
log_group = QGroupBox("Messages")
|
|
log_layout = QVBoxLayout()
|
|
self.text_log = QTextEdit()
|
|
self.text_log.setReadOnly(True)
|
|
self.text_log.setMaximumHeight(300)
|
|
log_layout.addWidget(self.text_log)
|
|
log_group.setLayout(log_layout)
|
|
layout.addWidget(log_group)
|
|
|
|
layout.addStretch()
|
|
return widget
|
|
|
|
def create_terminal_tab(self) -> QWidget:
|
|
"""Create the raw command terminal tab."""
|
|
widget = QWidget()
|
|
layout = QVBoxLayout(widget)
|
|
|
|
# Input row
|
|
cmd_group = QGroupBox("Raw Command")
|
|
cmd_layout = QHBoxLayout()
|
|
|
|
cmd_layout.addWidget(QLabel("Command:"))
|
|
self.le_raw_cmd = QLineEdit()
|
|
self.le_raw_cmd.setPlaceholderText("e.g. LDF or LDS 1000 or LER 0")
|
|
self.le_raw_cmd.returnPressed.connect(self.on_send_raw)
|
|
cmd_layout.addWidget(self.le_raw_cmd)
|
|
|
|
btn_send_raw = QPushButton("Send")
|
|
btn_send_raw.setMaximumWidth(80)
|
|
btn_send_raw.clicked.connect(self.on_send_raw)
|
|
cmd_layout.addWidget(btn_send_raw)
|
|
|
|
btn_clear_terminal = QPushButton("Clear")
|
|
btn_clear_terminal.setMaximumWidth(70)
|
|
btn_clear_terminal.clicked.connect(lambda: self.text_terminal.clear())
|
|
cmd_layout.addWidget(btn_clear_terminal)
|
|
|
|
cmd_group.setLayout(cmd_layout)
|
|
layout.addWidget(cmd_group)
|
|
|
|
# Terminal output
|
|
term_group = QGroupBox("Response Log")
|
|
term_layout = QVBoxLayout()
|
|
self.text_terminal = QTextEdit()
|
|
self.text_terminal.setReadOnly(True)
|
|
mono_font = QFont("Monospace")
|
|
mono_font.setStyleHint(QFont.StyleHint.Monospace)
|
|
self.text_terminal.setFont(mono_font)
|
|
self.text_terminal.setPlaceholderText(
|
|
"TX/RX pairs will appear here.\n"
|
|
"Commands are sent as-is with a trailing CR.\n"
|
|
"No ? suffix needed — send the command name to query (e.g. LDF)."
|
|
)
|
|
term_layout.addWidget(self.text_terminal)
|
|
term_group.setLayout(term_layout)
|
|
layout.addWidget(term_group)
|
|
|
|
return widget
|
|
|
|
def update_port_list(self):
|
|
"""Update the available serial ports."""
|
|
self.combo_port.clear()
|
|
try:
|
|
ports = HeliosLaser.list_available_ports()
|
|
if ports:
|
|
self.combo_port.addItems(ports)
|
|
else:
|
|
self.combo_port.addItem("No ports found")
|
|
except Exception as e:
|
|
self.combo_port.addItem(f"Error: {e}")
|
|
logger.error(f"Error listing ports: {e}")
|
|
|
|
def toggle_connection(self):
|
|
"""Toggle connection to the laser."""
|
|
if self.laser.is_connected:
|
|
self.disconnect_laser()
|
|
else:
|
|
self.connect_laser()
|
|
|
|
def connect_laser(self):
|
|
"""Connect to the laser."""
|
|
port = self.combo_port.currentText()
|
|
if not port or "No ports" in port or "Error" in port:
|
|
QMessageBox.warning(self, "Connection Error", "No valid port selected")
|
|
return
|
|
|
|
if self.laser.connect(port):
|
|
self.lbl_status.setText(f"Status: Connected to {port}")
|
|
self.lbl_status.setStyleSheet("color: green")
|
|
self.btn_connect.setText("Disconnect")
|
|
self.combo_port.setEnabled(False)
|
|
|
|
# Create worker for async operations
|
|
self.worker = LaserWorker(self.laser)
|
|
self.worker_thread = QThread()
|
|
self.worker.moveToThread(self.worker_thread)
|
|
# Queued (cross-thread) connection: the worker's methods run on
|
|
# the worker thread, not on whichever thread emits.
|
|
self.worker_call.connect(self.worker.invoke)
|
|
self.worker.operation_complete.connect(self.on_operation_complete)
|
|
self.worker.frequency_updated.connect(self.on_frequency_updated)
|
|
self.worker.current_updated.connect(self.on_current_updated)
|
|
self.worker.power_updated.connect(self.on_power_updated)
|
|
self.worker.enabled_updated.connect(self.on_enabled_updated)
|
|
self.worker.serial_updated.connect(self.on_serial_updated)
|
|
self.worker.status_registers_updated.connect(self.on_status_registers_updated)
|
|
self.worker.remote_enable_updated.connect(self.on_remote_enable_updated)
|
|
self.worker.raw_response_received.connect(self.on_raw_response)
|
|
self.worker_thread.start()
|
|
|
|
self.statusBar().showMessage(f"Connected to {port}")
|
|
logger.info(f"Connected to {port}")
|
|
else:
|
|
QMessageBox.critical(self, "Connection Error", f"Failed to connect to {port}")
|
|
|
|
def _require_connection(self) -> bool:
|
|
"""Warn and return False when no laser is connected."""
|
|
if self.laser.is_connected and self.worker is not None:
|
|
return True
|
|
QMessageBox.warning(self, "Error", "Not connected to laser")
|
|
return False
|
|
|
|
def _call_worker(self, method_name: str, *args):
|
|
"""Run a worker method on the worker thread via a queued signal."""
|
|
if self.worker is not None:
|
|
self.worker_call.emit(method_name, args)
|
|
|
|
def disconnect_laser(self):
|
|
"""Disconnect from the laser."""
|
|
if self.worker_thread:
|
|
self.worker_thread.quit()
|
|
self.worker_thread.wait(2000)
|
|
# Drop both references: a reconnect used to leak the previous
|
|
# QThread, worker, and all nine signal connections.
|
|
self.worker = None
|
|
self.worker_thread = None
|
|
|
|
self.laser.disconnect()
|
|
self.lbl_status.setText("Status: Disconnected")
|
|
self.lbl_status.setStyleSheet("color: red")
|
|
self.btn_connect.setText("Connect")
|
|
self.combo_port.setEnabled(True)
|
|
self.statusBar().showMessage("Disconnected")
|
|
logger.info("Disconnected from laser")
|
|
|
|
def on_set_frequency(self):
|
|
"""Set the laser frequency."""
|
|
if not self._require_connection():
|
|
return
|
|
|
|
freq = self.spin_frequency.value()
|
|
self._call_worker("set_frequency", freq)
|
|
|
|
def on_query_frequency(self):
|
|
"""Query the laser frequency."""
|
|
if not self._require_connection():
|
|
return
|
|
|
|
self._call_worker("query_frequency")
|
|
|
|
def on_set_current(self):
|
|
"""Set the laser current."""
|
|
if not self._require_connection():
|
|
return
|
|
|
|
current = self.spin_current.value()
|
|
self._call_worker("set_current", current)
|
|
|
|
def on_query_current(self):
|
|
"""Query the laser current."""
|
|
if not self._require_connection():
|
|
return
|
|
|
|
self._call_worker("query_current")
|
|
|
|
def on_set_mode(self):
|
|
"""Set the laser pulse mode."""
|
|
if not self._require_connection():
|
|
return
|
|
|
|
mode = self.combo_mode.currentData()
|
|
self._call_worker("set_pulse_mode", mode)
|
|
|
|
def on_enable_laser(self):
|
|
"""Enable the laser."""
|
|
if not self._require_connection():
|
|
return
|
|
|
|
self._call_worker("set_laser_enable", True)
|
|
|
|
def on_disable_laser(self):
|
|
"""Disable the laser."""
|
|
if not self._require_connection():
|
|
return
|
|
|
|
self._call_worker("set_laser_enable", False)
|
|
|
|
def on_query_enabled(self):
|
|
"""Query if laser is enabled."""
|
|
if not self._require_connection():
|
|
return
|
|
|
|
self._call_worker("query_enabled")
|
|
|
|
def on_query_power(self):
|
|
"""Query the laser output power."""
|
|
if not self._require_connection():
|
|
return
|
|
|
|
self._call_worker("query_power")
|
|
|
|
def on_query_all(self):
|
|
"""Query all laser parameters."""
|
|
if not self._require_connection():
|
|
return
|
|
|
|
self._call_worker("query_serials")
|
|
self._call_worker("query_frequency")
|
|
self._call_worker("query_current")
|
|
self._call_worker("query_power")
|
|
self._call_worker("query_enabled")
|
|
self._call_worker("query_status_registers")
|
|
self._call_worker("query_remote_enable")
|
|
|
|
def on_query_status(self):
|
|
"""Query LER/LCE/CCE status registers."""
|
|
if not self._require_connection():
|
|
return
|
|
self._call_worker("query_status_registers")
|
|
|
|
def on_reset_faults(self):
|
|
"""Send the fault reset sequence."""
|
|
if not self._require_connection():
|
|
return
|
|
self._call_worker("do_reset_faults")
|
|
|
|
def on_query_remote_enable(self):
|
|
"""Query the remote enable (LRE) state."""
|
|
if not self._require_connection():
|
|
return
|
|
self._call_worker("query_remote_enable")
|
|
|
|
def on_set_remote_enable(self, enable: bool):
|
|
"""Set the remote enable (LRE) state."""
|
|
if not self._require_connection():
|
|
return
|
|
self._call_worker("set_remote_enable", enable)
|
|
|
|
def on_ler_reset(self):
|
|
"""Send LER 0 only."""
|
|
if not self._require_connection():
|
|
return
|
|
self._call_worker("do_ler_reset")
|
|
|
|
def on_send_raw(self):
|
|
"""Send the raw command from the terminal input."""
|
|
if not self._require_connection():
|
|
return
|
|
cmd = self.le_raw_cmd.text().strip()
|
|
if not cmd:
|
|
return
|
|
self._call_worker("send_raw", cmd)
|
|
|
|
def on_raw_response(self, cmd: str, response: str):
|
|
"""Display raw TX/RX pair in the terminal log."""
|
|
self.text_terminal.append(f"TX: {cmd}")
|
|
self.text_terminal.append(f"RX: {response if response else '<no response>'}")
|
|
self.text_terminal.append("")
|
|
self.text_terminal.verticalScrollBar().setValue(
|
|
self.text_terminal.verticalScrollBar().maximum()
|
|
)
|
|
|
|
def on_operation_complete(self, success: bool, message: str):
|
|
"""Handle operation completion."""
|
|
self.log_message(message)
|
|
if not success:
|
|
QMessageBox.warning(self, "Operation Failed", message)
|
|
|
|
def on_frequency_updated(self, freq: int):
|
|
"""Update frequency display."""
|
|
self.spin_frequency.setValue(freq)
|
|
|
|
def on_current_updated(self, current: int):
|
|
"""Update current display."""
|
|
self.spin_current.setValue(current)
|
|
|
|
def on_power_updated(self, power: float):
|
|
"""Update power display."""
|
|
self.lbl_power.setText(f"{power:.2f}")
|
|
|
|
def on_enabled_updated(self, enabled: bool):
|
|
"""Update enabled status display."""
|
|
state = "Enabled" if enabled else "Disabled"
|
|
color = "green" if enabled else "red"
|
|
self.lbl_enabled.setText(f"Status: {state}")
|
|
self.lbl_enabled.setStyleSheet(f"color: {color}")
|
|
|
|
def on_serial_updated(self, controller_sn: str, head_sn: str):
|
|
"""Update serial number displays."""
|
|
self.lbl_controller_sn.setText(controller_sn)
|
|
self.lbl_head_sn.setText(head_sn)
|
|
|
|
def on_status_registers_updated(self, ler, lce, cce):
|
|
"""Update status register displays, interlock indicator, and decoded flags."""
|
|
def _fmt(val):
|
|
return str(val) if val is not None else "ERR"
|
|
|
|
self.lbl_ler.setText(_fmt(ler))
|
|
self.lbl_lce.setText(_fmt(lce))
|
|
self.lbl_cce.setText(_fmt(cce))
|
|
|
|
# Interlock fault: any non-zero register value indicates an active fault
|
|
if any(v is None for v in (ler, lce, cce)):
|
|
self.lbl_interlock.setText("UNKNOWN")
|
|
self.lbl_interlock.setStyleSheet("color: gray")
|
|
elif any(v != 0 for v in (ler, lce, cce)):
|
|
self.lbl_interlock.setText("FAULT")
|
|
self.lbl_interlock.setStyleSheet("color: red; font-weight: bold")
|
|
else:
|
|
self.lbl_interlock.setText("OK")
|
|
self.lbl_interlock.setStyleSheet("color: green; font-weight: bold")
|
|
|
|
# Decode and display individual flags
|
|
lines = []
|
|
for reg_name, value, flags_dict in (
|
|
("LER", ler, _LER_FLAGS),
|
|
("LCE", lce, _LCE_FLAGS),
|
|
("CCE", cce, _CCE_FLAGS),
|
|
):
|
|
if value is None:
|
|
lines.append(f"{reg_name}: <read error>")
|
|
continue
|
|
active = _decode_register(flags_dict, value)
|
|
if not active:
|
|
lines.append(f"{reg_name} (raw={value}): OK — no flags set")
|
|
else:
|
|
lines.append(f"{reg_name} (raw={value}):")
|
|
for bit, sev, desc, comment in active:
|
|
label = _SEVERITY_LABEL.get(sev, '[ ]')
|
|
lines.append(f" {label} bit {bit:2d} ({1 << bit:>5}): {desc}")
|
|
lines.append(f" → {comment}")
|
|
self.text_register_decode.setPlainText("\n".join(lines))
|
|
|
|
def on_remote_enable_updated(self, state):
|
|
"""Update remote enable (LRE) indicator."""
|
|
if state is None:
|
|
self.lbl_lre.setText("UNKNOWN")
|
|
self.lbl_lre.setStyleSheet("color: gray")
|
|
elif state:
|
|
self.lbl_lre.setText("ACTIVE")
|
|
self.lbl_lre.setStyleSheet("color: green; font-weight: bold")
|
|
else:
|
|
self.lbl_lre.setText("INACTIVE")
|
|
self.lbl_lre.setStyleSheet("color: red; font-weight: bold")
|
|
|
|
def log_message(self, message: str):
|
|
"""Add message to log."""
|
|
self.text_log.append(message)
|
|
# Auto-scroll to bottom
|
|
self.text_log.verticalScrollBar().setValue(
|
|
self.text_log.verticalScrollBar().maximum()
|
|
)
|
|
|
|
def closeEvent(self, event):
|
|
"""Handle application close."""
|
|
if self.laser.is_connected:
|
|
self.disconnect_laser()
|
|
if self.worker_thread and self.worker_thread.isRunning():
|
|
self.worker_thread.quit()
|
|
self.worker_thread.wait()
|
|
event.accept()
|
|
|
|
|
|
def main():
|
|
"""Run the application."""
|
|
app = QApplication(sys.argv)
|
|
window = HeliosTestApp()
|
|
window.show()
|
|
sys.exit(app.exec())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|