#!/opt/srasenv/bin/python3 """ sc3_aui_app.py — Scanengine-3 AUI Loads sc3-aui-main.ui, sc3-aui-camera.ui, sc3-aui-scanprogress.ui via uic and wires up T3R, BBD202, oscilloscope, and camera hardware workers. """ import struct import sys import time from pathlib import Path from threading import Thread import numpy as np from PyQt6 import uic from PyQt6.QtCore import QThread, QTimer, Qt, pyqtSignal from PyQt6.QtGui import QImage, QPixmap from PyQt6.QtWidgets import ( QApplication, QDialog, QDialogButtonBox, QFileDialog, QHBoxLayout, QLabel, QListWidget, QListWidgetItem, QMainWindow, QMessageBox, QPushButton, QSizePolicy, QVBoxLayout, QWidget, ) from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvas from matplotlib.figure import Figure ROOT = Path(__file__).parent sys.path.insert(0, str(ROOT)) from core.config import ScanDefaults from core.rotation import DEFAULT_ROTATION, RotationAxis from core.scan_engine import ( ResumeState, LASER_FREQ_HZ, SCAN_VELOCITY_MM_S, ) from core.scan_geometry import ScanPlan, EtaEstimator, build_plan, format_eta from core.scan_resume import is_compatible, plan_resume from core.scope_sras import SAMPLE_RATE_HZ, configure_channels from core.sras_format import SCAN_CHANNELS, SrasFile, plan_from_header from gui.qt_t3r import QtT3RAdapter from gui.qt_workers import PollingQueueWorker, QueueWorker from gui.inspect_bridge import QtAngleInspector from gui.scan_bridge import QtScanController from hardware.helios_laser import HeliosLaser from hardware.pybbd202 import AXIS_X, AXIS_Y, ThorlabsServoDriver from hardware.tektronix_base import TektronixOscilloscopeBase from hardware.uc480_camera import ( CameraStreamThread, UC480Camera, find_camera_bus_conflicts, ) from t3r_control_panel import T3RControlPanel DEFAULTS = ScanDefaults.load() BBD_DEFAULT_JOG_MM = 0.5 # default jog step for BBD202 class DCBiasImageWidget(FigureCanvas): """Live 2D DC-bias image: rows × frames, matching the viewer's DC image. The image buffer is allocated once per scan and updated in place. The previous version rebuilt the whole array and the entire matplotlib artist tree on every row, so the live preview got steadily slower as a scan progressed (O(rows²) work for one scan). """ def __init__(self, parent=None): fig = Figure(tight_layout=True) super().__init__(fig) self.setParent(parent) self.setMinimumSize(300, 150) self.setSizePolicy( QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding, ) self.ax = fig.add_subplot(111) self._img: np.ndarray | None = None self._im = None self._placeholder = None self._reset_axes() def _reset_axes(self): self.ax.clear() self.ax.set_title("DC Bias Preview (ADC counts)") self.ax.set_xlabel("Frame") self.ax.set_ylabel("Row") self._placeholder = self.ax.text( 0.5, 0.5, "Waiting for data…", transform=self.ax.transAxes, ha="center", va="center", color="gray") self._im = None self.draw_idle() def begin_scan(self, n_rows: int, n_frames: int): """Preallocate the image for a scan of this shape.""" self._img = np.full((max(1, n_rows), max(1, n_frames)), np.nan, dtype=np.float32) self._reset_axes() def add_row(self, row_index: int, values): """Store per-frame DC means for row_index (1-based) and refresh.""" row = np.asarray(values, dtype=np.float32) idx = row_index - 1 # Grow only if the scan turned out taller/wider than announced. if self._img is None: self.begin_scan(idx + 1, len(row)) if idx >= self._img.shape[0] or len(row) > self._img.shape[1]: grown = np.full((max(idx + 1, self._img.shape[0]), max(len(row), self._img.shape[1])), np.nan, dtype=np.float32) grown[:self._img.shape[0], :self._img.shape[1]] = self._img self._img = grown self._im = None # extent changed; rebuild the artist once self._img[idx, :len(row)] = row if self._im is None: if self._placeholder is not None: self._placeholder.remove() self._placeholder = None self._im = self.ax.imshow( self._img, aspect="auto", origin="upper", cmap="viridis", interpolation="nearest", ) else: self._im.set_data(self._img) finite = self._img[np.isfinite(self._img)] if finite.size: self._im.set_clim(float(finite.min()), float(finite.max())) self.draw_idle() BBD_JOG_SPEED_MM_S = 10.0 BBD_JOG_ACCEL_MM_S2 = 50.0 # ── BBD202 worker ───────────────────────────────────────────────────────────── class BBD202Worker(PollingQueueWorker): """Manages ThorlabsServoDriver (MLS203-1) in a worker thread.""" position_updated = pyqtSignal(float, float) # x_mm, y_mm homed_status = pyqtSignal(bool, bool) # x_homed, y_homed POLL_INTERVAL_S = 0.2 def __init__(self): super().__init__(poll_interval_s=self.POLL_INTERVAL_S) self.controller: ThorlabsServoDriver | None = None self.scanning_active = False # pause polling during a scan self._jog_step = BBD_DEFAULT_JOG_MM self._last_pos: tuple[float, float] | None = None self._last_homed: tuple[bool, bool] | None = None self._handlers.update({ "connect": self._do_connect, "disconnect": self._do_disconnect, "jog": self._do_jog, "home_all": self._do_home_all, "enable_all": self._do_enable_all, }) def _on_stop(self): if self.controller: try: self.controller.disconnect() except Exception: pass def _do_connect(self, port: str): try: self.controller = ThorlabsServoDriver() self.controller.serial_port = port self.controller.connect() self.controller.enable_axis(AXIS_X) self.controller.enable_axis(AXIS_Y) self.controller.start_polling(interval=0.2) time.sleep(0.3) self.controller.set_velocity_params(AXIS_X, max_velocity=BBD_JOG_SPEED_MM_S, acceleration=BBD_JOG_ACCEL_MM_S2) self.controller.set_velocity_params(AXIS_Y, max_velocity=BBD_JOG_SPEED_MM_S, acceleration=BBD_JOG_ACCEL_MM_S2) self.is_connected = True self.connected.emit() self.start_polling() except Exception as e: self.connection_failed.emit(str(e)) def _do_disconnect(self): self.stop_polling() if self.controller: try: self.controller.disconnect() except Exception: pass self.controller = None self.is_connected = False self.disconnected.emit() def _do_jog(self, axis: str, direction: int): if not self.controller: return dest = AXIS_X if axis == "x" else AXIS_Y try: self.controller.move_axis_relative(dest, self._jog_step * direction, timeout=2.0) except TimeoutError: pass def _do_home_all(self): if not self.controller: return def _home_task(): for dest, name in [(AXIS_X, "X"), (AXIS_Y, "Y")]: try: self.controller.home_axis(dest, timeout=120.0) except Exception as e: self.error_occurred.emit(f"Home {name} failed: {e}") Thread(target=_home_task, daemon=True).start() def _do_enable_all(self): if not self.controller: return self.controller.toggle_enabled_state(AXIS_X) self.controller.toggle_enabled_state(AXIS_Y) def _poll_once(self): """Emit position/homed updates, but only when they actually change.""" if self.scanning_active or not self.controller: return try: pos = (self.controller.positions[0], self.controller.positions[1]) if pos != self._last_pos: self._last_pos = pos self.position_updated.emit(*pos) homed = (self.controller.am_homed[0], self.controller.am_homed[1]) if homed != self._last_homed: self._last_homed = homed self.homed_status.emit(*homed) except Exception: pass # ── Queue API ───────────────────────────────────────────────────────────── def queue_connect(self, port: str): self._enqueue("connect", port=port) def queue_disconnect(self): self._enqueue("disconnect") def queue_jog(self, axis: str, direction: int): self._enqueue("jog", axis=axis, direction=direction) def queue_home_all(self): self._enqueue("home_all") def queue_enable_all(self): self._enqueue("enable_all") # ── Oscilloscope worker ─────────────────────────────────────────────────────── class OscopeWorker(QueueWorker): """Manages TektronixOscilloscopeBase in a worker thread.""" def __init__(self): super().__init__() self.scope: TektronixOscilloscopeBase | None = None self._handlers.update({ "connect": self._do_connect, "disconnect": self._do_disconnect, }) def _on_stop(self): if self.scope: try: self.scope.disconnect() except Exception: pass def _do_connect(self, ip: str): try: self.scope = TektronixOscilloscopeBase(resource_name=ip, port=4000, timeout=10.0) self.scope.connect() configure_channels(self.scope) self.is_connected = True self.connected.emit() except Exception as e: self.connection_failed.emit(str(e)) def _do_disconnect(self): if self.scope: try: self.scope.disconnect() except Exception: pass self.scope = None self.is_connected = False self.disconnected.emit() def queue_connect(self, ip: str): self._enqueue("connect", ip=ip) def queue_disconnect(self): self._enqueue("disconnect") # ── Helios laser worker ─────────────────────────────────────────────────────── class HeliosWorker(PollingQueueWorker): """Manages HeliosLaser in a worker thread using a command queue. Status polling is self-rescheduling: the next poll is queued only once the previous completes. A free-running 1 s timer used to queue a ~2 s job, so the queue grew without bound and the panel fell further behind the longer it stayed connected. """ enabled_updated = pyqtSignal(bool) current_updated = pyqtSignal(int) diode_temp_updated = pyqtSignal(float) pstage_temp_updated = pyqtSignal(float) qswitch_temp_updated = pyqtSignal(float) status_registers_updated = pyqtSignal(object, object, object) # LER, LCE, CCE (int|None) POLL_INTERVAL_S = 1.0 def __init__(self): super().__init__(poll_interval_s=self.POLL_INTERVAL_S) self._laser: HeliosLaser | None = None self._handlers.update({ "connect": self._do_connect, "disconnect": self._do_disconnect, "set_enable": self._do_set_enable, "set_current": self._do_set_current, }) def _on_stop(self): if self._laser: try: self._laser.disconnect() except Exception: pass def _do_connect(self, port: str): try: self._laser = HeliosLaser(timeout=1.0) if not self._laser.connect(port): self._laser = None self.connection_failed.emit("Failed to open serial port") return self.is_connected = True self.connected.emit() self.start_polling() except Exception as e: self._laser = None self.connection_failed.emit(str(e)) def _do_disconnect(self): self.stop_polling() if self._laser: try: self._laser.set_laser_enable(False) except Exception: pass try: self._laser.disconnect() except Exception: pass self._laser = None self.is_connected = False self.disconnected.emit() def _do_set_enable(self, enable: bool): if not self._laser: return self._laser.set_laser_enable(enable) self.enabled_updated.emit(self._laser.is_laser_enabled()) def _do_set_current(self, current_ma: int): if not self._laser: return self._laser.set_current_ma(current_ma) self.current_updated.emit(current_ma) def _poll_once(self): """Read the full status set. Individual reads are allowed to fail (a timed-out register shouldn't suppress the rest of the panel).""" if not self._laser or not self._laser.is_connected: return try: self.status_registers_updated.emit(*self._laser.get_status_registers()) except Exception: pass try: self.enabled_updated.emit(self._laser.is_laser_enabled()) except Exception: pass for getter, signal in ( (self._laser.get_current_ma, self.current_updated), (self._laser.get_diode_temp_c, self.diode_temp_updated), (self._laser.get_power_stage_temp_c, self.pstage_temp_updated), (self._laser.get_qswitch_temp_c, self.qswitch_temp_updated), ): try: value = getter() if value is not None: signal.emit(value) except Exception: pass def queue_connect(self, port: str): self._enqueue("connect", port=port) def queue_disconnect(self): self._enqueue("disconnect") def queue_set_enable(self, enable: bool): self._enqueue("set_enable", enable=enable) def queue_set_current(self, current_ma: int): self._enqueue("set_current", current_ma=current_ma) # ── Camera window popup ─────────────────────────────────────────────────────── class CameraWindow(QWidget): """Camera display popup — auto-connects on show, auto-disconnects on close.""" closed = pyqtSignal() def __init__(self, parent: QWidget | None = None): super().__init__(parent, Qt.WindowType.Window) uic.loadUi(ROOT / "sc3-aui-camera.ui", self) self.setWindowTitle("uC480 Camera") self._camera: UC480Camera | None = None self._stream: CameraStreamThread | None = None # Banner warning when the camera shares a USB controller with serial # adapters — opening any of those ports (T3R, BBD202, …) collapses # the delivered frame rate to <2 fps (USB split-transaction # contention), which makes focusing impossible. self._bus_warn_label = QLabel(self) self._bus_warn_label.setWordWrap(True) self._bus_warn_label.setStyleSheet( "background: #7f1d1d; color: white; padding: 6px; font-weight: bold;") self._bus_warn_label.setVisible(False) self.verticalLayout.insertWidget(0, self._bus_warn_label) # Overlay a QLabel for frame rendering on top of the native display widget self._frame_label = QLabel(self.uc480_display_area) self._frame_label.setGeometry( 0, 0, self.uc480_display_area.width(), self.uc480_display_area.height(), ) self._frame_label.setAlignment(Qt.AlignmentFlag.AlignCenter) self._frame_label.setStyleSheet("background: black;") self.uc480_exposure_slider.setRange(1, 500) # 0.1 ms … 50 ms (×0.1) self.uc480_gain_slider.setRange(0, 100) self.uc480_exposure_slider.valueChanged.connect(self._set_exposure) self.uc480_gain_slider.valueChanged.connect(self._set_gain) self.uc480_start_btn.clicked.connect(self._start_stream) self.uc480_stop_btn.clicked.connect(self._stop_stream) self.uc480_close_window_btn.clicked.connect(self.close) self._update_controls() def showEvent(self, event): super().showEvent(event) self._connect_camera() self._start_stream() def closeEvent(self, event): self._stop_stream() self._disconnect_camera() super().closeEvent(event) self.closed.emit() def _connect_camera(self): if self._camera is not None: return try: self._camera = UC480Camera() self._camera.initialize() exp_ms = self._camera.get_exposure() self.uc480_exposure_slider.blockSignals(True) self.uc480_exposure_slider.setValue(max(1, round(exp_ms * 10))) self.uc480_exposure_slider.blockSignals(False) except Exception as e: self._camera = None self._frame_label.setText(f"Camera error:\n{e}") self._refresh_bus_warning() self._update_controls() def _refresh_bus_warning(self): conflicts = find_camera_bus_conflicts() if conflicts: devs = ", ".join( f"/dev/{name} ({product})" if product else f"/dev/{name}" for name, _, product in conflicts ) self._bus_warn_label.setText( f"⚠ Camera shares a USB controller with: {devs}. Opening any of " f"these serial ports (T3R panel, BBD202, …) will collapse the " f"frame rate. Fix: plug the camera into a USB-3 (xHCI) port." ) self._bus_warn_label.setVisible(bool(conflicts)) def _disconnect_camera(self): if self._camera: try: self._camera.cleanup() except Exception: pass self._camera = None self._update_controls() def _start_stream(self): if self._stream is not None or self._camera is None: return self._stream = CameraStreamThread(self._camera) self._stream.frame_ready.connect(self._on_frame) self._stream.start() self._update_controls() def _stop_stream(self): if self._stream: self._stream.stop() self._stream = None self._update_controls() def _on_frame(self, image: QImage): px = QPixmap.fromImage(image).scaled( self._frame_label.width(), self._frame_label.height(), Qt.AspectRatioMode.KeepAspectRatio, Qt.TransformationMode.FastTransformation, ) self._frame_label.setPixmap(px) def _set_exposure(self, val: int): if self._camera: try: self._camera.set_exposure(val * 0.1) except Exception: pass def _set_gain(self, val: int): if self._camera: try: self._camera.set_gain(val) except Exception: pass def _update_controls(self): camera_ok = self._camera is not None streaming = self._stream is not None self.uc480_start_btn.setEnabled(camera_ok and not streaming) self.uc480_stop_btn.setEnabled(streaming) self.uc480_exposure_slider.setEnabled(camera_ok) self.uc480_gain_slider.setEnabled(camera_ok) # ── Helios laser panel ──────────────────────────────────────────────────────── class HeliosWindow(QWidget): """Helios laser control panel — shown/hidden with a toggle in the main window.""" closed = pyqtSignal() def __init__(self, worker: HeliosWorker, parent: QWidget | None = None): super().__init__(parent, Qt.WindowType.Window) uic.loadUi(ROOT / "sc3-aui-helios.ui", self) self.setWindowTitle("Helios Laser") self._worker = worker # Polling is driven by the worker itself (self-rescheduling), so # there is no timer here to outpace the device. self._wire_signals() self._update_controls(False) self._set_emission_indicator(False) def closeEvent(self, event): if self._worker.is_connected: self._worker.queue_disconnect() super().closeEvent(event) self.closed.emit() def _wire_signals(self): self.helios_connect_toggle.toggled.connect(self._on_connect_toggle) self._worker.connected.connect(self._on_connected) self._worker.disconnected.connect(self._on_disconnected) self._worker.connection_failed.connect(self._on_connection_failed) self.helios_enable_btn.toggled.connect(self._on_enable_toggle) self._worker.enabled_updated.connect(self._on_enabled_updated) self.helios_set_current_btn.clicked.connect(self._on_set_current) self._worker.current_updated.connect(self._on_current_updated) self._worker.diode_temp_updated.connect( lambda t: self.helios_temp_diode_label.setText(f"{t:.1f} °C") ) self._worker.pstage_temp_updated.connect( lambda t: self.helios_temp_pstage_label.setText(f"{t:.1f} °C") ) self._worker.qswitch_temp_updated.connect( lambda t: self.helios_temp_qswitch_label.setText(f"{t:.1f} °C") ) self._worker.status_registers_updated.connect(self._on_status_registers) self._worker.error_occurred.connect( lambda m: self.helios_status_label.setText(m) ) def _on_connect_toggle(self, checked: bool): if checked: self.helios_connect_toggle.setText("Connecting…") self.helios_connect_toggle.setEnabled(False) port = self.helios_port_edit.text().strip() DEFAULTS.helios_port = port DEFAULTS.save() self._worker.queue_connect(port) else: self._worker.queue_disconnect() def _on_connected(self): self.helios_connect_toggle.setEnabled(True) self.helios_connect_toggle.setText("Disconnect") self.helios_status_label.setText("Connected") self._update_controls(True) def _on_disconnected(self): self.helios_connect_toggle.blockSignals(True) self.helios_connect_toggle.setChecked(False) self.helios_connect_toggle.setText("Connect") self.helios_connect_toggle.setEnabled(True) self.helios_connect_toggle.blockSignals(False) self.helios_status_label.setText("Disconnected") self._update_controls(False) def _on_connection_failed(self, msg: str): self.helios_connect_toggle.blockSignals(True) self.helios_connect_toggle.setChecked(False) self.helios_connect_toggle.setText("Connect") self.helios_connect_toggle.setEnabled(True) self.helios_connect_toggle.blockSignals(False) self.helios_status_label.setText(f"Failed: {msg}") self._update_controls(False) def _set_emission_indicator(self, enabled: bool): if enabled: self.helios_emission_indicator.setStyleSheet( "background-color: #ff2222; border-radius: 10px;" ) else: self.helios_emission_indicator.setStyleSheet( "background-color: #444444; border-radius: 10px;" ) def _on_enable_toggle(self, checked: bool): self.helios_enable_btn.setText("Disable Laser" if checked else "Enable Laser") self._worker.queue_set_enable(checked) def _on_enabled_updated(self, enabled: bool): self.helios_enable_btn.blockSignals(True) self.helios_enable_btn.setChecked(enabled) self.helios_enable_btn.setText("Disable Laser" if enabled else "Enable Laser") self.helios_enable_btn.blockSignals(False) self._set_emission_indicator(enabled) def _on_set_current(self): self._worker.queue_set_current(self.helios_current_spin.value()) def _on_current_updated(self, current_ma: int): self.helios_current_spin.blockSignals(True) self.helios_current_spin.setValue(current_ma) self.helios_current_spin.blockSignals(False) def _reset_temperatures(self): self.helios_temp_diode_label.setText("--- °C") self.helios_temp_pstage_label.setText("--- °C") self.helios_temp_qswitch_label.setText("--- °C") def _on_status_registers(self, ler, lce, cce): self.helios_ler_label.setText(f"0x{ler:04X}" if ler is not None else "---") self.helios_lce_label.setText(f"0x{lce:04X}" if lce is not None else "---") self.helios_cce_label.setText(f"0x{cce:04X}" if cce is not None else "---") def _update_controls(self, connected: bool): self.helios_enable_btn.setEnabled(connected) self.helios_current_spin.setEnabled(connected) self.helios_set_current_btn.setEnabled(connected) if not connected: self.helios_enable_btn.blockSignals(True) self.helios_enable_btn.setChecked(False) self.helios_enable_btn.setText("Enable Laser") self.helios_enable_btn.blockSignals(False) self._set_emission_indicator(False) self._reset_temperatures() self.helios_ler_label.setText("---") self.helios_lce_label.setText("---") self.helios_cce_label.setText("---") # ── Resume angle picker ─────────────────────────────────────────────────────── class ResumeAngleDialog(QDialog): """Lets the operator pick which angles of a .sras file to (re)acquire. Angles whose data is missing or truncated on disk are pre-checked; any other angle can also be checked to force it to be redone (e.g. a completed angle that's known to be bad). """ def __init__(self, statuses, parent: QWidget | None = None): super().__init__(parent) self.setWindowTitle("Select Angles to Rescan") self.resize(480, 420) layout = QVBoxLayout(self) layout.addWidget(QLabel( "Check the angle(s) to (re)acquire. Missing/truncated angles are " "pre-checked; check any other angle to force it to be redone." )) self.list_widget = QListWidget(self) for s in statuses: label = ( f"Angle {s.index + 1}/{len(statuses)} — {s.angle_deg:.2f}° — " f"{s.n_rows_available}/{s.n_rows} rows — {s.status}" ) item = QListWidgetItem(label) item.setData(Qt.ItemDataRole.UserRole, s.index) item.setCheckState( Qt.CheckState.Unchecked if s.complete else Qt.CheckState.Checked ) self.list_widget.addItem(item) layout.addWidget(self.list_widget) btn_row = QHBoxLayout() select_all_btn = QPushButton("Select All") select_none_btn = QPushButton("Select None") select_all_btn.clicked.connect(lambda: self._set_all_checked(True)) select_none_btn.clicked.connect(lambda: self._set_all_checked(False)) btn_row.addWidget(select_all_btn) btn_row.addWidget(select_none_btn) btn_row.addStretch(1) layout.addLayout(btn_row) buttons = QDialogButtonBox( QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel ) buttons.accepted.connect(self.accept) buttons.rejected.connect(self.reject) layout.addWidget(buttons) def _set_all_checked(self, checked: bool): state = Qt.CheckState.Checked if checked else Qt.CheckState.Unchecked for i in range(self.list_widget.count()): self.list_widget.item(i).setCheckState(state) def selected_indices(self) -> list[int]: out = [] for i in range(self.list_widget.count()): item = self.list_widget.item(i) if item.checkState() == Qt.CheckState.Checked: out.append(item.data(Qt.ItemDataRole.UserRole)) return sorted(out) # ── Scan progress popup ─────────────────────────────────────────────────────── class ScanProgressWindow(QWidget): abort_requested = pyqtSignal() pause_toggled = pyqtSignal(bool) # True = pause requested, False = resume def __init__(self, parent: QWidget | None = None): super().__init__(parent, Qt.WindowType.Window) uic.loadUi(ROOT / "sc3-aui-scanprogress.ui", self) self.setWindowTitle("Scan Progress") self.abort_btn.clicked.connect(self.abort_requested.emit) self.pause_btn.toggled.connect(self._on_pause_btn) self.bias_image_widget = DCBiasImageWidget() insert_pos = self.verticalLayout.count() - 2 # above PAUSE + ABORT self.verticalLayout.insertWidget(insert_pos, QLabel("DC Bias Preview")) self.verticalLayout.insertWidget(insert_pos + 1, self.bias_image_widget) self.verticalLayout.setStretch(insert_pos + 1, 1) self._eta = EtaEstimator() def _on_pause_btn(self, checked: bool): # Pause takes effect at the next row boundary; show the intermediate # state until the worker confirms via on_worker_paused(). self.pause_btn.setText("PAUSING…" if checked else "PAUSE") self.pause_toggled.emit(checked) def on_worker_paused(self, paused: bool): if paused and self.pause_btn.isChecked(): self.pause_btn.setText("RESUME") def reset_pause_btn(self): self.pause_btn.blockSignals(True) self.pause_btn.setChecked(False) self.pause_btn.setText("PAUSE") self.pause_btn.blockSignals(False) def on_row_started(self, row: int, n_rows: int, angle_idx: int, n_angles: int): self._eta.row_started() def update_progress(self, row: int, n_rows: int, angle_idx: int, n_angles: int): n_rows = max(1, n_rows) n_angles = max(1, n_angles) self.current_scan_current_row_indicator.setText(f"Row {row} of {n_rows}") self.overall_scan_current_angle_indicator.setText(f"Angle {angle_idx} of {n_angles}") if row == 0: self._eta.reset() self.current_scan_progbar.setMaximum(n_rows) self.current_scan_progbar.setValue(0) self.current_scan_progbar.setFormat("Row %v/%m") else: self._eta.row_finished(angle_idx) self.current_scan_progbar.setMaximum(n_rows) self.current_scan_progbar.setValue(row) rows_left = n_rows - row eta_secs = self._eta.eta_secs(rows_left) if eta_secs is not None: self.current_scan_progbar.setFormat( f"Row %v/%m — ETA: {format_eta(eta_secs)}") elif rows_left == 0: self.current_scan_progbar.setFormat("Row %v/%m — Done") else: self.current_scan_progbar.setFormat("Row %v/%m") pct_ang = round(angle_idx / n_angles * 100) self.overall_scan_progbar.setValue(pct_ang) def update_dc_bias_plot(self, row: int, values): self.bias_image_widget.add_row(row, values) def begin_scan(self, n_rows: int, n_frames: int): """Size the DC preview for the scan about to run.""" self.bias_image_widget.begin_scan(n_rows, n_frames) # ── Scan worker ─────────────────────────────────────────────────────────────── # ── Main window ─────────────────────────────────────────────────────────────── class AngleInspectWindow(QWidget): """Click through a plan's angles, parking the rig at a point in each. Deliberately shows no waveform. The operator reads the SAW response and the bias levels off the oscilloscope itself — this window only says where the rig is and lets them move it somewhere else. """ goto_requested = pyqtSignal(int) new_point_requested = pyqtSignal() stop_requested = pyqtSignal() def __init__(self, angle_labels: list[str], parent: QWidget | None = None): super().__init__(parent, Qt.WindowType.Window) self.setWindowTitle("Inspect Angles") self.resize(420, 460) layout = QVBoxLayout(self) layout.addWidget(QLabel( "Select an angle to rotate to it and park at a random point in " "its scan area.\nRead the SAW response on the oscilloscope." )) self.angle_list = QListWidget(self) for i, label in enumerate(angle_labels): item = QListWidgetItem(label) item.setData(Qt.ItemDataRole.UserRole, i) self.angle_list.addItem(item) self.angle_list.setCurrentRow(0) # currentRowChanged would also fire when the code syncs the highlight # back after a move, re-triggering the move it is reporting. self.angle_list.itemClicked.connect(self._on_item_clicked) layout.addWidget(self.angle_list) nav_row = QHBoxLayout() self.prev_btn = QPushButton("◀ Previous") self.next_btn = QPushButton("Next ▶") self.new_point_btn = QPushButton("New Point") self.new_point_btn.setToolTip( "Pick another random point at this angle without rotating — a bad " "spot on the sample looks the same as a bad angle until you move." ) self.prev_btn.clicked.connect(self._on_prev) self.next_btn.clicked.connect(self._on_next) self.new_point_btn.clicked.connect(self.new_point_requested.emit) nav_row.addWidget(self.prev_btn) nav_row.addWidget(self.next_btn) nav_row.addWidget(self.new_point_btn) layout.addLayout(nav_row) self.point_label = QLabel("—") self.point_label.setStyleSheet("font-weight: bold;") layout.addWidget(self.point_label) self.status_label = QLabel("Starting …") self.status_label.setWordWrap(True) layout.addWidget(self.status_label) self.close_btn = QPushButton("Close") self.close_btn.clicked.connect(self.close) layout.addWidget(self.close_btn) self._n_angles = len(angle_labels) self._angle_idx = 0 self.set_busy(True) # ── Worker → window ─────────────────────────────────────────────────────── def set_busy(self, busy: bool): """Lock navigation while the stage is moving; the rig is not re-entrant.""" for w in (self.angle_list, self.prev_btn, self.next_btn, self.new_point_btn): w.setEnabled(not busy) def on_status(self, msg: str): self.status_label.setText(msg) def on_point(self, point): self._angle_idx = point.angle_idx self.angle_list.setCurrentRow(point.angle_idx) self.point_label.setText(point.describe()) # ── Window → worker ─────────────────────────────────────────────────────── def _on_item_clicked(self, item): self.goto_requested.emit(item.data(Qt.ItemDataRole.UserRole)) def _on_prev(self): self.goto_requested.emit((self._angle_idx - 1) % self._n_angles) def _on_next(self): self.goto_requested.emit((self._angle_idx + 1) % self._n_angles) def closeEvent(self, event): self.stop_requested.emit() super().closeEvent(event) class MainWindow(QMainWindow): def __init__(self): super().__init__() uic.loadUi(ROOT / "sc3-aui-main.ui", self) self.setWindowTitle("Scanengine-3 AUI") # ── T3R driver (owns serial + reader thread internally) ────────────── # QtT3RAdapter re-emits the driver's thread-side callbacks as queued # Qt signals so widget slots run on the GUI thread. self._t3r_driver = QtT3RAdapter(parent=self) self._t3r_panel: T3RControlPanel | None = None # ── Worker threads ──────────────────────────────────────────────────── self._bbd_thread = QThread(self) self._bbd_worker = BBD202Worker() self._bbd_worker.moveToThread(self._bbd_thread) self._bbd_thread.started.connect(self._bbd_worker.run) self._oscope_thread = QThread(self) self._oscope_worker = OscopeWorker() self._oscope_worker.moveToThread(self._oscope_thread) self._oscope_thread.started.connect(self._oscope_worker.run) self._helios_thread = QThread(self) self._helios_worker = HeliosWorker() self._helios_worker.moveToThread(self._helios_thread) self._helios_thread.started.connect(self._helios_worker.run) # ── Popup windows ───────────────────────────────────────────────────── self._camera_win = CameraWindow() self._helios_win = HeliosWindow(self._helios_worker) self._scan_progress = ScanProgressWindow() self._scan_worker: QtScanController | None = None self._inspect_thread: QThread | None = None self._inspect_worker: QtAngleInspector | None = None self._inspect_window: AngleInspectWindow | None = None self._scan_thread: QThread | None = None self._bbd_active_jog: tuple[str, int] | None = None # (axis, direction) self._bbd_jog_timer = QTimer(self) self._bbd_jog_timer.setInterval(200) self._bbd_jog_timer.timeout.connect(self._bbd_jog_tick) self._init_menu_bar() self._init_ui_fields() self._wire_signals() self._bbd_thread.start() self._oscope_thread.start() self._helios_thread.start() # ── Menu bar ─────────────────────────────────────────────────────────────── def _init_menu_bar(self): scan_menu = self.menuBar().addMenu("&Scan") self.resume_scan_action = scan_menu.addAction("&Resume Scan…") self.resume_scan_action.triggered.connect(self._on_resume_scan_action) # ── UI initialisation ───────────────────────────────────────────────────── def _init_ui_fields(self): self.t3r_comport_edit.setText(DEFAULTS.t3r_port) self.bbd202_comport_edit.setText(DEFAULTS.bbd_port) self.oscope_ip_edit.setText(DEFAULTS.oscope_ip) self._helios_win.helios_port_edit.setText(DEFAULTS.helios_port) self.lineEdit_10.setText(str(BBD_DEFAULT_JOG_MM)) self.x_start_edit.setText("10.000") self.y_start_edit.setText("10.000") self.x_delta_edit.setText("80.000") self.y_delta_edit.setText("50.000") self.num_angles_edit.setText("1") self.row_spacing_edit.setText("0.250") self.scan_prefix_edit.setText("scan") self.scan_save_dir_edit.setText(DEFAULTS.save_dir) self.burst_mode_check.setChecked(DEFAULTS.burst_mode) self.strict_rows_check.setChecked(DEFAULTS.strict_rows) # Hide the old T3R manual controls; the connect toggle becomes the panel button. for w in ( self.t3r_enable_t1_btn, self.t3r_enable_t2_btn, self.t3r_enable_t3_btn, self.t3r_enable_gr_btn, self.jog_t1_up_btn, self.jog_t1_down_btn, self.jog_t2_up_btn, self.jog_t2_down_btn, self.jog_t3_up_btn, self.jog_t3_down_btn, self.jog_gr_ccw_btn, self.jog_gr_cw_btn, self.t3r_set_current_t1_btn, self.t3r_set_current_t2_btn, self.t3r_set_current_t3_btn, self.t3r_set_current_gr_btn, self.t3r_current_t1_spin, self.t3r_current_t2_spin, self.t3r_current_t3_spin, self.t3r_current_gr_spin, self.t3r_jog_speed_edit, ): w.setVisible(False) self.t3r_connect_toggle.setCheckable(True) self.t3r_connect_toggle.setText("Show T3R Panel") self._set_bbd_controls_enabled(False) # ── Signal wiring ───────────────────────────────────────────────────────── def _wire_signals(self): # T3R — toggle button opens/closes the T3RControlPanel self.t3r_connect_toggle.toggled.connect(self._on_t3r_panel_toggle) # BBD202 self.bbd202_connect_toggle.toggled.connect(self._on_bbd_toggle) self._bbd_worker.connected.connect(self._on_bbd_connected) self._bbd_worker.disconnected.connect(self._on_bbd_disconnected) self._bbd_worker.connection_failed.connect(self._on_bbd_failed) self._bbd_worker.position_updated.connect(self._on_bbd_position) self._bbd_worker.error_occurred.connect( lambda m: print(f"[BBD] {m}") ) self.bbd_home_all_btn.clicked.connect(lambda: self._bbd_worker.queue_home_all()) self.bbd_enable_all_btn.clicked.connect(lambda: self._bbd_worker.queue_enable_all()) self.bbd_jog_x_pos_btn.pressed.connect(lambda: self._start_bbd_jog("x", 1)) self.bbd_jog_x_pos_btn.released.connect(self._stop_bbd_jog) self.bbd_jog_x_neg_btn.pressed.connect(lambda: self._start_bbd_jog("x", -1)) self.bbd_jog_x_neg_btn.released.connect(self._stop_bbd_jog) self.bbd_jog_y_pos_btn.pressed.connect(lambda: self._start_bbd_jog("y", 1)) self.bbd_jog_y_pos_btn.released.connect(self._stop_bbd_jog) self.bbd_jog_y_neg_btn.pressed.connect(lambda: self._start_bbd_jog("y", -1)) self.bbd_jog_y_neg_btn.released.connect(self._stop_bbd_jog) self.bbd_set_current_start_btn.clicked.connect(self._on_set_start_from_pos) self.bbd_set_delta_current_btn.clicked.connect(self._on_calc_delta) # Oscilloscope self.oscope_connect_toggle.toggled.connect(self._on_oscope_toggle) self._oscope_worker.connected.connect(self._on_oscope_connected) self._oscope_worker.disconnected.connect(self._on_oscope_disconnected) self._oscope_worker.connection_failed.connect(self._on_oscope_failed) # Persist port fields to aui_defaults.json on change self.t3r_comport_edit.editingFinished.connect(self._persist_defaults) self.bbd202_comport_edit.editingFinished.connect(self._persist_defaults) self.oscope_ip_edit.editingFinished.connect(self._persist_defaults) self.burst_mode_check.toggled.connect(self._persist_defaults) self.strict_rows_check.toggled.connect(self._persist_defaults) # Camera self.show_camera_toggle.toggled.connect(self._on_camera_toggle) self._camera_win.closed.connect( lambda: self._set_toggle(self.show_camera_toggle, False, "Show Camera Window") ) # Helios self.show_helios_toggle.toggled.connect(self._on_helios_toggle) self._helios_win.closed.connect( lambda: self._set_toggle(self.show_helios_toggle, False, "Show Helios Panel") ) self._helios_win.helios_close_btn.clicked.connect( lambda: self.show_helios_toggle.setChecked(False) ) # Scan self.start_scan_btn.clicked.connect(self._on_start_scan) self.inspect_angles_btn.clicked.connect(self._on_inspect_angles) self.save_dir_browse_btn.clicked.connect(self._on_browse_save_dir) self._scan_progress.abort_requested.connect(self._on_abort_scan) self._scan_progress.pause_toggled.connect(self._on_pause_scan) # ── T3R panel toggle ────────────────────────────────────────────────────── def _on_t3r_panel_toggle(self, checked: bool): if checked: self._show_t3r_panel() elif self._t3r_panel is not None: self._t3r_panel.hide() self.t3r_connect_toggle.setText("Show T3R Panel") def _show_t3r_panel(self): if self._t3r_panel is None: self._t3r_panel = T3RControlPanel(self._t3r_driver, self) # Pre-fill port from the field that's still shown in the main UI port = self.t3r_comport_edit.text().strip() if port: idx = self._t3r_panel.port_combo.findData(port) if idx < 0: self._t3r_panel.port_combo.insertItem(0, port, port) idx = 0 self._t3r_panel.port_combo.setCurrentIndex(idx) self._t3r_panel.finished.connect(self._on_t3r_panel_closed) self._t3r_panel.show() self._t3r_panel.raise_() self.t3r_connect_toggle.setText("Hide T3R Panel") def _on_t3r_panel_closed(self): self._set_toggle(self.t3r_connect_toggle, False, "Show T3R Panel") # ── BBD202 slots ────────────────────────────────────────────────────────── def _on_bbd_toggle(self, checked: bool): if checked: self.bbd202_connect_toggle.setText("Connecting…") self.bbd202_connect_toggle.setEnabled(False) self._bbd_worker.queue_connect(self.bbd202_comport_edit.text().strip()) else: self._bbd_worker.queue_disconnect() def _on_bbd_connected(self): self.bbd202_connect_toggle.setEnabled(True) self.bbd202_connect_toggle.setText("Disconnect") self._set_bbd_controls_enabled(True) def _on_bbd_disconnected(self): self._set_toggle(self.bbd202_connect_toggle, False, "Connect") self._set_bbd_controls_enabled(False) self.bbd_current_x_position_indicator.setText("---.-") self.bbd_current_y_position_indicator.setText("---.-") def _on_bbd_failed(self, msg: str): self._set_toggle(self.bbd202_connect_toggle, False, "Connect") self.bbd202_connect_toggle.setEnabled(True) QMessageBox.warning(self, "BBD202 Connection Failed", msg) def _on_bbd_position(self, x: float, y: float): self.bbd_current_x_position_indicator.setText(f"{x:07.3f}") self.bbd_current_y_position_indicator.setText(f"{y:07.3f}") def _start_bbd_jog(self, axis: str, direction: int): self._bbd_active_jog = (axis, direction) self._bbd_jog_tick() self._bbd_jog_timer.start() def _bbd_jog_tick(self): if self._bbd_active_jog is None: return axis, direction = self._bbd_active_jog try: self._bbd_worker._jog_step = float(self.lineEdit_10.text()) except ValueError: self._bbd_worker._jog_step = BBD_DEFAULT_JOG_MM self._bbd_worker.queue_jog(axis, direction) def _stop_bbd_jog(self): self._bbd_jog_timer.stop() self._bbd_active_jog = None def _on_set_start_from_pos(self): try: x = float(self.bbd_current_x_position_indicator.text()) y = float(self.bbd_current_y_position_indicator.text()) self.x_start_edit.setText(f"{x:.3f}") self.y_start_edit.setText(f"{y:.3f}") except ValueError: pass def _on_calc_delta(self): try: xs = float(self.x_start_edit.text()) ys = float(self.y_start_edit.text()) cx = float(self.bbd_current_x_position_indicator.text()) cy = float(self.bbd_current_y_position_indicator.text()) self.x_delta_edit.setText(f"{abs(cx - xs):.3f}") self.y_delta_edit.setText(f"{abs(cy - ys):.3f}") except ValueError: pass def _set_bbd_controls_enabled(self, on: bool): for w in ( self.bbd_home_all_btn, self.bbd_enable_all_btn, self.bbd_jog_x_pos_btn, self.bbd_jog_x_neg_btn, self.bbd_jog_y_pos_btn, self.bbd_jog_y_neg_btn, self.bbd_set_current_start_btn, self.bbd_set_delta_current_btn, ): w.setEnabled(on) # ── Oscilloscope slots ──────────────────────────────────────────────────── def _on_oscope_toggle(self, checked: bool): if checked: self.oscope_connect_toggle.setText("Connecting…") self.oscope_connect_toggle.setEnabled(False) self._oscope_worker.queue_connect(self.oscope_ip_edit.text().strip()) else: self._oscope_worker.queue_disconnect() def _on_oscope_connected(self): self.oscope_connect_toggle.setEnabled(True) self.oscope_connect_toggle.setText("Disconnect") def _on_oscope_disconnected(self): self._set_toggle(self.oscope_connect_toggle, False, "Connect") def _on_oscope_failed(self, msg: str): self._set_toggle(self.oscope_connect_toggle, False, "Connect") self.oscope_connect_toggle.setEnabled(True) QMessageBox.warning(self, "Oscilloscope Connection Failed", msg) # ── Persist defaults ────────────────────────────────────────────────────── def _persist_defaults(self): DEFAULTS.t3r_port = self.t3r_comport_edit.text().strip() DEFAULTS.bbd_port = self.bbd202_comport_edit.text().strip() DEFAULTS.oscope_ip = self.oscope_ip_edit.text().strip() DEFAULTS.save_dir = self.scan_save_dir_edit.text().strip() DEFAULTS.burst_mode = self.burst_mode_check.isChecked() DEFAULTS.strict_rows = self.strict_rows_check.isChecked() DEFAULTS.save() # ── Camera toggle ───────────────────────────────────────────────────────── def _on_camera_toggle(self, checked: bool): if checked: self.show_camera_toggle.setText("Hide Camera Window") self._camera_win.show() else: self.show_camera_toggle.setText("Show Camera Window") self._camera_win.close() # ── Helios toggle ───────────────────────────────────────────────────────── def _on_helios_toggle(self, checked: bool): if checked: self.show_helios_toggle.setText("Hide Helios Panel") self._helios_win.show() else: self.show_helios_toggle.setText("Show Helios Panel") self._helios_win.close() # ── Scan ────────────────────────────────────────────────────────────────── def _on_browse_save_dir(self): d = QFileDialog.getExistingDirectory( self, "Select Scan Save Directory", self.scan_save_dir_edit.text() ) if d: self.scan_save_dir_edit.setText(d) self._persist_defaults() def _on_inspect_angles(self): """Open the pre-scan angle inspector for the plan currently entered.""" if self._scan_thread is not None and self._scan_thread.isRunning(): QMessageBox.warning( self, "Scan In Progress", "A scan is running — abort it before inspecting angles." ) return if self._inspect_thread is not None and self._inspect_thread.isRunning(): self._inspect_window.raise_() self._inspect_window.activateWindow() return try: plan, _, _ = self._build_scan_plan() except ValueError as e: QMessageBox.warning(self, "Invalid Scan Parameters", str(e)) return rotator = RotationAxis(self._t3r_driver.driver, DEFAULT_ROTATION) self._inspect_thread = QThread(self) self._inspect_worker = QtAngleInspector( stage=self._bbd_worker.controller, scope=self._oscope_worker.scope, rotator=rotator, plan=plan, # Same reason the scan does it: the inspector drives the stage from # its own thread, and the position poll shares the BBD TX queue. on_inspect_active=lambda active: setattr( self._bbd_worker, "scanning_active", active), ) self._inspect_worker.moveToThread(self._inspect_thread) window = AngleInspectWindow(self._inspect_worker.angle_labels(), self) self._inspect_window = window window.goto_requested.connect(self._inspect_worker.request_goto) window.new_point_requested.connect(self._inspect_worker.request_new_point) window.stop_requested.connect(self._on_inspect_window_closed) self._inspect_worker.status_msg.connect(window.on_status) self._inspect_worker.point_changed.connect(window.on_point) self._inspect_worker.busy_changed.connect(window.set_busy) self._inspect_worker.start_failed.connect(self._on_inspect_failed) self._inspect_worker.error_occurred.connect( lambda m: QMessageBox.warning(self, "Inspection Error", m)) self._inspect_thread.started.connect(self._inspect_worker.run) self.start_scan_btn.setEnabled(False) self.inspect_angles_btn.setEnabled(False) window.show() self._inspect_thread.start() self._inspect_worker.request_start() def _on_inspect_failed(self, message: str): QMessageBox.warning(self, "Cannot Inspect Angles", message) if self._inspect_window is not None: self._inspect_window.close() def _on_inspect_window_closed(self): """Tear the worker down and hand the hardware back to the scan panel.""" if self._inspect_worker is not None: self._inspect_worker.request_stop() self._inspect_worker.stop_worker() if self._inspect_thread is not None: self._inspect_thread.quit() self._inspect_thread.wait(10000) self._inspect_thread = None self._inspect_worker = None self._inspect_window = None self.start_scan_btn.setEnabled(True) self.inspect_angles_btn.setEnabled(True) def _on_start_scan(self): try: plan, prefix, save_dir = self._build_scan_plan() except ValueError as e: QMessageBox.warning(self, "Invalid Scan Parameters", str(e)) return self._launch_scan_worker(plan, prefix, save_dir) def _on_resume_scan_action(self): if self._scan_thread is not None and self._scan_thread.isRunning(): QMessageBox.warning( self, "Scan In Progress", "A scan is already running — abort it before resuming another one." ) return start_dir = self.scan_save_dir_edit.text().strip() or DEFAULTS.save_dir path_str, _ = QFileDialog.getOpenFileName( self, "Select Scan File to Resume", start_dir, "SRAS Scan Files (*.sras)" ) if not path_str: return path = Path(path_str) try: sras = SrasFile(path) statuses = sras.angle_status() except (ValueError, struct.error, OSError) as e: QMessageBox.warning(self, "Cannot Resume Scan", f"Could not read scan file:\n\n{e}") return if not is_compatible(sras, velocity=SCAN_VELOCITY_MM_S, laser_freq=LASER_FREQ_HZ, sample_rate=SAMPLE_RATE_HZ, n_channels=len(SCAN_CHANNELS)): QMessageBox.warning( self, "Cannot Resume Scan", f"{path.name} was recorded with acquisition settings that don't " "match this app's current fixed settings, so appending to it " "would corrupt the file. Resume is only possible for files " "produced by this exact version of the app." ) return dialog = ResumeAngleDialog(statuses, self) if dialog.exec() != QDialog.DialogCode.Accepted: return selected = set(dialog.selected_indices()) if not selected: return resume_plan = plan_resume(statuses, selected) if resume_plan.auto_added: QMessageBox.information( self, "Additional Angles Required", "The scan file has no data yet past angle " f"{resume_plan.frontier_idx + 1}, so angle(s) can't be skipped over. " "The following angle(s) will also be (re)acquired so there's " f"no gap: {[i + 1 for i in resume_plan.auto_added]}" ) angle_list = ", ".join(str(t.angle_idx + 1) for t in resume_plan.targets) reply = QMessageBox.question( self, "Resume Scan", f"Resume {path.name}?\n\n" f"Angle(s) to (re)acquire: {angle_list}\n" f"Total rows to acquire: {resume_plan.total_rows}\n\n" "Existing data for these angle(s) (if any) will be overwritten.", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, ) if reply != QMessageBox.StandardButton.Yes: return self._launch_scan_worker(plan_from_header(sras), path.stem, str(path.parent), resume_plan.to_state(sras)) def _launch_scan_worker(self, plan: ScanPlan, prefix: str, save_dir: str, resume: ResumeState | None = None): rotator = RotationAxis(self._t3r_driver.driver, DEFAULT_ROTATION) self._scan_thread = QThread(self) self._scan_worker = QtScanController( stage=self._bbd_worker.controller, scope=self._oscope_worker.scope, rotator=rotator, plan=plan, out_path=Path(save_dir) / f"{prefix}.sras", resume=resume, # Suppress BBD position polling for the duration of the scan — # a worker concern, not the engine's. on_scan_active=lambda active: setattr( self._bbd_worker, "scanning_active", active), burst_mode=self.burst_mode_check.isChecked(), strict_rows=self.strict_rows_check.isChecked(), ) self._scan_worker.moveToThread(self._scan_thread) self._scan_thread.started.connect(self._scan_worker.run) self._scan_worker.row_started.connect(self._scan_progress.on_row_started) self._scan_worker.row_done.connect(self._on_row_done) self._scan_worker.dc_bias_updated.connect(self._scan_progress.update_dc_bias_plot) self._scan_worker.completed.connect(self._on_scan_complete) self._scan_worker.failed.connect(self._on_scan_failed) self._scan_worker.status_msg.connect(lambda m: print(f"[SCAN] {m}")) self._scan_worker.user_prompt.connect(self._on_scan_user_prompt) self._scan_worker.paused_changed.connect(self._scan_progress.on_worker_paused) self.start_scan_btn.setEnabled(False) self._scan_progress.reset_pause_btn() ai0 = 0 if resume is None else resume.targets[0].angle_idx self._scan_progress.update_progress( 0, plan.per_angle[ai0].n_rows, 0 if resume is None else ai0 + 1, plan.n_angles ) self._scan_progress.begin_scan(plan.per_angle[ai0].n_rows, plan.per_angle[ai0].n_frames) self._scan_progress.show() self._scan_thread.start() def _build_scan_plan(self) -> tuple[ScanPlan, str, str]: def _f(w, label): try: return float(w.text()) except ValueError: raise ValueError(f"'{label}' is not a valid number: {w.text()!r}") from None def _i(w, label): try: return int(w.text()) except ValueError: raise ValueError(f"'{label}' is not a valid integer: {w.text()!r}") from None plan = build_plan( _f(self.x_start_edit, "XS"), _f(self.y_start_edit, "YS"), _f(self.x_delta_edit, "XD"), _f(self.y_delta_edit, "YD"), _i(self.num_angles_edit, "NumAngles"), _f(self.row_spacing_edit, "RowSpacing"), laser_freq_hz=LASER_FREQ_HZ, velocity_mm_s=SCAN_VELOCITY_MM_S, rotation_sign=DEFAULT_ROTATION.rotation_sign, ) prefix = self.scan_prefix_edit.text().strip() or "scan" save_dir = self.scan_save_dir_edit.text().strip() or DEFAULTS.save_dir return plan, prefix, save_dir def _on_row_done(self, row: int, n_rows: int, angle_idx: int, n_angles: int): self._scan_progress.update_progress(row, n_rows, angle_idx, n_angles) def _on_scan_complete(self): self._scan_progress.close() self.start_scan_btn.setEnabled(True) QMessageBox.information( self, "Scan Complete", "All rows and angles have been acquired.\n\n" "Hardware will now be disconnected and the application will close." ) self._disconnect_all_and_close() def _on_scan_failed(self, msg: str): self._scan_progress.close() self.start_scan_btn.setEnabled(True) if "aborted" in msg.lower(): QMessageBox.warning(self, "Scan Aborted", msg) else: QMessageBox.critical(self, "Scan Error", f"Scan stopped with an error:\n\n{msg}") def _on_scan_user_prompt(self, title: str, message: str): QMessageBox.information(self, title, message) if self._scan_worker: self._scan_worker.acknowledge_prompt() def _on_abort_scan(self): if self._scan_worker: self._scan_worker.abort() def _on_pause_scan(self, pause: bool): if self._scan_worker: if pause: self._scan_worker.pause() else: self._scan_worker.resume() def _disconnect_all_and_close(self): """Cleanly disconnect all hardware then quit.""" if self._t3r_driver.is_open: self._t3r_driver.close() if self._bbd_worker.is_connected: self._bbd_worker.queue_disconnect() if self._oscope_worker.is_connected: self._oscope_worker.queue_disconnect() if self._helios_worker.is_connected: self._helios_worker.queue_disconnect() self._camera_win.close() self._helios_win.close() QTimer.singleShot(1500, QApplication.instance().quit) # ── Helpers ─────────────────────────────────────────────────────────────── @staticmethod def _set_toggle(btn, checked: bool, text: str): """Silently update a checkable button without triggering its toggled signal.""" btn.blockSignals(True) btn.setChecked(checked) btn.setText(text) btn.setEnabled(True) btn.blockSignals(False) # ── Cleanup ─────────────────────────────────────────────────────────────── def closeEvent(self, event): self._camera_win.close() self._scan_progress.close() self._helios_win.close() if self._t3r_driver.is_open: self._t3r_driver.close() for w in (self._bbd_worker, self._oscope_worker, self._helios_worker): w.stop_worker() for t in (self._bbd_thread, self._oscope_thread, self._helios_thread): t.quit() t.wait(2000) super().closeEvent(event) # ── Entry point ─────────────────────────────────────────────────────────────── def main(): app = QApplication(sys.argv) app.setStyle("Fusion") win = MainWindow() win.show() sys.exit(app.exec()) if __name__ == "__main__": main()