Thomas Ales
6e8c1cb7a2
The operator frames a good spot, confirms the two DC levels the detector reads there, and the rig then measures its own tilt: step 1.5 mm either side on X and then on Y, and tilt the platform until those levels come back. The correction that fixes an offset point is the correction that levels the whole travel — height error and tilt effect are both proportional to the offset — so the procedure ends by applying it and leaving it applied. Both directions are measured from the same starting tilt and averaged, which makes their disagreement a flatness read-out rather than something averaged away silently. core/auto_align.py holds the geometry and the search, Qt-free. The three T-axes' azimuths are the whole geometry: T1 lies along +X so it alone tilts along X, and T0/T2 move as an equal-and-opposite pair to tilt along Y without touching X (tilt_response derives that, and the tests pin it — an axis map that drifts would still converge, on the wrong axis). The search is a secant null on the split-detector difference: probe once to learn what a microstep is worth, sign included, then step at the null. It refuses to servo on a scope that has not re-triggered, escalates a probe that reads as no response before calling an axis dead, and stops at a per-axis travel limit. gui/align_bridge.py runs it on a worker thread; stopping is a threading.Event rather than a queued command, because the worker is inside a long handler for the whole run. The camera window carries the button and the progress window, and locks the scan panel and the jog pads while a run owns the stage. Adds immediate MEAN measurements and an acquisition count to the scope driver, and read_bias_mv to core/scope_inspect — the one scalar the inspection state was missing. KNOWN_ISSUES.md records what only the rig can settle: the probe step, the travel limit, the hold current, and whether the piston the X phase applies alongside its tilt matters. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2111 lines
88 KiB
Python
Executable File
2111 lines
88 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 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, QPlainTextEdit,
|
||
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.auto_align import DEFAULT_ALIGN, DEFAULT_T_AXIS
|
||
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_inspect import BIAS_LABELS
|
||
from core.scope_sras import SAMPLE_RATE_HZ, configure_channels
|
||
from core.sras_format import (
|
||
SCAN_CHANNELS, VERSION, VERSION_SAW_CHECK, WRITABLE_VERSIONS, SrasFile,
|
||
plan_from_header,
|
||
)
|
||
from gui.align_bridge import QtAutoAligner
|
||
from gui.jog_panel import (
|
||
BBD_JOG_ACCEL_MM_S2, BBD_JOG_SPEED_MM_S, BBD_JOG_STEP_MM,
|
||
BBDJogPanel, T3RJogPanel,
|
||
)
|
||
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 gui.widgets import mono_font
|
||
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()
|
||
|
||
# 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()
|
||
|
||
|
||
# ── 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_JOG_STEP_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,
|
||
"set_velocity": self._do_set_velocity,
|
||
"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, step_mm: float | None = None):
|
||
if not self.controller:
|
||
return
|
||
dest = AXIS_X if axis == "x" else AXIS_Y
|
||
step = self._jog_step if step_mm is None else step_mm
|
||
try:
|
||
self.controller.move_axis_relative(dest, step * direction, timeout=2.0)
|
||
except TimeoutError:
|
||
pass
|
||
|
||
def _do_set_velocity(self, max_velocity: float, acceleration: float):
|
||
if not self.controller:
|
||
return
|
||
for dest in (AXIS_X, AXIS_Y):
|
||
self.controller.set_velocity_params(dest, max_velocity=max_velocity,
|
||
acceleration=acceleration)
|
||
|
||
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, step_mm: float | None = None):
|
||
self._enqueue("jog", axis=axis, direction=direction, step_mm=step_mm)
|
||
|
||
def queue_set_velocity(self, max_velocity: float, acceleration: float):
|
||
self._enqueue("set_velocity", max_velocity=max_velocity,
|
||
acceleration=acceleration)
|
||
|
||
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 ───────────────────────────────────────────────────────
|
||
|
||
# The channel map auto-align assumes. Nothing reconfigures the scope inputs —
|
||
# CH3 is the max-velocity gate during a scan and the DC 1 monitor here — so the
|
||
# operator is asked to confirm the cabling instead of being trusted to
|
||
# remember it.
|
||
ALIGN_SCOPE_CHANNELS = ((1, "SAW"), (2, "Trigger"), (3, "DC 1"), (4, "DC 2"))
|
||
|
||
|
||
def _align_scope_prompt() -> str:
|
||
channels = "\n".join(f" CH{ch} — {role}" for ch, role in ALIGN_SCOPE_CHANNELS)
|
||
# The channels arrive on screen carrying the inspection labels, which are
|
||
# the names for the same two monitors — say so, or the operator is left
|
||
# comparing "DC 1" here against "Bias - A" on the instrument.
|
||
labels = " and ".join(BIAS_LABELS[ch] for ch in sorted(BIAS_LABELS))
|
||
return (
|
||
"Auto-align reads the DC levels off CH3 and CH4. Check the scope is "
|
||
"cabled this way before starting:\n\n"
|
||
f"{channels}\n\n"
|
||
f"The scope will label the last two {labels}.\n\n"
|
||
"The laser must be pulsing and the detection beam on the sample — the "
|
||
"procedure stops rather than servo on a scope that is not triggering.\n\n"
|
||
"Start auto-align?"
|
||
)
|
||
|
||
|
||
def _align_reference_prompt(reading) -> str:
|
||
return (
|
||
"The detector reads, where the stage is standing:\n\n"
|
||
f" DC 1 {reading.dc1_mv:8.1f} mV\n"
|
||
f" DC 2 {reading.dc2_mv:8.1f} mV\n\n"
|
||
"Is the image correct?\n\n"
|
||
f"Yes takes these as the good values and holds them to within "
|
||
f"{DEFAULT_ALIGN.tolerance_mv:.0f} mV at "
|
||
f"{DEFAULT_ALIGN.offset_mm:.2f} mm either side of here, first on X "
|
||
f"(T1) and then on Y (T0/T2 as a pair)."
|
||
)
|
||
|
||
|
||
class CameraWindow(QWidget):
|
||
"""Camera display popup — auto-connects on show, auto-disconnects on close.
|
||
|
||
Focusing and framing are done by eye, so the T3R and BBD202 jog controls
|
||
live beside the image. Both take the objects the main window already
|
||
owns; passing neither leaves the window as a plain viewer.
|
||
|
||
Auto-align lives here for the same reason: the operator judges the image
|
||
to decide the rig is on a good spot, and that judgement is the first step
|
||
of the procedure. It needs the oscilloscope as well, so the button is
|
||
only offered when all three are on hand.
|
||
"""
|
||
|
||
closed = pyqtSignal()
|
||
align_active = pyqtSignal(bool) # lock the scan panel while aligning
|
||
|
||
def __init__(self, t3r_driver: QtT3RAdapter | None = None,
|
||
bbd_worker: BBD202Worker | None = None,
|
||
oscope_worker: "OscopeWorker | None" = None,
|
||
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
|
||
|
||
self._t3r_driver = t3r_driver
|
||
self._bbd_worker = bbd_worker
|
||
self._oscope_worker = oscope_worker
|
||
self._align_thread: QThread | None = None
|
||
self._align_worker: QtAutoAligner | None = None
|
||
self._align_window: "AutoAlignWindow | None" = None
|
||
self._align_available = True
|
||
|
||
# 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)
|
||
|
||
# Every one of the three is load-bearing: the T3R moves the tilt
|
||
# platform, the BBD202 makes the 1.5 mm steps, and the scope is the
|
||
# only thing that can say whether either helped.
|
||
self.uc480_auto_align_btn.clicked.connect(self._on_auto_align)
|
||
self.uc480_auto_align_btn.setVisible(
|
||
None not in (t3r_driver, bbd_worker, oscope_worker))
|
||
|
||
self.t3r_jog_panel = self.bbd_jog_panel = None
|
||
self._build_jog_column(t3r_driver, bbd_worker)
|
||
|
||
self._update_controls()
|
||
|
||
def _build_jog_column(self, t3r_driver, bbd_worker):
|
||
"""Add the stage controls to the right of the image."""
|
||
if t3r_driver is None and bbd_worker is None:
|
||
return
|
||
column = QVBoxLayout()
|
||
column.setContentsMargins(0, 0, 0, 0)
|
||
if t3r_driver is not None:
|
||
self.t3r_jog_panel = T3RJogPanel(t3r_driver, self)
|
||
column.addWidget(self.t3r_jog_panel)
|
||
if bbd_worker is not None:
|
||
self.bbd_jog_panel = BBDJogPanel(bbd_worker, self)
|
||
column.addWidget(self.bbd_jog_panel)
|
||
column.addStretch()
|
||
self.horizontalLayout.addLayout(column)
|
||
|
||
def showEvent(self, event):
|
||
super().showEvent(event)
|
||
self._connect_camera()
|
||
self._start_stream()
|
||
|
||
def closeEvent(self, event):
|
||
# An alignment outlives this window otherwise, and it owns the stage.
|
||
self._teardown_align()
|
||
# A jog whose button-release lands after the window is gone would
|
||
# otherwise leave an axis running.
|
||
for panel in (self.t3r_jog_panel, self.bbd_jog_panel):
|
||
if panel is not None:
|
||
panel.stop_jogs()
|
||
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)
|
||
|
||
# ── Auto-align ────────────────────────────────────────────────────────────
|
||
|
||
def _on_auto_align(self):
|
||
if self._align_thread is not None and self._align_thread.isRunning():
|
||
if self._align_window is not None:
|
||
self._align_window.raise_()
|
||
self._align_window.activateWindow()
|
||
return
|
||
|
||
problem = self._align_prerequisite_problem()
|
||
if problem:
|
||
QMessageBox.warning(self, "Cannot Auto-Align", problem)
|
||
return
|
||
|
||
if QMessageBox.question(self, "Check the Oscilloscope",
|
||
_align_scope_prompt()) \
|
||
!= QMessageBox.StandardButton.Yes:
|
||
return
|
||
|
||
self._align_thread = QThread(self)
|
||
self._align_worker = QtAutoAligner(
|
||
stage=self._bbd_worker.controller,
|
||
scope=self._oscope_worker.scope,
|
||
t3r=self._t3r_driver.driver,
|
||
# Same reason the scan and the inspector do it: the procedure
|
||
# drives the stage from its own thread, and the position poll
|
||
# shares the BBD TX queue.
|
||
on_align_active=lambda active: setattr(
|
||
self._bbd_worker, "scanning_active", active),
|
||
)
|
||
self._align_worker.moveToThread(self._align_thread)
|
||
|
||
window = AutoAlignWindow(self)
|
||
self._align_window = window
|
||
worker = self._align_worker
|
||
worker.status_msg.connect(window.on_status)
|
||
worker.reading_taken.connect(window.on_reading)
|
||
worker.offset_done.connect(window.on_offset_done)
|
||
worker.axis_done.connect(window.on_axis_done)
|
||
worker.prepared.connect(self._on_align_prepared)
|
||
worker.prepare_failed.connect(self._on_align_prepare_failed)
|
||
worker.finished.connect(self._on_align_finished)
|
||
worker.failed.connect(self._on_align_failed)
|
||
worker.aborted.connect(self._on_align_aborted)
|
||
worker.stopped.connect(self._release_align_thread)
|
||
worker.error_occurred.connect(window.on_failed)
|
||
window.abort_requested.connect(worker.request_abort)
|
||
window.close_requested.connect(self._teardown_align)
|
||
|
||
self._align_thread.started.connect(worker.run)
|
||
self._set_align_ui_active(True)
|
||
window.show()
|
||
self._align_thread.start()
|
||
worker.request_prepare()
|
||
|
||
def _align_prerequisite_problem(self) -> str:
|
||
"""Why auto-align cannot start, or "" if it can."""
|
||
if self._oscope_worker is None or not self._oscope_worker.is_connected:
|
||
return ("The oscilloscope is not connected, and it is the only "
|
||
"thing that can read the DC levels. Connect it from the "
|
||
"main window and try again.")
|
||
if self._bbd_worker is None or not self._bbd_worker.is_connected:
|
||
return ("The BBD202 stage is not connected, so the 1.5 mm steps "
|
||
"cannot be made. Connect it from the main window and try "
|
||
"again.")
|
||
if self._t3r_driver is None or not self._t3r_driver.is_open:
|
||
return ("The T3R is not connected, so the T-axes cannot be moved. "
|
||
"Connect it from the T3R panel and try again.")
|
||
return ""
|
||
|
||
def _on_align_prepared(self, reading):
|
||
"""The rig is configured and standing on the point — ask the operator."""
|
||
if self._align_window is None:
|
||
return
|
||
confirmed = QMessageBox.question(
|
||
self, "Is the Image Correct?", _align_reference_prompt(reading)
|
||
) == QMessageBox.StandardButton.Yes
|
||
|
||
if not confirmed:
|
||
self._align_window.on_status(
|
||
"Cancelled: the image was not confirmed. Nothing has moved — "
|
||
"re-align by hand and start again.")
|
||
self._finish_align()
|
||
return
|
||
|
||
self._align_window.set_reference(reading)
|
||
self._align_worker.request_run()
|
||
|
||
def _on_align_prepare_failed(self, message: str):
|
||
if self._align_window is not None:
|
||
self._align_window.on_failed(message)
|
||
QMessageBox.warning(self, "Cannot Auto-Align", message)
|
||
self._finish_align()
|
||
|
||
def _on_align_finished(self, result):
|
||
if self._align_window is not None:
|
||
self._align_window.on_finished(result)
|
||
self._finish_align()
|
||
|
||
def _on_align_failed(self, message: str):
|
||
if self._align_window is not None:
|
||
self._align_window.on_failed(message)
|
||
QMessageBox.warning(self, "Auto-Align Failed", message)
|
||
self._finish_align()
|
||
|
||
def _on_align_aborted(self):
|
||
if self._align_window is not None:
|
||
self._align_window.on_aborted()
|
||
self._finish_align()
|
||
|
||
def _finish_align(self):
|
||
"""Park the rig and release the thread, leaving the summary on screen."""
|
||
if self._align_worker is not None:
|
||
self._align_worker.request_stop()
|
||
|
||
def _release_align_thread(self):
|
||
"""Worker idle — take its thread down. Safe to call more than once."""
|
||
if self._align_worker is not None:
|
||
self._align_worker.stop_worker()
|
||
if self._align_thread is not None:
|
||
self._align_thread.quit()
|
||
self._align_thread.wait(30000)
|
||
self._align_thread = None
|
||
self._align_worker = None
|
||
self._set_align_ui_active(False)
|
||
|
||
def _teardown_align(self):
|
||
"""Stop an alignment and close its window — the window or camera is
|
||
going away, so the procedure cannot be left running."""
|
||
window, self._align_window = self._align_window, None
|
||
if self._align_worker is not None:
|
||
self._align_worker.request_stop()
|
||
self._release_align_thread()
|
||
if window is not None:
|
||
window.close()
|
||
|
||
def set_align_available(self, available: bool):
|
||
"""The main window handing the rig over, or taking it back.
|
||
|
||
A scan or an angle inspection owns the same stage, so auto-align has
|
||
to be off the table while either runs.
|
||
"""
|
||
self._align_available = available
|
||
self._refresh_align_button()
|
||
|
||
def _refresh_align_button(self):
|
||
self.uc480_auto_align_btn.setEnabled(
|
||
self._align_available and self._align_thread is None)
|
||
|
||
def _set_align_ui_active(self, active: bool):
|
||
"""A running alignment owns the stage and the T-axes; nothing else
|
||
may drive them, here or in the main window."""
|
||
self._refresh_align_button()
|
||
for panel in (self.t3r_jog_panel, self.bbd_jog_panel):
|
||
if panel is not None:
|
||
panel.stop_jogs()
|
||
panel.setEnabled(not active)
|
||
self.align_active.emit(active)
|
||
|
||
|
||
class AutoAlignWindow(QWidget):
|
||
"""Live progress for one auto-align run, and the summary it ends with.
|
||
|
||
Shows the deviation from the good values rather than the raw levels: the
|
||
procedure is a null search, so how far off it is says more than what it
|
||
reads, and 5 mV out of ~400 mV does not show up in the raw number.
|
||
"""
|
||
|
||
abort_requested = pyqtSignal()
|
||
close_requested = pyqtSignal()
|
||
|
||
def __init__(self, parent: QWidget | None = None):
|
||
super().__init__(parent, Qt.WindowType.Window)
|
||
self.setWindowTitle("Auto-Align")
|
||
self.resize(560, 480)
|
||
self._reference = None
|
||
self._done = False
|
||
|
||
layout = QVBoxLayout(self)
|
||
|
||
self.header_label = QLabel(
|
||
f"Stepping {DEFAULT_ALIGN.offset_mm:.2f} mm either side of this "
|
||
f"point on X (T1), then on Y (T0/T2 as a pair), re-tilting until "
|
||
f"the DC levels come back to within "
|
||
f"{DEFAULT_ALIGN.tolerance_mv:.0f} mV.\n"
|
||
f"T-axes: {DEFAULT_T_AXIS.microsteps} µsteps, "
|
||
f"{DEFAULT_T_AXIS.run_current_ma} mA.")
|
||
self.header_label.setWordWrap(True)
|
||
layout.addWidget(self.header_label)
|
||
|
||
self.reading_label = QLabel("—")
|
||
self.reading_label.setFont(mono_font(13))
|
||
self.reading_label.setStyleSheet("font-weight: bold;")
|
||
layout.addWidget(self.reading_label)
|
||
|
||
self.status_label = QLabel("Configuring the rig …")
|
||
self.status_label.setWordWrap(True)
|
||
layout.addWidget(self.status_label)
|
||
|
||
# The verdict outlives the status line, which keeps reporting the
|
||
# parking moves after the answer is known.
|
||
self.verdict_label = QLabel()
|
||
self.verdict_label.setWordWrap(True)
|
||
self.verdict_label.setVisible(False)
|
||
layout.addWidget(self.verdict_label)
|
||
|
||
self.log = QPlainTextEdit(self)
|
||
self.log.setReadOnly(True)
|
||
self.log.setFont(mono_font(11))
|
||
layout.addWidget(self.log, 1)
|
||
|
||
buttons = QHBoxLayout()
|
||
self.stop_btn = QPushButton("Stop")
|
||
self.stop_btn.setToolTip(
|
||
"Stop at the next move and put the stage back on the reference "
|
||
"point. Any tilt already applied stays applied.")
|
||
self.stop_btn.clicked.connect(self._on_stop_clicked)
|
||
self.close_btn = QPushButton("Close")
|
||
self.close_btn.clicked.connect(self.close)
|
||
self.close_btn.setEnabled(False)
|
||
buttons.addWidget(self.stop_btn)
|
||
buttons.addWidget(self.close_btn)
|
||
layout.addLayout(buttons)
|
||
|
||
# ── Worker → window ───────────────────────────────────────────────────────
|
||
|
||
def set_reference(self, reading):
|
||
self._reference = reading
|
||
self._append(f"Reference: {reading.describe()}")
|
||
|
||
def on_status(self, msg: str):
|
||
self.status_label.setText(msg)
|
||
|
||
def on_reading(self, reading):
|
||
if self._reference is None:
|
||
self.reading_label.setText(reading.describe())
|
||
return
|
||
d1, d2 = reading.error_vs(self._reference)
|
||
self.reading_label.setText(
|
||
f"DC1 {reading.dc1_mv:8.1f} mV ({d1:+6.1f}) "
|
||
f"DC2 {reading.dc2_mv:8.1f} mV ({d2:+6.1f})")
|
||
|
||
def on_offset_done(self, result):
|
||
self._append(f" {result.describe()}")
|
||
|
||
def on_axis_done(self, result):
|
||
self._append(result.describe())
|
||
|
||
def on_finished(self, result):
|
||
# The per-axis lines are already in the log, put there as they
|
||
# happened; only the closing verdict is new.
|
||
self._append("")
|
||
self._append(result.verdict())
|
||
self._set_verdict(result.verdict(), ok=result.ok)
|
||
|
||
def on_failed(self, message: str):
|
||
self._append(f"FAILED: {message}")
|
||
self._set_verdict(message, ok=False)
|
||
|
||
def on_aborted(self):
|
||
message = ("Stopped by the operator. Any tilt already applied stays "
|
||
"applied.")
|
||
self._append(message)
|
||
self._set_verdict(message, ok=False)
|
||
|
||
# ── Window → worker ───────────────────────────────────────────────────────
|
||
|
||
def _on_stop_clicked(self):
|
||
self.stop_btn.setEnabled(False)
|
||
self.on_status("Stopping at the next move …")
|
||
self.abort_requested.emit()
|
||
|
||
def _set_verdict(self, text: str, ok: bool):
|
||
self.verdict_label.setText(text)
|
||
self.verdict_label.setStyleSheet(
|
||
f"font-weight: bold; color: {'green' if ok else '#b8860b'};")
|
||
self.verdict_label.setVisible(True)
|
||
self._done = True
|
||
self.stop_btn.setEnabled(False)
|
||
self.close_btn.setEnabled(True)
|
||
|
||
def _append(self, line: str):
|
||
self.log.appendPlainText(line)
|
||
|
||
def closeEvent(self, event):
|
||
# Closing part-way through is a stop: the procedure owns the stage,
|
||
# and nothing else can call it off once this window is gone.
|
||
if not self._done:
|
||
self.abort_requested.emit()
|
||
self.close_requested.emit()
|
||
super().closeEvent(event)
|
||
|
||
|
||
# ── 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._t3r_driver, self._bbd_worker,
|
||
self._oscope_worker)
|
||
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_JOG_STEP_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")
|
||
)
|
||
self._camera_win.align_active.connect(self._on_camera_align_active)
|
||
|
||
# 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:
|
||
step = float(self.lineEdit_10.text())
|
||
except ValueError:
|
||
step = BBD_JOG_STEP_MM
|
||
# The step rides along with the command: writing it onto the worker
|
||
# from here would be a cross-thread poke at a field the worker reads.
|
||
self._bbd_worker.queue_jog(axis, direction, step_mm=step)
|
||
|
||
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)
|
||
# Auto-align lives in the camera window but drives this same stage.
|
||
self._camera_win.set_align_available(enabled)
|
||
|
||
def _on_camera_align_active(self, active: bool):
|
||
"""An alignment started or finished in the camera window."""
|
||
self._set_scan_buttons_enabled(not active)
|
||
self.inspect_angles_btn.setEnabled(not active)
|
||
|
||
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"
|
||
"Each angle starts with its own background capture, so you will be "
|
||
"asked to switch the Genesis laser off and back on at every angle."
|
||
f"\n\nSave → {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 sras.version not in WRITABLE_VERSIONS:
|
||
QMessageBox.warning(
|
||
self, "Cannot Resume Scan",
|
||
f"{path.name} is a v{sras.version} file, written before each "
|
||
"angle carried its own background waveform. Every angle this "
|
||
"app acquires now writes a background block that the older "
|
||
"layout has no room for, so resuming would shift the file's "
|
||
"data. Start a new scan instead — the old file still opens in "
|
||
"the viewer."
|
||
)
|
||
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()
|