#!/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 ''}") 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}: ") 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()