tka/scanengine-3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
dev-saw-check
scanengine-3/sc3_aui_app.py
T
Thomas Ales 844fcd0297 SAW quality check: one middle row per angle, and a viewer that overlays them 
A full multi-angle scan takes hours, and a rig whose angles disagree produces
all of them before anyone finds out. This adds a test mode that acquires one
row per angle — the row-wise middle of the ROI — and a viewer that puts every
angle's SAW frequency on one graph. The default 80×50 mm ROI at 5 angles goes
from 1461 rows to 5.

Why the middle row answers an alignment question at all: build_plan centres
every angle's rotated bounding box on the same nominal ROI centre, so each
angle's middle row crosses that one point on the sample. All the angles
measure the same material, so a spread in their frequencies belongs to the rig
rather than to where each row happened to land. test_every_angles_middle_row_
crosses_the_roi_centre pins that premise, since the whole comparison rests on
it and nothing else in the geometry code would notice it breaking.

core/saw_check.py — both halves of the mode, kept together because neither is
much use alone. middle_row_plan() reduces a ScanPlan to one row per angle
(n_rows // 2, the upper of two centre rows when even); frequency_traces() and
alignment_summary() turn the resulting file back into per-angle frequency
traces and the scalars an operator is actually asking about — the spread of
the per-angle medians, the worst drift along a row, the sparsest row. The
verdict thresholds are labelled as rules of thumb, not physics: an anisotropic
sample genuinely varies with angle, so a wide spread is a prompt to look at
the curves rather than a verdict.

Format v10: byte-identical to v6, one row per angle. The version byte earns
its keep because the two are otherwise indistinguishable — a v6 scan aborted
after its first row is not a check, and a reader guessing from the row count
would read a failed scan as a deliberate measurement. create_scan_file()
enforces the one-row rule at write time, since nothing downstream can recover
from a v10 file that breaks it. ScanEngine gains file_version and is otherwise
untouched: the acquisition, the abort/pause path and the background capture
are the scan's, unchanged.

sras_scan_manager.py now carries the source file's version through an export
instead of stamping v6 on everything, which the wider reader would otherwise
have made a lie.

saw_check_viewer.py — frequency along the row, one curve per angle, over a
common offset axis so the curves lie on the same piece of sample; a summary of
each angle's median ±1σ against angle; and the per-angle numbers in a table.
Analysis parameters (DC threshold, background, time gate) recompute on a
worker thread; display ones (smoothing, axis, MHz↔m/s) only redraw. A full v6
scan opens too — the same middle row is pulled out of it — so a finished scan
can be re-examined with the check's own read-out.

In the app, a check finishes by handing the operator the file and an "Open
Viewer" button rather than shutting the rig down the way a completed scan
does. Burst mode is not offered: one row per angle means every burst would be
a single row, so it buys nothing and still pays for the gate preflight.

137 tests passing, ruff clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-04 08:13:54 -05:00

1685 lines
69 KiB
Python
Executable File
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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 subprocess
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.saw_check import middle_row_plan
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, VERSION, VERSION_SAW_CHECK, 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
# A SAW check is written beside the scan it belongs to, under the same prefix.
# The suffix keeps it from overwriting the scan itself, which is the one file
# in the directory that cost hours to acquire.
SAW_CHECK_SUFFIX = "-sawcheck"
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
# A SAW check runs through the same worker as a scan; this says which,
# since the two finish very differently (a check hands the operator a
# file to look at; a scan shuts the rig down).
self._scan_is_saw_check = False
self._saw_check_path: Path | 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.saw_check_btn.clicked.connect(self._on_saw_check)
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 _set_scan_buttons_enabled(self, enabled: bool):
"""Both entry points drive the same rig, so they lock and unlock together."""
self.start_scan_btn.setEnabled(enabled)
self.saw_check_btn.setEnabled(enabled)
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_saw_check(self):
"""Acquire the middle row of the current ROI at every angle.
Same engine, same hardware sequence, same file format as a scan — the
plan is just reduced to one row per angle and the result is tagged v10
so the viewer knows it is a check rather than a scan cut short.
"""
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 starting a SAW check."
)
return
try:
plan, prefix, save_dir = self._build_scan_plan()
check_plan = middle_row_plan(plan) # ScanGeometryError is a ValueError
except ValueError as e:
QMessageBox.warning(self, "Invalid Scan Parameters", str(e))
return
check_prefix = f"{prefix}{SAW_CHECK_SUFFIX}"
out_path = Path(save_dir) / f"{check_prefix}.sras"
rows = ", ".join(f"{pa.angle_deg:.1f}°: Y={pa.y_positions[0]:.3f} mm"
for pa in check_plan.per_angle)
overwrite = ("\n\nThis will overwrite the existing file."
if out_path.exists() else "")
reply = QMessageBox.question(
self, "SAW Quality Check",
f"Acquire the middle row of the ROI at {check_plan.n_angles} angle(s)?\n\n"
f"{rows}\n\n"
f"Save → {out_path.name}{overwrite}",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
)
if reply != QMessageBox.StandardButton.Yes:
return
self._saw_check_path = out_path
# Burst mode is deliberately not offered here: one row per angle means
# every burst would be a single row, so it buys nothing and still pays
# for the gate preflight.
self._launch_scan_worker(check_plan, check_prefix, save_dir, saw_check=True)
def _on_saw_check_complete(self):
"""A check is a thing to look at, not a run to shut down after."""
path = self._saw_check_path
box = QMessageBox(self)
box.setIcon(QMessageBox.Icon.Information)
box.setWindowTitle("SAW Check Complete")
box.setText(
f"Middle-row SAW check written to:\n{path}\n\n"
"Open it in the SAW Check Viewer to compare each angle's "
"frequency and judge the alignment."
)
open_btn = box.addButton("Open Viewer", QMessageBox.ButtonRole.AcceptRole)
box.addButton(QMessageBox.StandardButton.Close)
box.exec()
if box.clickedButton() is open_btn:
self._launch_saw_check_viewer(path)
def _launch_saw_check_viewer(self, path: Path):
"""Open the viewer as its own process.
Deliberately not in-process: the acquisition app owns the hardware and
must stay responsive, and the viewer is a separate entry point that
outlives any one scan session.
"""
try:
subprocess.Popen([sys.executable,
str(ROOT / "saw_check_viewer.py"), str(path)])
except OSError as e:
QMessageBox.warning(
self, "Could Not Open Viewer",
f"Could not start the SAW Check Viewer:\n\n{e}\n\n"
f"Run it manually: python saw_check_viewer.py {path}"
)
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._set_scan_buttons_enabled(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._set_scan_buttons_enabled(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,
saw_check: bool = False):
rotator = RotationAxis(self._t3r_driver.driver, DEFAULT_ROTATION)
self._scan_is_saw_check = saw_check
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() and not saw_check,
strict_rows=self.strict_rows_check.isChecked(),
file_version=VERSION_SAW_CHECK if saw_check else VERSION,
)
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._set_scan_buttons_enabled(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._set_scan_buttons_enabled(True)
if self._scan_is_saw_check:
self._scan_is_saw_check = False
self._on_saw_check_complete()
return
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._set_scan_buttons_enabled(True)
self._scan_is_saw_check = False
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()
Reference in New Issue View Git Blame Copy Permalink