Thomas Ales
52fdcdd9f3
A mis-triggered row cannot be written as it arrived — v6 declares n_frames per row in the header and has no per-row length field, so a short or long row would shift every later row in the file. Until now the only policy was to square it up, which keeps the scan running but leaves the affected row indistinguishable from a good one afterwards: nothing in the file records that it was padded. strict_rows selects the other trade. On any frame-count mismatch the scan stops instead of writing the row, so a data run either produces rows that mean what the header says they mean or fails loudly. Default stays pad, so existing behaviour is unchanged. _warn_frame_delta becomes _check_frame_delta, since it now decides rather than just reports. Both acquisition paths already call it before writing anything for the row (CH1 leads SCAN_CHANNELS, and the burst path checks every row up front), so an abort leaves the file on a whole-row boundary rather than a half-written row — test_strict_row_packing_writes_nothing_for_the_failed_row pins that. Plumbed through QtScanController to a checkbox in the scan panel, persisted in ScanDefaults alongside burst_mode. scan_format.md documents both policies and notes that the choice is not recorded in the file. The row-clipping setup in the padding test is now a _clip_one_row helper, reused by the strict tests. 92 tests passing, ruff clean. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1406 lines
58 KiB
Python
Executable File
1406 lines
58 KiB
Python
Executable File
#!/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.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 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._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.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_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()
|