From afe33249d1f1377427e4061a39c0d1940e212e18 Mon Sep 17 00:00:00 2001 From: Thomas Ales Date: Tue, 28 Jul 2026 11:11:09 -0500 Subject: [PATCH] Phase 4: extract headless ScanEngine; de-Qt the T3R driver MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The headline of the refactor. Scan orchestration no longer lives inside a QObject that reaches through Qt workers for its hardware handles. core/scan_engine.py — ScanEngine(stage, scope, rotator, plan, out_path, resume, callbacks). Takes the concrete drivers, blocks in run(), reports via plain callables, and prompts through an injected blocking callable. No Qt import anywhere in the path (test_engine_imports_without_qt proves it), so a simpler GUI or a CLI can drive the identical acquisition. Supporting extractions, all Qt-free: - core/scope_sras.py — SCPI policy: channel profiles, trigger programming, background average, per-row FastFrame transfer - core/rotation.py — RotationAxis + RotationSettings (the GR_* constants) - core/scan_resume.py — frontier contiguity rule + settings compatibility - gui/scan_bridge.py — QtScanController, exposing exactly the signal surface the old ScanWorker had, so MainWindow's connections are unchanged hardware/t3r_driver.py is now Qt-free: a plain Signal class, a threading reader, and a polling thread instead of QObject/QThread/QTimer. gui/qt_t3r.py re-emits its callbacks as queued Qt signals for the panels. Fixes carried by the extraction: - rotation waits on the driver's MOTION_DONE event instead of time.sleep(estimate + 0.5) - abort during an operator prompt now takes effect; the old _prompt_event.wait() had no timeout and could not be interrupted - the poll timer is a thread, so an I/O error tearing down the driver no longer calls QTimer.stop() from the wrong thread - T3RDriver.disconnect() renamed close(); it shadowed QObject.disconnect() - per-frame DC means use np.frombuffer over the joined block instead of struct.unpack per frame (~16k tuple allocations per row) tests/fakes.py + test_scan_engine.py (14 tests) assert the exact command sequence, file layout, resume seeking, abort/pause, and geometry rejection before any hardware call; test_scan_resume.py covers the frontier rule. 58 passing. Co-Authored-By: Claude Opus 5 --- core/rotation.py | 87 ++++++ core/scan_engine.py | 393 +++++++++++++++++++++++++ core/scan_resume.py | 64 +++++ core/scope_sras.py | 167 +++++++++++ gui/__init__.py | 1 + gui/qt_t3r.py | 55 ++++ gui/scan_bridge.py | 96 +++++++ hardware/t3r_driver.py | 182 ++++++++---- sc3_aui_app.py | 592 ++++---------------------------------- t3r_control_panel.py | 17 +- tests/fakes.py | 185 ++++++++++++ tests/test_scan_engine.py | 282 ++++++++++++++++++ tests/test_scan_resume.py | 77 +++++ 13 files changed, 1585 insertions(+), 613 deletions(-) create mode 100644 core/rotation.py create mode 100644 core/scan_engine.py create mode 100644 core/scan_resume.py create mode 100644 core/scope_sras.py create mode 100644 gui/__init__.py create mode 100644 gui/qt_t3r.py create mode 100644 gui/scan_bridge.py create mode 100644 tests/fakes.py create mode 100644 tests/test_scan_engine.py create mode 100644 tests/test_scan_resume.py diff --git a/core/rotation.py b/core/rotation.py new file mode 100644 index 0000000..3b72e01 --- /dev/null +++ b/core/rotation.py @@ -0,0 +1,87 @@ +"""GR rotation axis: the T3R configuration and move policy for scanning. + +Qt-free façade over hardware.t3r_driver.T3RDriver that owns the drive +settings the scan depends on, so the engine never re-derives them. +""" +from __future__ import annotations + +import logging +from dataclasses import dataclass + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class RotationSettings: + """Drive settings for the GR axis during a scan.""" + microsteps: int = 8 # microsteps/full-step on the GR axis (ch3) + velocity: int = 4000 # steps/s for inter-angle moves + accel: int = 2000 # steps/s² for inter-angle moves + run_current_ma: int = 1200 # drive current while moving + hold_current_ma: int = 400 # standstill current + ihold_delay: int = 6 # run→hold current ramp delay (TMC IHOLDDELAY units) + # The sample rotates CW instead of CCW to clear wiring and avoid a stall. + rotation_sign: int = -1 + + +DEFAULT_ROTATION = RotationSettings() + + +class RotationAxis: + """Blocking rotation control for the GR axis. + + ``configure()`` must run before any move: ``steps_for_angle()`` assumes + the configured microstep setting, so the device has to be told to match + rather than trusting whatever the T3R panel or firmware default left it + at. + """ + + def __init__(self, driver, settings: RotationSettings = DEFAULT_ROTATION): + self.driver = driver + self.settings = settings + self._current_deg = 0.0 + + @property + def is_available(self) -> bool: + return self.driver is not None and self.driver.is_open + + @property + def current_deg(self) -> float: + return self._current_deg + + def configure(self) -> None: + s = self.settings + ch = self.driver.GR_AXIS_CH + self.driver.set_microstep(ch, s.microsteps) + self.driver.set_current(ch, s.run_current_ma, s.hold_current_ma, s.ihold_delay) + self.driver.enable(ch) + + def estimate_move_secs(self, delta_deg: float) -> float: + """Trapezoidal move time: cruise + accel/decel ramps.""" + s = self.settings + steps = abs(self.driver.steps_for_angle(delta_deg, s.microsteps)) + return steps / s.velocity + s.velocity / s.accel + + def rotate_to(self, angle_deg: float, timeout_margin_s: float = 5.0) -> float: + """Rotate to an absolute angle and block until the move completes. + + Returns the estimated move time (for status reporting). Waits on the + driver's MOTION_DONE event rather than sleeping for a guessed + duration; falls back to the estimate only if the event never arrives. + """ + delta_deg = angle_deg - self._current_deg + if abs(delta_deg) <= 0.001: + return 0.0 + s = self.settings + est_secs = self.estimate_move_secs(delta_deg) + self.driver.rotate_stage(delta_deg, s.microsteps, s.velocity, s.accel) + if not self.driver.wait_motion_done(self.driver.GR_AXIS_CH, + est_secs + timeout_margin_s): + logger.warning( + "GR axis did not report MOTION_DONE within %.1f s for a " + "%.1f° move; continuing", est_secs + timeout_margin_s, delta_deg) + self._current_deg = angle_deg + return est_secs + + def return_to_zero(self) -> float: + return self.rotate_to(0.0) diff --git a/core/scan_engine.py b/core/scan_engine.py new file mode 100644 index 0000000..175438e --- /dev/null +++ b/core/scan_engine.py @@ -0,0 +1,393 @@ +"""Headless SRAS scan engine. + +Takes plain hardware drivers, a ScanPlan, and callbacks — no Qt, no +widgets. ``run()`` blocks, so the caller owns the thread; GUIs wrap this +with gui.scan_bridge.QtScanController, which adapts the callbacks to Qt +signals. A CLI or a simpler GUI can drive the same engine with nothing but +functions. +""" +from __future__ import annotations + +import logging +import threading +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Callable + +from core import scope_sras +from core.rotation import RotationAxis +from core.scan_geometry import ScanPlan, validate_plan +from core.sras_format import SCAN_CHANNELS, create_scan_file + +logger = logging.getLogger(__name__) + +SCAN_VELOCITY_MM_S = 100.0 +SCAN_ACCEL_MM_S2 = 1500.0 +LASER_FREQ_HZ = 20000.0 # laser pulse frequency during data acquisition +# Theoretical ramp distance: d = v² / (2a) = 100² / (2×1500) ≈ 3.33 mm +SCAN_RAMP_MM = SCAN_VELOCITY_MM_S**2 / (2.0 * SCAN_ACCEL_MM_S2) +# Extra buffer added to both ends of the ramp. The BBD202 controller begins +# decelerating slightly before the theoretical point to avoid overshoot, +# which drops TRIGOUT_MAXV early and clips the last few data points. +SCAN_RAMP_BUFFER_MM = 1.0 + +AXIS_X = 0x21 +AXIS_Y = 0x22 + + +class ScanAborted(Exception): + """Raised inside the engine thread to unwind a scan cleanly.""" + + +@dataclass +class ResumeTarget: + """One angle selected for (re)acquisition in an existing file.""" + angle_idx: int + data_offset: int + n_rows: int + angle_deg: float + + +@dataclass +class ResumeState: + path: Path + targets: list[ResumeTarget] + samples_per_frame: int + + @property + def target_indices(self) -> set[int]: + return {t.angle_idx for t in self.targets} + + +@dataclass +class ScanCallbacks: + """Progress reporting hooks. Every one is optional.""" + on_status: Callable[[str], None] = lambda msg: None + on_started: Callable[[], None] = lambda: None + on_row_started: Callable[[int, int, int, int], None] = lambda r, nr, a, na: None + on_row_done: Callable[[int, int, int, int], None] = lambda r, nr, a, na: None + on_dc_bias: Callable[[int, list], None] = lambda row, means: None + on_paused_changed: Callable[[bool], None] = lambda paused: None + # Blocking operator prompt: must not return until acknowledged. + prompt: Callable[[str, str], None] = lambda title, msg: None + + +@dataclass +class ScanResult: + path: Path + rows_written: int = 0 + aborted: bool = False + angles_acquired: list[int] = field(default_factory=list) + + +class ScanEngine: + """Runs a full SRAS acquisition: stage, rotation, scope, and file output. + + Constructed with the concrete drivers (not worker/queue wrappers), so + any front end can reuse it:: + + engine = ScanEngine(stage, scope, rotator, plan, out_path, + callbacks=ScanCallbacks(on_status=print)) + result = engine.run() # blocking + """ + + def __init__(self, stage, scope, rotator: RotationAxis | None, + plan: ScanPlan, out_path: Path, + resume: ResumeState | None = None, + callbacks: ScanCallbacks | None = None): + self._stage = stage + self._scope = scope + self._rotator = rotator + self._plan = plan + self._out_path = Path(out_path) + self._resume = resume + self._cb = callbacks if callbacks is not None else ScanCallbacks() + + self._abort = threading.Event() + self._resume_event = threading.Event() + self._resume_event.set() # set = running, cleared = pause requested + + # ── External control (thread-safe) ──────────────────────────────────────── + + def abort(self): + self._abort.set() + self._resume_event.set() # unblock a paused scan so it can exit + + def pause(self): + """Request a pause; takes effect at the next row boundary.""" + self._resume_event.clear() + + def resume(self): + self._resume_event.set() + + @property + def aborted(self) -> bool: + return self._abort.is_set() + + # ── Internals ───────────────────────────────────────────────────────────── + + def _check_abort(self): + if self._abort.is_set(): + raise ScanAborted("Scan aborted by user.") + + def _pause_point(self): + """Block here (between rows, hardware idle) while a pause is requested.""" + if self._resume_event.is_set(): + self._check_abort() + return + self._cb.on_status( + "Scan paused — lasers may be switched off. " + "Turn lasers back on before resuming." + ) + self._cb.on_paused_changed(True) + while not self._resume_event.wait(0.2): + if self._abort.is_set(): + break + self._cb.on_paused_changed(False) + self._check_abort() + self._cb.on_status("Scan resumed.") + + def _prompt(self, title: str, message: str): + self._cb.prompt(title, message) + self._check_abort() + + # ── Main sequence ───────────────────────────────────────────────────────── + + def run(self) -> ScanResult: + """Execute the scan. Blocking; returns a ScanResult. + + Raises ScanGeometryError for an unrunnable plan, RuntimeError for + missing/mismatched hardware, or ScanAborted if the operator aborts. + """ + plan = self._plan + per_angle = plan.per_angle + n_angles = plan.n_angles + result = ScanResult(path=self._out_path) + + validate_plan(plan, SCAN_RAMP_MM, SCAN_RAMP_BUFFER_MM) + + geometry_summary = ", ".join( + f"{pa.angle_deg:.1f}°: {pa.n_rows} row(s) × {pa.n_frames} pts/row" + for pa in per_angle + ) + self._cb.on_status( + f"Scan geometry: {n_angles} angle(s), {plan.total_rows} row(s) total " + f"(per-angle bounding box) | save → {self._out_path.parent}\n" + f"{geometry_summary}" + ) + self._cb.on_started() + + if self._stage is None: + raise RuntimeError("BBD202 not connected") + if self._scope is None: + raise RuntimeError("Oscilloscope not connected") + rotator_ready = self._rotator is not None and self._rotator.is_available + if n_angles > 1 and not rotator_ready: + raise RuntimeError( + f"NumAngles={n_angles} requires the T3R rotation stage (GR-axis), " + "but it is not connected. Connect T3R from the T3R panel before " + "starting a multi-angle scan, or set NumAngles to 1." + ) + + if rotator_ready: + s = self._rotator.settings + self._cb.on_status( + f"Configuring GR axis: {s.microsteps} µsteps, " + f"{s.run_current_ma}/{s.hold_current_ma} mA run/hold …" + ) + self._rotator.configure() + time.sleep(0.2) + + self._prepare_stage() + samples_per_frame = self._prepare_scope() + scan_file = self._open_output(samples_per_frame, result) + + try: + self._scan_loop(scan_file, samples_per_frame, result) + finally: + scan_file.close() + # Return the GR axis home regardless of abort or error + if rotator_ready and abs(self._rotator.current_deg) > 0.001: + self._cb.on_status("Returning GR to home …") + try: + self._rotator.return_to_zero() + except Exception: + logger.exception("GR return-to-home failed") + + if self._abort.is_set(): + result.aborted = True + raise ScanAborted("Scan aborted by user.") + self._cb.on_status("Scan complete.") + return result + + def _prepare_stage(self): + ctrl = self._stage + self._cb.on_status("Enabling stage axes …") + if not ctrl.am_enabled[0]: + ctrl.enable_axis(AXIS_X) + if not ctrl.am_enabled[1]: + ctrl.enable_axis(AXIS_Y) + time.sleep(0.2) + + if not ctrl.am_homed[0] or not ctrl.am_homed[1]: + self._cb.on_status("Homing stage (may take up to 2 min) …") + if not ctrl.am_homed[0]: + ctrl.home_axis(AXIS_X, timeout=120.0) + if not ctrl.am_homed[1]: + ctrl.home_axis(AXIS_Y, timeout=120.0) + + self._cb.on_status("Setting scan velocity …") + ctrl.set_velocity_params(AXIS_X, max_velocity=SCAN_VELOCITY_MM_S, + acceleration=SCAN_ACCEL_MM_S2) + ctrl.set_velocity_params(AXIS_Y, max_velocity=SCAN_VELOCITY_MM_S, + acceleration=SCAN_ACCEL_MM_S2) + # X trigger: logic-high output while the stage is at maximum velocity + ctrl.set_trigger_trigout_maxv(AXIS_X) + + def _prepare_scope(self) -> int: + self._cb.on_status("Configuring oscilloscope …") + samples_per_frame = scope_sras.configure_acquisition(self._scope) + + if self._resume is None: + self._preambles = scope_sras.read_preambles(self._scope, SCAN_CHANNELS) + self._prompt( + "Background Capture", + "Please ensure the Helios laser is ON and the Genesis laser is OFF,\n" + "then click OK to capture the background waveform." + ) + self._background = scope_sras.capture_background( + self._scope, should_abort=self._abort.is_set, + on_status=self._cb.on_status) + self._prompt( + "Begin Scanning", + "Background captured successfully.\n\n" + "Please ensure the Genesis laser is back ON,\n" + "then click OK to begin scanning." + ) + else: + # Resuming: the file's existing background waveform and channel + # preambles are reused as-is (the format has no way to replace + # them without rewriting the whole file), so background capture is + # skipped. Sanity-check that this scope still produces the same + # record length the file was started with — a mismatch would + # silently corrupt the ragged per-row byte layout on append. + if samples_per_frame != self._resume.samples_per_frame: + raise RuntimeError( + f"Oscilloscope record length ({samples_per_frame} samples/frame) " + f"does not match the {self._resume.samples_per_frame} samples/frame " + f"this scan file was started with — cannot safely resume." + ) + targets = ", ".join(str(t.angle_idx + 1) for t in self._resume.targets) + self._prompt( + "Resume Scan", + f"Resuming {self._resume.path.name} — will (re)acquire " + f"angle(s) {targets} of {self._plan.n_angles}.\n\n" + "Please re-home the GR axis to 0° before continuing — the scan " + "will rotate it directly from angle to angle before scanning resumes.\n\n" + "Please ensure the Genesis laser is ON,\n" + "then click OK to continue scanning." + ) + + scope_sras.configure_scan_trigger(self._scope) + return samples_per_frame + + def _open_output(self, samples_per_frame: int, result: ScanResult): + if self._resume is not None: + result.path = self._resume.path + self._cb.on_status( + f"Resuming {self._resume.path.name} — " + f"{len(self._resume.targets)} angle(s) to (re)acquire …" + ) + return open(self._resume.path, "r+b") + return create_scan_file( + self._out_path, self._plan, samples_per_frame, + scope_sras.SAMPLE_RATE_HZ, self._preambles, self._background, + ) + + def _scan_loop(self, scan_file, samples_per_frame: int, result: ScanResult): + plan = self._plan + n_angles = plan.n_angles + x_ramp_total = SCAN_RAMP_MM + SCAN_RAMP_BUFFER_MM + ctrl = self._stage + scope = self._scope + + targets_by_ai = None + if self._resume is not None: + targets_by_ai = {t.angle_idx: t for t in self._resume.targets} + + # Both fresh and resumed scans assume the GR axis starts at home (0°) + # — the resume prompt instructs the operator to re-home it — so the + # first move always rotates directly from 0° to the starting angle. + for ai, pa in enumerate(plan.per_angle): + if targets_by_ai is not None and ai not in targets_by_ai: + continue # not selected for (re)acquisition + self._pause_point() + + if targets_by_ai is not None: + # Interior angles may already have valid data on either side, + # so seek to this angle's fixed offset rather than relying on + # the file's current position. + scan_file.seek(targets_by_ai[ai].data_offset) + + if self._rotator is not None and self._rotator.is_available: + delta = pa.angle_deg - self._rotator.current_deg + if abs(delta) > 0.001: + self._cb.on_status( + f"Rotating GR to {pa.angle_deg:.1f}° (Δ{delta:+.1f}°) …") + self._rotator.rotate_to(pa.angle_deg) + + # Each angle's bounding box gives it its own points/row count, so + # the scope's FastFrame count must be re-armed per angle. + scope.set_fastframe_count(pa.n_frames) + + for ri, y_pos in enumerate(pa.y_positions): + self._pause_point() + + self._cb.on_row_started(ri + 1, pa.n_rows, ai + 1, n_angles) + self._cb.on_status( + f"Angle {ai+1}/{n_angles} Row {ri+1}/{pa.n_rows} " + f"(Y={y_pos:.3f} mm)" + ) + + # Position the stage one ramp-length + buffer before the data + # window so it is at full velocity before x_start. + ctrl.move_axis_absolute(AXIS_Y, y_pos, timeout=60.0) + ctrl.move_axis_absolute(AXIS_X, pa.x_start - x_ramp_total, timeout=30.0) + + scope_sras.arm_row(scope) + + # Data window + ramp + buffer run-off, so the stage does not + # begin decelerating before the last point. + x_end = pa.x_start + pa.x_delta + x_ramp_total + ctrl.move_axis_absolute(AXIS_X, x_end, timeout=120.0) + + scope_sras.finish_row(scope) + self._write_row(scan_file, samples_per_frame, ri) + + result.rows_written += 1 + self._cb.on_row_done(ri + 1, pa.n_rows, ai + 1, n_angles) + + result.angles_acquired.append(ai) + + def _write_row(self, scan_file, samples_per_frame: int, row_idx: int): + """Stream every channel from the scope into the file. + + CH3 is the max-vel gate signal — no useful waveform data — so zeroed + frames are written to keep the file layout intact. + """ + scope = self._scope + for ch in SCAN_CHANNELS: + if ch == 3: + self._cb.on_status("Writing zeroed CH3 frames …") + zero_frame = bytes(samples_per_frame) + for _ in range(scope_sras.frames_acquired(scope)): + scan_file.write(zero_frame) + continue + + self._cb.on_status(f"Fetching CH{ch} data …") + waveforms = scope_sras.transfer_channel(scope, ch) + if ch == 4 and waveforms: + self._cb.on_dc_bias(row_idx + 1, scope_sras.frame_means(waveforms)) + for w in waveforms: + scan_file.write(w) diff --git a/core/scan_resume.py b/core/scan_resume.py new file mode 100644 index 0000000..53f2b27 --- /dev/null +++ b/core/scan_resume.py @@ -0,0 +1,64 @@ +"""Resume planning: turn a file's frontier into a set of angles to re-acquire. + +Pure logic, no Qt and no file I/O beyond what SrasFile already parsed, so +the non-obvious contiguity rule is testable on its own. +""" +from __future__ import annotations + +from dataclasses import dataclass, field + +from core.scan_engine import ResumeState, ResumeTarget +from core.sras_format import SrasFile + + +@dataclass +class ResumePlan: + targets: list[ResumeTarget] + auto_added: list[int] = field(default_factory=list) # indices forced in + frontier_idx: int = 0 + + @property + def total_rows(self) -> int: + return sum(t.n_rows for t in self.targets) + + def to_state(self, sras: SrasFile) -> ResumeState: + return ResumeState(path=sras.path, targets=self.targets, + samples_per_frame=sras.header.samples_per_frame) + + +def plan_resume(statuses, selected: set[int]) -> ResumePlan: + """Expand an operator's angle selection into a runnable resume plan. + + Waveform data is one contiguous append-only stream, so nothing can be + written past a gap: if the operator picks an angle at or beyond the + frontier (the first incomplete angle), every angle from the frontier up + to it must be re-acquired too. Those extras are reported in + ``auto_added`` so the UI can say so. + """ + frontier_idx = next((s.index for s in statuses if not s.complete), len(statuses)) + + at_or_past = {i for i in selected if i >= frontier_idx} + if at_or_past: + final = selected | set(range(frontier_idx, max(at_or_past) + 1)) + else: + final = set(selected) + + targets = [ + ResumeTarget(angle_idx=s.index, data_offset=s.data_offset, + n_rows=s.n_rows, angle_deg=s.angle_deg) + for s in statuses if s.index in final + ] + return ResumePlan(targets=targets, + auto_added=sorted(final - set(selected)), + frontier_idx=frontier_idx) + + +def is_compatible(sras: SrasFile, *, velocity: float, laser_freq: float, + sample_rate: float, n_channels: int) -> bool: + """Whether appending to this file with the current settings is safe.""" + h = sras.header + return (h.bytes_per_sample == 1 + and h.n_channels == n_channels + and abs(h.velocity - velocity) <= 1e-3 + and abs(h.laser_freq - laser_freq) <= 1e-3 + and abs(h.sample_rate - sample_rate) <= 1.0) diff --git a/core/scope_sras.py b/core/scope_sras.py new file mode 100644 index 0000000..30ef616 --- /dev/null +++ b/core/scope_sras.py @@ -0,0 +1,167 @@ +"""Oscilloscope SCPI policy for SRAS acquisition. + +All the Tektronix-specific instrument setup the scan depends on, in one +Qt-free place: per-channel display/coupling config, trigger programming, +the background average, and per-row FastFrame transfer. +""" +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass + +import numpy as np + +logger = logging.getLogger(__name__) + +SAMPLE_RATE_HZ = 6.25e9 # 6.25 GS/s → 160 ps/sample +TRIG_LEVEL_V = 0.500 +BACKGROUND_AVERAGES = 1024 +BACKGROUND_TIMEOUT_S = 60.0 + + +@dataclass(frozen=True) +class ChannelProfile: + """Display/input configuration for one scope channel.""" + label: str + scale_v_div: float + position_div: float + termination_ohm: int + coupling: str + bandwidth_hz: float + + +# Standard SRAS front-end configuration. +SRAS_CHANNELS = { + 1: ChannelProfile("RF Acoustic Packet", 0.07, 0.0, 50, "DC", 250e6), + 2: ChannelProfile("Trigger Signal", 0.5, -2.72, 1_000_000, "DC", 20e6), + 3: ChannelProfile("Max Vel Gate", 1.0, -2.72, 1_000_000, "DC", 20e6), + 4: ChannelProfile("Bias - B", 0.1, -2.72, 1_000_000, "DC", 20e6), +} + + +def configure_channels(scope, profiles=None) -> None: + """Apply the standard SRAS channel configuration.""" + profiles = profiles if profiles is not None else SRAS_CHANNELS + for ch, p in profiles.items(): + scope.write(f"SELect:CH{ch} ON") + scope.set_channel_label_name(ch, p.label) + scope.set_channel_scale(ch, p.scale_v_div) + scope.set_channel_position(ch, p.position_div) + scope.set_channel_termination(ch, p.termination_ohm) + scope.set_channel_coupling(ch, p.coupling) + scope.set_channel_bandwidth(ch, p.bandwidth_hz) + + +def configure_acquisition(scope) -> int: + """Program the edge trigger and timebase; returns samples per frame. + + Edge trigger on the rising edge of CH2 (laser pulse). FastFrame stays + off here so the background capture runs as a single record. + """ + scope.write("TRIGger:A:TYPe EDGE") + scope.set_trigger_source(2) + scope.set_trigger_slope("RISE") + scope.set_trigger_level(2, TRIG_LEVEL_V) + scope.set_trigger_mode("NORMAL") # wait for trigger (don't auto-sweep) + scope.set_acquire_mode("SAMPLE") + scope.set_fastframe_state(False) + scope.set_sample_rate(SAMPLE_RATE_HZ) + scope.write("HORizontal:POSition 30") # 10 % trigger offset + time.sleep(0.3) # let the timebase settle before reading back + return scope.get_record_length() + + +def read_preambles(scope, channels) -> list[str]: + """Snapshot WFMOutpre per channel (captures YMULT/YOFF/YZERO).""" + preambles = [] + for ch in channels: + scope.set_data_source(ch) + preambles.append(scope.query_wfmoutpre()) + return preambles + + +def capture_background(scope, should_abort=lambda: False, + on_status=lambda msg: None) -> bytes: + """Capture one CH1 waveform averaged over BACKGROUND_AVERAGES shots. + + The scope auto-stops after the sequence; poll ACQuire:STATE until it + does rather than assuming a duration. + """ + on_status(f"Capturing background waveform ({BACKGROUND_AVERAGES}-average) …") + scope.set_acquire_mode("AVERAGE") + scope.write(f"ACQuire:NUMAVg {BACKGROUND_AVERAGES}") + scope.write("ACQuire:STOPAfter SEQuence") + scope.set_data_source(1) + scope.write("ACQuire:STATE RUN") + + deadline = time.time() + BACKGROUND_TIMEOUT_S + while time.time() < deadline: + if should_abort(): + break + if scope.query("ACQuire:STATE?").strip() == "0": + break + time.sleep(0.25) + else: + scope.write("ACQuire:STATE STOP") + on_status("Warning: background average timed out; stopping early.") + time.sleep(0.1) + return scope.transfer_curve() + + +def configure_scan_trigger(scope) -> None: + """Switch to the scan-time logic-AND trigger (CH2 HIGH AND CH3 HIGH). + + CH3 is the BBD202 TRIGOUT_MAXV gate, so frames only accumulate while the + stage is at full scan velocity. + """ + scope.write("ACQuire:STOPAfter RUNSTop") + scope.set_acquire_mode("SAMPLE") + scope.set_fastframe_state(True) + + scope.write("TRIGger:A:TYPe LOGIc") + scope.write("TRIGger:A:LOGIc:FUNCtion AND") + scope.set_trigger_level(2, TRIG_LEVEL_V) + scope.set_trigger_level(3, TRIG_LEVEL_V) + scope.write("TRIGger:A:LOGICPattern:CH2 HIGH") + scope.write("TRIGger:A:LOGICPattern:CH3 HIGH") + time.sleep(0.2) + + +def arm_row(scope) -> None: + """Start acquisition for one scan row.""" + scope.write("ACQuire:STATE RUN") + time.sleep(0.05) + + +def finish_row(scope) -> None: + """Wait for trailing frames, then stop acquisition.""" + time.sleep(0.2) + scope.write("ACQuire:STATE STOP") + + +def frames_acquired(scope) -> int: + return int(scope.query("ACQuire:NUMFRAMESACQuired?")) + + +def transfer_channel(scope, ch: int) -> list[bytes]: + """Fetch one channel's FastFrame block as raw int8 frames.""" + scope.set_data_source(ch) + return scope.transfer_fastframe(parse=False) + + +def frame_means(waveforms: list[bytes]) -> list[float]: + """Per-frame DC mean of a raw int8 FastFrame block. + + numpy over the joined buffer: the per-frame struct.unpack this replaces + allocated a tuple of Python ints per frame (~16k frames per row). + """ + if not waveforms: + return [] + n = len(waveforms[0]) + if n == 0 or any(len(w) != n for w in waveforms): + # Ragged block (shouldn't happen) — fall back to per-frame means. + return [float(np.frombuffer(w, dtype=np.int8).mean()) if len(w) else 0.0 + for w in waveforms] + block = np.frombuffer(b"".join(waveforms), dtype=np.int8).reshape(len(waveforms), n) + return block.mean(axis=1, dtype=np.float32).tolist() diff --git a/gui/__init__.py b/gui/__init__.py new file mode 100644 index 0000000..6e98141 --- /dev/null +++ b/gui/__init__.py @@ -0,0 +1 @@ +"""Shared PyQt6 layer: adapters and widgets used by more than one app.""" diff --git a/gui/qt_t3r.py b/gui/qt_t3r.py new file mode 100644 index 0000000..cc39e67 --- /dev/null +++ b/gui/qt_t3r.py @@ -0,0 +1,55 @@ +"""Qt adapter over the (Qt-free) T3RDriver. + +The driver fires its callbacks on the reader thread. This adapter turns +each one into a Qt signal emitted from that thread; because the adapter +lives on the GUI thread, Qt queues the delivery and slots run on the GUI +thread — which is what widget code requires. + +Command methods are forwarded to the driver, so panels can hold the adapter +alone and use it exactly like the old QObject driver. +""" + +from __future__ import annotations + +from PyQt6.QtCore import QObject, pyqtSignal + +from hardware.t3r_driver import T3RDriver + + +class QtT3RAdapter(QObject): + port_opened = pyqtSignal() + handshake_ok = pyqtSignal(int, int, int) # proto_ver, fw_ver, num_channels + disconnected = pyqtSignal(str) # reason ("" = user-initiated) + info_updated = pyqtSignal(int, object) # ch, proto.Info + drv_status_updated = pyqtSignal(int, object) + position_updated = pyqtSignal(int, int) + motion_done = pyqtSignal(int, int) + stopped = pyqtSignal(int, int) + fault_occurred = pyqtSignal(int, int) + ack_received = pyqtSignal(int, int) + frame_received = pyqtSignal(int, bytes) + + _EVENTS = ("port_opened", "handshake_ok", "disconnected", "info_updated", + "drv_status_updated", "position_updated", "motion_done", + "stopped", "fault_occurred", "ack_received", "frame_received") + + def __init__(self, driver: T3RDriver | None = None, parent=None): + super().__init__(parent) + self.driver = driver if driver is not None else T3RDriver() + for name in self._EVENTS: + getattr(self.driver, name).connect(getattr(self, name).emit) + + # Class attributes (constants) the panels read off the driver + CHANNEL_NAMES = T3RDriver.CHANNEL_NAMES + GR_AXIS_CH = T3RDriver.GR_AXIS_CH + + @property + def is_open(self) -> bool: + return self.driver.is_open + + def __getattr__(self, name): + # Only reached for attributes this QObject doesn't define, i.e. the + # driver's command API (open/close/move/jog/enable/...). + if name.startswith("_"): + raise AttributeError(name) + return getattr(object.__getattribute__(self, "driver"), name) diff --git a/gui/scan_bridge.py b/gui/scan_bridge.py new file mode 100644 index 0000000..2a3aff1 --- /dev/null +++ b/gui/scan_bridge.py @@ -0,0 +1,96 @@ +"""Qt bridge over the headless ScanEngine. + +Exposes exactly the signal surface the old in-GUI ScanWorker had, so window +code connects to it unchanged, and adapts prompts/progress to Qt. The +engine itself stays Qt-free and reusable by any other front end. +""" +from __future__ import annotations + +import threading +import traceback + +from PyQt6.QtCore import QObject, pyqtSignal, pyqtSlot + +from core.scan_engine import ScanAborted, ScanCallbacks, ScanEngine + + +class QtScanController(QObject): + """Runs a ScanEngine on the caller's QThread and republishes its events. + + Move this to a QThread and connect `started` → `run`, exactly like the + previous ScanWorker. + """ + + started = pyqtSignal() + completed = pyqtSignal() + dc_bias_updated = pyqtSignal(int, object) # row index, per-frame DC means + failed = pyqtSignal(str) + row_started = pyqtSignal(int, int, int, int) # row, n_rows, angle_idx, n_angles + row_done = pyqtSignal(int, int, int, int) + status_msg = pyqtSignal(str) + user_prompt = pyqtSignal(str, str) # title, message + paused_changed = pyqtSignal(bool) # True while paused at a row boundary + + def __init__(self, stage, scope, rotator, plan, out_path, + resume=None, on_scan_active=None): + super().__init__() + self._prompt_event = threading.Event() + self._on_scan_active = on_scan_active + + callbacks = ScanCallbacks( + on_status=self.status_msg.emit, + on_started=self.started.emit, + on_row_started=self.row_started.emit, + on_row_done=self.row_done.emit, + on_dc_bias=self.dc_bias_updated.emit, + on_paused_changed=self.paused_changed.emit, + prompt=self._blocking_prompt, + ) + self._engine = ScanEngine(stage, scope, rotator, plan, out_path, + resume=resume, callbacks=callbacks) + + # ── Engine control (called from the GUI thread) ─────────────────────────── + + def abort(self): + self._engine.abort() + self._prompt_event.set() # release a scan parked on a prompt + + def pause(self): + self._engine.pause() + + def resume(self): + self._engine.resume() + + def acknowledge_prompt(self): + """Called from the GUI thread when the operator dismisses a prompt.""" + self._prompt_event.set() + + # ── Callback plumbing ───────────────────────────────────────────────────── + + def _blocking_prompt(self, title: str, message: str): + """Ask the GUI thread, then block the scan thread until answered. + + Polls rather than waiting forever so an abort during a prompt takes + effect immediately instead of deadlocking the scan thread. + """ + self._prompt_event.clear() + self.user_prompt.emit(title, message) + while not self._prompt_event.wait(0.2): + if self._engine.aborted: + return + + @pyqtSlot() + def run(self): + if self._on_scan_active is not None: + self._on_scan_active(True) + try: + self._engine.run() + self.completed.emit() + except ScanAborted as exc: + self.failed.emit(str(exc)) + except Exception as exc: + traceback.print_exc() + self.failed.emit(str(exc)) + finally: + if self._on_scan_active is not None: + self._on_scan_active(False) diff --git a/hardware/t3r_driver.py b/hardware/t3r_driver.py index e346301..25609fa 100755 --- a/hardware/t3r_driver.py +++ b/hardware/t3r_driver.py @@ -1,8 +1,12 @@ """T3R Stepper Controller driver for ScanEngine-3. -Qt-based driver that owns the serial connection and an internal reader QThread. -All events arrive as Qt signals; all commands are fire-and-forget writes. -Create in the main (GUI) thread; no additional thread management required. +Owns the serial connection and an internal reader thread. Events are +delivered as plain-Python callbacks (see ``Signal``); commands are +fire-and-forget writes. No Qt — GUIs wrap this with gui.qt_t3r.QtT3RAdapter, +which re-emits every event as a queued Qt signal on the GUI thread. + +Callbacks run on the reader thread. Keep them short, and never touch Qt +widgets from one directly. Gear train (stage rotation via GR-axis, ch3): Motor → 10T pinion → 30T idler → 125T index gear (stage) @@ -11,23 +15,53 @@ Gear train (stage rotation via GR-axis, ch3): from __future__ import annotations +import logging import threading -import serial -from PyQt6.QtCore import QObject, QThread, QTimer, pyqtSignal - from . import t3r_protocol as proto +from .serial_util import open_8n1 + +logger = logging.getLogger(__name__) -class _T3RReader(QThread): - """Blocking read loop — runs on its own QThread.""" +class Signal: + """Minimal observer slot: ``connect(fn)`` then ``emit(*args)``. - frame = pyqtSignal(int, bytes) # (cmd, payload) for each valid frame - finished_reason = pyqtSignal(str) # "" = clean stop, else I/O error string + Mirrors the pyqtSignal API used by the existing panels so the same call + sites work against either this driver or a Qt adapter over it. A raising + subscriber is logged and skipped so one bad listener cannot kill the + reader thread. + """ - def __init__(self, ser: serial.Serial): - super().__init__() + __slots__ = ("_subs", "_name") + + def __init__(self, name: str = ""): + self._subs: list = [] + self._name = name + + def connect(self, fn) -> None: + self._subs.append(fn) + + def disconnect(self, fn) -> None: + if fn in self._subs: + self._subs.remove(fn) + + def emit(self, *args) -> None: + for fn in list(self._subs): + try: + fn(*args) + except Exception: + logger.exception("T3R %s subscriber failed", self._name) + + +class _Reader(threading.Thread): + """Blocking read loop — decodes frames and hands them to `on_frame`.""" + + def __init__(self, ser, on_frame, on_finished): + super().__init__(daemon=True, name="T3RReader") self._ser = ser + self._on_frame = on_frame + self._on_finished = on_finished self._running = True self._parser = proto.FrameParser() @@ -44,20 +78,20 @@ class _T3RReader(QThread): break if data: for cmd, payload in self._parser.feed(data): - self.frame.emit(cmd, payload) - self.finished_reason.emit(reason) + self._on_frame(cmd, payload) + self._on_finished(reason) -class T3RDriver(QObject): - """Qt-based driver for the T3R four-channel stepper controller. +class T3RDriver: + """Driver for the T3R four-channel stepper controller. Usage:: driver = T3RDriver() - driver.handshake_ok.connect(lambda pv, fw, nc: print("connected")) - driver.info_updated.connect(on_info) - driver.connect("/dev/ttyUSB0") + driver.handshake_ok.connect(lambda pv, fw, nc: ...) + driver.open("/dev/ttyUSB0") driver.move(0, steps=3200, velocity=8000, accel=4000) + driver.wait_motion_done(0, timeout=30.0) """ CHANNEL_NAMES = ["T-axis (focus)", "Axis 1", "Axis 2", "GR-axis"] @@ -68,32 +102,33 @@ class T3RDriver(QObject): GEAR_TEETH_MOTOR = 10 GEAR_TEETH_STAGE = 125 # idler is 30T but does not change ratio - # ── Signals ─────────────────────────────────────────────────────────────── + POLL_INTERVAL_S = 0.25 - port_opened = pyqtSignal() # serial port open; PING sent - handshake_ok = pyqtSignal(int, int, int) # proto_ver, fw_ver, num_channels - disconnected = pyqtSignal(str) # reason ("" = user-initiated) - - info_updated = pyqtSignal(int, object) # ch, proto.Info - drv_status_updated = pyqtSignal(int, object) # ch, proto.DrvStatus - position_updated = pyqtSignal(int, int) # ch, position (microsteps) - motion_done = pyqtSignal(int, int) # ch, final_position - stopped = pyqtSignal(int, int) # ch, final_position - fault_occurred = pyqtSignal(int, int) # ch, fault_mask - ack_received = pyqtSignal(int, int) # req_cmd, status (0=OK) - frame_received = pyqtSignal(int, bytes) # raw (cmd, payload) for log - - def __init__(self, parent=None): - super().__init__(parent) - self._ser: serial.Serial | None = None - self._reader: _T3RReader | None = None + def __init__(self): + self._ser = None + self._reader: _Reader | None = None self._write_lock = threading.Lock() self._tearing_down = False self._is_open = False - self._poll_timer = QTimer(self) - self._poll_timer.setInterval(250) - self._poll_timer.timeout.connect(self._poll) + self._poll_stop = threading.Event() + self._poll_thread: threading.Thread | None = None + + # Per-channel motion-completion events, so a caller can block on a + # move finishing instead of guessing its duration. + self._motion_events = [threading.Event() for _ in range(proto.NUM_CHANNELS)] + + self.port_opened = Signal("port_opened") # () + self.handshake_ok = Signal("handshake_ok") # proto_ver, fw_ver, n_ch + self.disconnected = Signal("disconnected") # reason ("" = user) + self.info_updated = Signal("info_updated") # ch, proto.Info + self.drv_status_updated = Signal("drv_status_updated") # ch, proto.DrvStatus + self.position_updated = Signal("position_updated") # ch, position + self.motion_done = Signal("motion_done") # ch, final_position + self.stopped = Signal("stopped") # ch, final_position + self.fault_occurred = Signal("fault_occurred") # ch, fault_mask + self.ack_received = Signal("ack_received") # req_cmd, status + self.frame_received = Signal("frame_received") # raw cmd, payload # ── Connection ──────────────────────────────────────────────────────────── @@ -101,26 +136,24 @@ class T3RDriver(QObject): def is_open(self) -> bool: return self._is_open - def connect(self, port: str, baud: int = 115200) -> None: - """Open the serial port and start the reader. Emits port_opened on success.""" + def open(self, port: str, baud: int = 115200) -> None: + """Open the serial port and start the reader; emits port_opened.""" if self._is_open: - self.disconnect() + self.close() try: - self._ser = serial.Serial(port, baudrate=baud, timeout=0.05) + self._ser = open_8n1(port, baudrate=baud, timeout=0.05) except Exception as exc: raise RuntimeError(f"Cannot open {port}: {exc}") from exc self._tearing_down = False self._is_open = True - self._reader = _T3RReader(self._ser) - self._reader.frame.connect(self._on_frame) - self._reader.finished_reason.connect(self._on_reader_finished) + self._reader = _Reader(self._ser, self._on_frame, self._on_reader_finished) self._reader.start() self.port_opened.emit() self.send_frame(proto.ping()) # handshake; polling starts on PONG - def disconnect(self) -> None: - """Close port and stop polling.""" + def close(self) -> None: + """Close the port and stop polling.""" self._teardown("") def _on_reader_finished(self, reason: str): @@ -131,16 +164,21 @@ class T3RDriver(QObject): if self._tearing_down or not self._is_open: return self._tearing_down = True - self._poll_timer.stop() + self.stop_polling() self._is_open = False + # Release anyone blocked in wait_motion_done so a disconnect during a + # move raises there instead of hanging until the timeout. + for ev in self._motion_events: + ev.set() + reader, self._reader = self._reader, None ser, self._ser = self._ser, None if reader is not None: reader.stop() - if QThread.currentThread() is not reader: - reader.wait(1000) + if threading.current_thread() is not reader: + reader.join(1.0) if ser is not None: try: ser.close() @@ -186,6 +224,7 @@ class T3RDriver(QObject): self.send_frame(proto.set_current(ch, run_ma, hold_ma, ihold_delay)) def move(self, ch: int, steps: int, velocity: int, accel: int): + self._motion_events[ch].clear() self.send_frame(proto.move(ch, steps, velocity, accel)) def jog(self, ch: int, velocity: int, accel: int): @@ -215,30 +254,44 @@ class T3RDriver(QObject): # ── Rotation helpers ────────────────────────────────────────────────────── def steps_for_angle(self, angle_deg: float, microsteps: int) -> int: - """Compute GR-axis microsteps needed to rotate the stage by angle_deg.""" + """GR-axis microsteps needed to rotate the stage by angle_deg.""" gear_ratio = self.GEAR_TEETH_STAGE / self.GEAR_TEETH_MOTOR steps_per_stage_rev = self.MOTOR_FULL_STEPS_PER_REV * microsteps * gear_ratio return round(steps_per_stage_rev * angle_deg / 360.0) def rotate_stage(self, angle_deg: float, microsteps: int, velocity: int = 8000, accel: int = 4000): - """Move GR-axis by the number of steps that rotate the stage by angle_deg.""" - steps = self.steps_for_angle(angle_deg, microsteps) - self.move(self.GR_AXIS_CH, steps, velocity, accel) + """Move GR-axis by the steps that rotate the stage by angle_deg.""" + self.move(self.GR_AXIS_CH, self.steps_for_angle(angle_deg, microsteps), + velocity, accel) + + def wait_motion_done(self, ch: int, timeout: float) -> bool: + """Block until the channel reports MOTION_DONE. False on timeout. + + Cleared by ``move()``, set by the MOTION_DONE event and by teardown, + so a disconnect mid-move unblocks immediately. + """ + return self._motion_events[ch].wait(timeout) # ── Polling ─────────────────────────────────────────────────────────────── def start_polling(self): - self._poll_timer.start() + if self._poll_thread is not None and self._poll_thread.is_alive(): + return + self._poll_stop.clear() + self._poll_thread = threading.Thread(target=self._poll_loop, daemon=True, + name="T3RPoll") + self._poll_thread.start() def stop_polling(self): - self._poll_timer.stop() + self._poll_stop.set() - def _poll(self): - if not self._is_open: - return - for ch in range(proto.NUM_CHANNELS): - self.send_frame(proto.get_info(ch)) + def _poll_loop(self): + while not self._poll_stop.wait(self.POLL_INTERVAL_S): + if not self._is_open: + break + for ch in range(proto.NUM_CHANNELS): + self.send_frame(proto.get_info(ch)) # ── Frame dispatcher ────────────────────────────────────────────────────── @@ -274,14 +327,17 @@ class T3RDriver(QObject): elif cmd == proto.EVT_MOTION_DONE: ev = proto.decode_event_position(payload) if ev and 0 <= ev.ch < proto.NUM_CHANNELS: + self._motion_events[ev.ch].set() self.motion_done.emit(ev.ch, ev.position) elif cmd == proto.EVT_STOPPED: ev = proto.decode_event_position(payload) if ev and 0 <= ev.ch < proto.NUM_CHANNELS: + self._motion_events[ev.ch].set() self.stopped.emit(ev.ch, ev.position) elif cmd == proto.EVT_FAULT: ev = proto.decode_fault(payload) if ev and 0 <= ev.ch < proto.NUM_CHANNELS: + self._motion_events[ev.ch].set() self.fault_occurred.emit(ev.ch, ev.position) diff --git a/sc3_aui_app.py b/sc3_aui_app.py index 2a04b43..01b0c94 100755 --- a/sc3_aui_app.py +++ b/sc3_aui_app.py @@ -10,7 +10,6 @@ import struct import sys import time from pathlib import Path -import threading from threading import Thread import numpy as np @@ -29,13 +28,19 @@ ROOT = Path(__file__).parent sys.path.insert(0, str(ROOT)) from core.config import ScanDefaults -from core.scan_geometry import ScanPlan, EtaEstimator, build_plan, format_eta, validate_plan -from core.sras_format import ( - SCAN_CHANNELS, SrasFile, create_scan_file, plan_from_header, +from core.rotation import DEFAULT_ROTATION, RotationAxis +from core.scan_engine import ( + ResumeState, + LASER_FREQ_HZ, SCAN_VELOCITY_MM_S, ) +from core.scan_geometry import ScanPlan, EtaEstimator, build_plan, format_eta +from core.scan_resume import is_compatible, plan_resume +from core.scope_sras import SAMPLE_RATE_HZ, configure_channels +from core.sras_format import SCAN_CHANNELS, SrasFile, plan_from_header +from gui.qt_t3r import QtT3RAdapter +from gui.scan_bridge import QtScanController from hardware.helios_laser import HeliosLaser from hardware.pybbd202 import AXIS_X, AXIS_Y, ThorlabsServoDriver -from hardware.t3r_driver import T3RDriver from hardware.tektronix_base import TektronixOscilloscopeBase from hardware.uc480_camera import ( CameraStreamThread, UC480Camera, find_camera_bus_conflicts, @@ -44,31 +49,6 @@ from t3r_control_panel import T3RControlPanel DEFAULTS = ScanDefaults.load() -# ── Scan constants ─────────────────────────────────────────────────────────── - -SCAN_VELOCITY_MM_S = 100.0 -SCAN_ACCEL_MM_S2 = 1500.0 -LASER_FREQ_HZ = 20000.0 # laser pulse frequency during data acquisition -# Theoretical ramp distance: d = v² / (2a) = 100² / (2×1500) ≈ 3.33 mm -SCAN_RAMP_MM = SCAN_VELOCITY_MM_S**2 / (2.0 * SCAN_ACCEL_MM_S2) -# Extra buffer added to both ends of the ramp. The BBD202 controller begins -# decelerating slightly before the theoretical point to avoid overshoot, which -# causes TRIGOUT_MAXV to drop early and clip the last few data points. -# 1 mm at 100 mm/s and 2000 Hz corresponds to 20 extra trigger windows. -SCAN_RAMP_BUFFER_MM = 1.0 -SCOPE_SAMPLE_RATE = 6.25e9 # 6.25 GS/s → 160 ps/sample -SCOPE_TRIG_LEVEL_V = 0.500 - -# GR-axis rotation defaults (used by ScanWorker between angles) -GR_MICROSTEPS = 8 # microsteps/full-step on GR axis (ch3) -GR_MOVE_VELOCITY = 4000 # steps/s for inter-angle moves -GR_MOVE_ACCEL = 2000 # steps/s² for inter-angle moves -GR_RUN_CURRENT_MA = 1200 # drive current while moving -GR_HOLD_CURRENT_MA = 400 # standstill current -GR_IHOLD_DELAY = 6 # run→hold current ramp delay (TMC IHOLDDELAY units) -# Sample rotates CW instead of CCW to clear wiring and avoid a stall condition. -GR_ROTATION_SIGN = -1 - BBD_DEFAULT_JOG_MM = 0.5 # default jog step for BBD202 @@ -317,52 +297,12 @@ class OscopeWorker(QObject): try: self.scope = TektronixOscilloscopeBase(resource_name=ip, port=4000, timeout=10.0) self.scope.connect() - self._configure_channels() + configure_channels(self.scope) self.is_connected = True self.connected.emit() except Exception as e: self.connection_failed.emit(str(e)) - def _configure_channels(self): - """Apply standard SRAS channel configuration after connecting.""" - s = self.scope - - # Turn on all four channels - for ch in (1, 2, 3, 4): - s.write(f"SELect:CH{ch} ON") - - # ── CH1 — RF Acoustic Packet ────────────────────────────────────────── - s.set_channel_label_name(1, "RF Acoustic Packet") - s.set_channel_scale(1, 0.07) # 70 mV/div - s.set_channel_position(1, 0.0) # 0 divs - s.set_channel_termination(1, 50) # 50 Ohm - s.set_channel_coupling(1, "DC") - s.set_channel_bandwidth(1, 250E6) # 250Mhz Low-Pass - - # ── CH2 — Trigger Signal ────────────────────────────────────────────── - s.set_channel_label_name(2, "Trigger Signal") - s.set_channel_scale(2, 0.5) # 500 mV/div - s.set_channel_position(2, -2.72) # -2.72 divs - s.set_channel_termination(2, 1000000) # 50 Ohm - s.set_channel_coupling(2, "DC") - s.set_channel_bandwidth(2, 20E6) # 20 MHz - - # ── CH3 — Max Velocity Gate ─────────────────────────────────────────── - s.set_channel_label_name(3, "Max Vel Gate") - s.set_channel_scale(3, 1.0) # 1 V/div - s.set_channel_position(3, -2.72) # -2.72 divs - s.set_channel_termination(3, 1000000) # 1 MOhm - s.set_channel_coupling(3, "DC") - s.set_channel_bandwidth(3, 20E6) # 20 MHz - - # ── CH4 — Bias B ────────────────────────────────────────────────────── - s.set_channel_label_name(4, "Bias - B") - s.set_channel_scale(4, 0.1) # 100 mV/div - s.set_channel_position(4, -2.72) # -2.72 divs - s.set_channel_termination(4, 1000000) # 1 Mohm - s.set_channel_coupling(4, "DC") - s.set_channel_bandwidth(4, 20E6) # 20 MHz - def _do_disconnect(self): if self.scope: try: @@ -968,424 +908,6 @@ class ScanProgressWindow(QWidget): # ── Scan worker ─────────────────────────────────────────────────────────────── -class ScanWorker(QObject): - """Runs the full SRAS scan sequence in a background thread. - - Accesses hardware drivers directly (not through worker queues) so it can - make blocking calls. BBD202Worker.scanning_active is set to True for the - duration to suppress position polling in the BBD worker thread. - """ - started = pyqtSignal() - completed = pyqtSignal() - dc_bias_updated = pyqtSignal(int, object) # row index, list[float] per-frame DC means - failed = pyqtSignal(str) - row_started = pyqtSignal(int, int, int, int) # row, n_rows, angle_idx, n_angles - row_done = pyqtSignal(int, int, int, int) # row, n_rows, angle_idx, n_angles - status_msg = pyqtSignal(str) - user_prompt = pyqtSignal(str, str) # title, message - paused_changed = pyqtSignal(bool) # True while paused at a row boundary - - def __init__(self, bbd: BBD202Worker, t3r: T3RDriver | None, - oscope: OscopeWorker, plan: ScanPlan, prefix: str, - save_dir: str, resume_info: dict | None = None): - super().__init__() - self._bbd = bbd - self._t3r = t3r - self._oscope = oscope - self._plan = plan - self._prefix = prefix - self._save_dir = save_dir - self._resume_info = resume_info - self._abort = False - self._prompt_event = threading.Event() - self._resume_event = threading.Event() - self._resume_event.set() # set = running, cleared = pause requested - - def abort(self): - self._abort = True - self._resume_event.set() # unblock a paused scan so it can exit - - def pause(self): - """Request a pause; takes effect at the next row boundary.""" - self._resume_event.clear() - - def resume(self): - self._resume_event.set() - - def _pause_point(self): - """Block here (between rows, hardware idle) while a pause is requested.""" - if self._resume_event.is_set() or self._abort: - return - self.status_msg.emit( - "Scan paused — lasers may be switched off. " - "Turn lasers back on before resuming." - ) - self.paused_changed.emit(True) - while not self._resume_event.wait(0.2): - if self._abort: - break - self.paused_changed.emit(False) - if not self._abort: - self.status_msg.emit("Scan resumed.") - - def acknowledge_prompt(self): - """Called from UI thread when user clicks OK on a prompt dialog.""" - self._prompt_event.set() - - def _request_user_prompt(self, title: str, message: str): - """Emit a prompt signal and block until the UI thread acknowledges it.""" - self._prompt_event.clear() - self.user_prompt.emit(title, message) - self._prompt_event.wait() - - @pyqtSlot() - def run(self): - try: - self._run_scan() - except Exception as e: - import traceback - traceback.print_exc() - self.failed.emit(str(e)) - - def _run_scan(self): - plan = self._plan - per_angle = plan.per_angle - n_angles = plan.n_angles - save_dir = Path(self._save_dir) - - # The actual X move starts one ramp-length + buffer before x_start and - # ends one ramp-length + buffer after x_start + x_delta, so the stage - # is at full velocity across the whole data window. - _x_ramp_total = SCAN_RAMP_MM + SCAN_RAMP_BUFFER_MM - validate_plan(plan, SCAN_RAMP_MM, SCAN_RAMP_BUFFER_MM) - - geometry_summary = ", ".join( - f"{pa.angle_deg:.1f}°: {pa.n_rows} row(s) × {pa.n_frames} pts/row" - for pa in per_angle - ) - self.status_msg.emit( - f"Scan geometry: {n_angles} angle(s), {plan.total_rows} row(s) total " - f"(per-angle bounding box) | save → {save_dir}\n{geometry_summary}" - ) - self.started.emit() - - # ── Validate hardware ───────────────────────────────────────────────── - ctrl = self._bbd.controller - scope = self._oscope.scope - if ctrl is None: - raise RuntimeError("BBD202 not connected") - if scope is None: - raise RuntimeError("Oscilloscope not connected") - if n_angles > 1 and (self._t3r is None or not self._t3r.is_open): - raise RuntimeError( - f"NumAngles={n_angles} requires the T3R rotation stage (GR-axis), " - "but it is not connected. Connect T3R from the T3R panel before " - "starting a multi-angle scan, or set NumAngles to 1." - ) - - # ── Configure GR rotation axis ──────────────────────────────────────── - # steps_for_angle() assumes GR_MICROSTEPS, so the device must be set to - # match — never rely on whatever the T3R panel (or firmware default) - # last left it at. Also set drive current and energise the channel. - if self._t3r is not None and self._t3r.is_open: - self.status_msg.emit( - f"Configuring GR axis: {GR_MICROSTEPS} µsteps, " - f"{GR_RUN_CURRENT_MA}/{GR_HOLD_CURRENT_MA} mA run/hold …" - ) - gr_ch = self._t3r.GR_AXIS_CH - self._t3r.set_microstep(gr_ch, GR_MICROSTEPS) - self._t3r.set_current(gr_ch, GR_RUN_CURRENT_MA, - GR_HOLD_CURRENT_MA, GR_IHOLD_DELAY) - self._t3r.enable(gr_ch) - time.sleep(0.2) - - # ── Prepare stage ───────────────────────────────────────────────────── - self.status_msg.emit("Enabling stage axes …") - if not ctrl.am_enabled[0]: - ctrl.enable_axis(AXIS_X) - if not ctrl.am_enabled[1]: - ctrl.enable_axis(AXIS_Y) - time.sleep(0.2) - - if not ctrl.am_homed[0] or not ctrl.am_homed[1]: - self.status_msg.emit("Homing stage (may take up to 2 min) …") - if not ctrl.am_homed[0]: - ctrl.home_axis(AXIS_X, timeout=120.0) - if not ctrl.am_homed[1]: - ctrl.home_axis(AXIS_Y, timeout=120.0) - - self.status_msg.emit("Setting scan velocity …") - ctrl.set_velocity_params(AXIS_X, max_velocity=SCAN_VELOCITY_MM_S, - acceleration=SCAN_ACCEL_MM_S2) - ctrl.set_velocity_params(AXIS_Y, max_velocity=SCAN_VELOCITY_MM_S, - acceleration=SCAN_ACCEL_MM_S2) - - # X trigger: logic-high output when stage is at maximum velocity - ctrl.set_trigger_trigout_maxv(AXIS_X) - - # ── Configure oscilloscope ──────────────────────────────────────────── - self.status_msg.emit("Configuring oscilloscope …") - # Edge trigger: rising edge of CH2 (laser pulse) at 1.0 V. - scope.write("TRIGger:A:TYPe EDGE") - scope.set_trigger_source(2) - scope.set_trigger_slope("RISE") - scope.set_trigger_level(2, SCOPE_TRIG_LEVEL_V) - scope.set_trigger_mode("NORMAL") # wait for trigger (don't auto-sweep) - scope.set_acquire_mode("SAMPLE") - scope.set_fastframe_state(False) # start non-FF for background capture - scope.set_sample_rate(SCOPE_SAMPLE_RATE) - scope.write("HORizontal:POSition 30") # 10 % trigger offset - time.sleep(0.3) - samples_per_frame = scope.get_record_length() - - resume_info = self._resume_info - if resume_info is None: - # ── Snapshot WFMOutpre for each channel (captures YMULT/YOFF/YZERO) ── - preambles = [] - for ch in SCAN_CHANNELS: - scope.set_data_source(ch) - preambles.append(scope.query_wfmoutpre()) - - # ── Background subtraction capture ────────────────────────────────── - # Prompt user to ensure Helios is ON and Genesis is OFF. - self._request_user_prompt( - "Background Capture", - "Please ensure the Helios laser is ON and the Genesis laser is OFF,\n" - "then click OK to capture the background waveform." - ) - if self._abort: - self.failed.emit("Scan aborted by user.") - return - - # Capture a single averaged CH1 frame (1024 waveforms averaged). - self.status_msg.emit("Capturing background waveform (1024-average) …") - scope.set_acquire_mode("AVERAGE") - scope.write("ACQuire:NUMAVg 1024") - scope.write("ACQuire:STOPAfter SEQuence") # auto-stop after all 1024 averages - scope.set_data_source(1) - scope.write("ACQuire:STATE RUN") - # Poll until the scope finishes all 1024 averages and auto-stops. - # Timeout: 1024 averages at 2 kHz (worst case) = ~0.5 s; allow 60 s. - _bg_deadline = time.time() + 60.0 - while time.time() < _bg_deadline: - if self._abort: - break - if scope.query("ACQuire:STATE?").strip() == "0": - break - time.sleep(0.25) - else: - scope.write("ACQuire:STATE STOP") - self.status_msg.emit("Warning: background average timed out; stopping early.") - time.sleep(0.1) - background_waveform = scope.transfer_curve() - - # Prompt user to turn Genesis back on before the actual scan. - self._request_user_prompt( - "Begin Scanning", - "Background captured successfully.\n\n" - "Please ensure the Genesis laser is back ON,\n" - "then click OK to begin scanning." - ) - if self._abort: - self.failed.emit("Scan aborted by user.") - return - else: - # Resuming an existing file: the background waveform and channel - # preambles it already contains are reused as-is (the format has - # no way to replace them without rewriting the whole file), so - # background capture is skipped entirely. Sanity-check that this - # scope is still producing the same record length the file was - # started with — a mismatch would silently corrupt the ragged - # per-row byte layout on append. - if samples_per_frame != resume_info["samples_per_frame"]: - raise RuntimeError( - f"Oscilloscope record length ({samples_per_frame} samples/frame) " - f"does not match the {resume_info['samples_per_frame']} samples/frame " - f"this scan file was started with — cannot safely resume." - ) - target_angles = ", ".join(str(t["ai"] + 1) for t in resume_info["targets"]) - self._request_user_prompt( - "Resume Scan", - f"Resuming {resume_info['path'].name} — will (re)acquire " - f"angle(s) {target_angles} of {n_angles}.\n\n" - "Please re-home the GR axis to 0° before continuing — the scan " - "will rotate it directly from angle to angle before scanning resumes.\n\n" - "Please ensure the Genesis laser is ON,\n" - "then click OK to continue scanning." - ) - if self._abort: - self.failed.emit("Scan aborted by user.") - return - - # Restore SAMPLE mode and FastFrame for the actual scan. FastFrame - # count (points/row) is set per-angle in the scan loop below, since - # it depends on that angle's bounding-box X extent. - scope.write("ACQuire:STOPAfter RUNSTop") - scope.set_acquire_mode("SAMPLE") - scope.set_fastframe_state(True) - - # Restore logic-AND trigger (CH2 HIGH AND CH3 HIGH) for the scan loop: - # CH3 is the BBD202 TRIGOUT_MAXV gate, so frames only accumulate while - # the stage is at full scan velocity. - scope.write("TRIGger:A:TYPe LOGIc") - scope.write("TRIGger:A:LOGIc:FUNCtion AND") - scope.set_trigger_level(2, SCOPE_TRIG_LEVEL_V) - scope.set_trigger_level(3, SCOPE_TRIG_LEVEL_V) - scope.write("TRIGger:A:LOGICPattern:CH2 HIGH") - scope.write("TRIGger:A:LOGICPattern:CH3 HIGH") - time.sleep(0.2) - - # ── Open output file (entire scan in one .sras) ───────────────────────── - if resume_info is not None: - targets_by_ai = {t["ai"]: t for t in resume_info["targets"]} - scan_file = open(resume_info["path"], "r+b") - self.status_msg.emit( - f"Resuming {resume_info['path'].name} — " - f"{len(targets_by_ai)} angle(s) to (re)acquire …" - ) - else: - targets_by_ai = None - fname = save_dir / f"{self._prefix}.sras" - scan_file = create_scan_file( - fname, plan, samples_per_frame, SCOPE_SAMPLE_RATE, - preambles, background_waveform, - ) - - # ── Scan loop ───────────────────────────────────────────────────────── - self._bbd.scanning_active = True - # Both fresh and resumed scans assume the GR axis starts at home (0°) - # — the resume prompt instructs the operator to re-home it before - # continuing — so the first move always rotates directly from 0° - # to the starting angle. - prev_gr_deg = 0.0 - try: - for ai, pa in enumerate(per_angle): - if targets_by_ai is not None and ai not in targets_by_ai: - continue # not selected for (re)acquisition - self._pause_point() - if self._abort: - break - - angle = pa.angle_deg - a_x_start = pa.x_start - a_x_delta = pa.x_delta - a_n_rows = pa.n_rows - a_n_frames = pa.n_frames - y_positions = pa.y_positions - - if targets_by_ai is not None: - # Interior angles may already have valid data on either - # side of them, so seek to this angle's fixed offset - # rather than relying on the file's current position. - scan_file.seek(targets_by_ai[ai]["data_offset"]) - - # ── Rotate GR ───────────────────────────────────────────────── - if self._t3r is not None and self._t3r.is_open: - delta_deg = angle - prev_gr_deg - if abs(delta_deg) > 0.001: - steps = self._t3r.steps_for_angle(delta_deg, GR_MICROSTEPS) - # Trapezoidal move time: cruise + accel/decel ramps - est_secs = (abs(steps) / GR_MOVE_VELOCITY - + GR_MOVE_VELOCITY / GR_MOVE_ACCEL) - self.status_msg.emit( - f"Rotating GR to {angle:.1f}° (Δ{delta_deg:+.1f}°, " - f"≈{est_secs:.1f} s) …" - ) - self._t3r.rotate_stage(delta_deg, GR_MICROSTEPS, - GR_MOVE_VELOCITY, GR_MOVE_ACCEL) - time.sleep(est_secs + 0.5) - prev_gr_deg = angle - - # This angle's bounding box gives it its own points/row count, - # so the scope's FastFrame count must be re-armed per angle. - scope.set_fastframe_count(a_n_frames) - - # ── Row loop ────────────────────────────────────────────────── - for ri, y_pos in enumerate(y_positions): - self._pause_point() - if self._abort: - break - - self.row_started.emit(ri + 1, a_n_rows, ai + 1, n_angles) - self.status_msg.emit( - f"Angle {ai+1}/{n_angles} Row {ri+1}/{a_n_rows} " - f"(Y={y_pos:.3f} mm)" - ) - - # Position stage one ramp-length + buffer before the data - # window so the stage is at full velocity before x_start. - ctrl.move_axis_absolute(AXIS_Y, y_pos, timeout=60.0) - ctrl.move_axis_absolute(AXIS_X, a_x_start - _x_ramp_total, timeout=30.0) - - # Arm oscilloscope — trigger is gated with TRIGOUT_MAXV so - # frames only accumulate once the stage reaches full velocity - scope.write("ACQuire:STATE RUN") - time.sleep(0.05) - - # Execute scan move: data window + ramp + buffer run-off so - # the stage does not begin decelerating before the last point. - x_end = a_x_start + a_x_delta + _x_ramp_total - ctrl.move_axis_absolute(AXIS_X, x_end, timeout=120.0) - - # Brief settle: wait for trailing frames then stop acquisition - time.sleep(0.2) - scope.write("ACQuire:STATE STOP") - - # Stream all channels from oscilloscope and append to file. - # CH3 is the max-vel gate signal — no useful waveform data, - # so write zeroed frames to keep the file format intact. - for ch in SCAN_CHANNELS: - if ch == 3: - self.status_msg.emit("Writing zeroed CH3 frames …") - zero_frame = bytes(samples_per_frame) - n_frames_acquired = int(scope.query("ACQuire:NUMFRAMESACQuired?")) - for _ in range(n_frames_acquired): - scan_file.write(zero_frame) - elif ch == 4: - self.status_msg.emit(f"Fetching CH{ch} data …") - scope.set_data_source(ch) - waveforms: list[bytes] = scope.transfer_fastframe(parse=False) - if waveforms: - frame_avgs = [] - for w in waveforms: - n = len(w) - frame_avgs.append( - sum(struct.unpack(f"{n}b", w)) / n if n > 0 else 0.0 - ) - self.dc_bias_updated.emit(ri + 1, frame_avgs) - for w in waveforms: - scan_file.write(w) - else: - self.status_msg.emit(f"Fetching CH{ch} data …") - scope.set_data_source(ch) - waveforms: list[bytes] = scope.transfer_fastframe(parse=False) - for w in waveforms: - scan_file.write(w) - - self.row_done.emit(ri + 1, a_n_rows, ai + 1, n_angles) - - finally: - scan_file.close() - self._bbd.scanning_active = False - # Return GR axis to home (0°) regardless of abort or error - if self._t3r is not None and self._t3r.is_open and abs(prev_gr_deg) > 0.001: - return_deg = -prev_gr_deg - est_secs = (abs(self._t3r.steps_for_angle(return_deg, GR_MICROSTEPS)) - / GR_MOVE_VELOCITY + GR_MOVE_VELOCITY / GR_MOVE_ACCEL) - self.status_msg.emit( - f"Returning GR to home ({return_deg:+.1f}°, ≈{est_secs:.1f} s) …" - ) - self._t3r.rotate_stage(return_deg, GR_MICROSTEPS, - GR_MOVE_VELOCITY, GR_MOVE_ACCEL) - time.sleep(est_secs + 0.5) - - if self._abort: - self.failed.emit("Scan aborted by user.") - else: - self.status_msg.emit("Scan complete.") - self.completed.emit() - - # ── Main window ─────────────────────────────────────────────────────────────── class MainWindow(QMainWindow): @@ -1394,8 +916,10 @@ class MainWindow(QMainWindow): uic.loadUi(ROOT / "sc3-aui-main.ui", self) self.setWindowTitle("Scanengine-3 AUI") - # ── T3R driver (owns serial + reader QThread internally) ───────────── - self._t3r_driver = T3RDriver(self) + # ── 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 ──────────────────────────────────────────────────── @@ -1419,8 +943,8 @@ class MainWindow(QMainWindow): self._helios_win = HeliosWindow(self._helios_worker) self._scan_progress = ScanProgressWindow() - self._scan_worker: ScanWorker | None = None - self._scan_thread: QThread | None = None + self._scan_worker: QtScanController | None = None + self._scan_thread: QThread | None = None self._bbd_active_jog: tuple[str, int] | None = None # (axis, direction) self._bbd_jog_timer = QTimer(self) @@ -1739,11 +1263,9 @@ class MainWindow(QMainWindow): QMessageBox.warning(self, "Cannot Resume Scan", f"Could not read scan file:\n\n{e}") return - hdr = sras.header - if (hdr.bytes_per_sample != 1 or hdr.n_channels != len(SCAN_CHANNELS) - or abs(hdr.velocity - SCAN_VELOCITY_MM_S) > 1e-3 - or abs(hdr.laser_freq - LASER_FREQ_HZ) > 1e-3 - or abs(hdr.sample_rate - SCOPE_SAMPLE_RATE) > 1.0): + 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 " @@ -1760,59 +1282,47 @@ class MainWindow(QMainWindow): if not selected: return - # Data is one contiguous stream, so angles past the frontier (the - # first incomplete one) can't be skipped over — back-fill the range. - frontier_idx = next((s.index for s in statuses if not s.complete), len(statuses)) - at_or_past_frontier = {i for i in selected if i >= frontier_idx} - if at_or_past_frontier: - final = selected | set(range(frontier_idx, max(at_or_past_frontier) + 1)) - else: - final = selected - - auto_added = sorted(final - selected) - if auto_added: + 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"{frontier_idx + 1}, so angle(s) can't be skipped over. " + 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 auto_added]}" + f"no gap: {[i + 1 for i in resume_plan.auto_added]}" ) - targets = [ - { - "ai": s.index, "data_offset": s.data_offset, - "n_rows": s.n_rows, "angle_deg": s.angle_deg, - } - for s in statuses if s.index in final - ] - total_rows = sum(t["n_rows"] for t in targets) - angle_list = ", ".join(f"{t['ai'] + 1}" for t in targets) + 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: {total_rows}\n\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 - plan = plan_from_header(sras) - resume_info = { - "path": path, - "targets": targets, - "samples_per_frame": hdr.samples_per_frame, - } - self._launch_scan_worker(plan, path.stem, str(path.parent), resume_info) + 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_info: dict | None = None): + resume: ResumeState | None = None): + rotator = RotationAxis(self._t3r_driver.driver, DEFAULT_ROTATION) + self._scan_thread = QThread(self) - self._scan_worker = ScanWorker( - self._bbd_worker, self._t3r_driver, self._oscope_worker, - plan, prefix, save_dir, resume_info + 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), ) self._scan_worker.moveToThread(self._scan_thread) self._scan_thread.started.connect(self._scan_worker.run) @@ -1827,13 +1337,11 @@ class MainWindow(QMainWindow): self.start_scan_btn.setEnabled(False) self._scan_progress.reset_pause_btn() - if resume_info is None: - self._scan_progress.update_progress(0, plan.per_angle[0].n_rows, 0, plan.n_angles) - else: - ai0 = resume_info["targets"][0]["ai"] - self._scan_progress.update_progress( - 0, plan.per_angle[ai0].n_rows, ai0 + 1, plan.n_angles - ) + 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.show() self._scan_thread.start() @@ -1855,7 +1363,7 @@ class MainWindow(QMainWindow): _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=GR_ROTATION_SIGN, + 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 @@ -1901,7 +1409,7 @@ class MainWindow(QMainWindow): def _disconnect_all_and_close(self): """Cleanly disconnect all hardware then quit.""" if self._t3r_driver.is_open: - self._t3r_driver.disconnect() + self._t3r_driver.close() if self._bbd_worker.is_connected: self._bbd_worker.queue_disconnect() if self._oscope_worker.is_connected: @@ -1930,7 +1438,7 @@ class MainWindow(QMainWindow): self._scan_progress.close() self._helios_win.close() if self._t3r_driver.is_open: - self._t3r_driver.disconnect() + 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): diff --git a/t3r_control_panel.py b/t3r_control_panel.py index 38de516..962feea 100755 --- a/t3r_control_panel.py +++ b/t3r_control_panel.py @@ -27,8 +27,9 @@ from PyQt6.QtWidgets import ( QSpinBox, QSplitter, QVBoxLayout, QWidget, ) -from hardware.t3r_driver import T3RDriver +from gui.qt_t3r import QtT3RAdapter from hardware.serial_util import scored_ports +from hardware.t3r_driver import T3RDriver # class constants (gear train, channel names) import hardware.t3r_protocol as proto @@ -62,7 +63,7 @@ def _spin(lo: int, hi: int, val: int) -> QSpinBox: class ChannelPanel(QGroupBox): """Controls and live readouts for one T3R axis.""" - def __init__(self, ch: int, driver: T3RDriver): + def __init__(self, ch: int, driver: QtT3RAdapter): label = f"Axis {ch} — {T3RDriver.CHANNEL_NAMES[ch]}" super().__init__(label) self.ch = ch @@ -278,7 +279,7 @@ class ChannelPanel(QGroupBox): class GroupPanel(QGroupBox): """Ganged motion — selected axes step in lockstep.""" - def __init__(self, driver: T3RDriver, log_fn): + def __init__(self, driver: QtT3RAdapter, log_fn): super().__init__("Ganged / synchronised motion — selected axes move in lockstep") self._driver = driver self._log = log_fn @@ -389,7 +390,7 @@ class RotationPanel(QGroupBox): Ratio = 125/10 = 12.5 """ - def __init__(self, driver: T3RDriver): + def __init__(self, driver: QtT3RAdapter): super().__init__( f"Stage Rotation (GR-axis ch{T3RDriver.GR_AXIS_CH}) — " f"gear: {T3RDriver.GEAR_TEETH_MOTOR}T motor → 30T idler → " @@ -481,12 +482,12 @@ class RotationPanel(QGroupBox): class T3RControlPanel(QDialog): """User-hidable T3R control window. - Pass a T3RDriver instance. The panel connects to its signals and forwards + Pass a QtT3RAdapter instance. The panel connects to its signals and forwards commands via its API. Connection management (port open/close) is handled inside the panel itself. """ - def __init__(self, driver: T3RDriver, parent=None): + def __init__(self, driver: QtT3RAdapter, parent=None): super().__init__(parent) self.setWindowTitle("T3R Stepper Controller") self.setWindowFlags( @@ -624,14 +625,14 @@ class T3RControlPanel(QDialog): def _toggle_connect(self): if self._driver.is_open: - self._driver.disconnect() + self._driver.close() return port = self.port_combo.currentData() if not port: self._log("No serial port selected", "err") return try: - self._driver.connect(port) + self._driver.open(port) except Exception as exc: self._log(f"Connect failed: {exc}", "err") self.conn_lbl.setText("connect failed") diff --git a/tests/fakes.py b/tests/fakes.py new file mode 100644 index 0000000..ec1e7ba --- /dev/null +++ b/tests/fakes.py @@ -0,0 +1,185 @@ +"""Recording fake hardware for headless ScanEngine tests. + +Each fake records an ordered call trace, so a test can assert the exact +command sequence the engine issues — the property that matters when the +real rig isn't available. +""" +from __future__ import annotations + + +class Trace: + """Ordered record of hardware calls, shared by all fakes in one test.""" + + def __init__(self): + self.calls: list[tuple] = [] + + def record(self, *entry): + self.calls.append(entry) + + def names(self) -> list[str]: + return [c[0] for c in self.calls] + + def of(self, name: str) -> list[tuple]: + return [c for c in self.calls if c[0] == name] + + def count(self, name: str) -> int: + return len(self.of(name)) + + +class FakeStage: + """Stands in for ThorlabsServoDriver.""" + + def __init__(self, trace: Trace, homed=(True, True), enabled=(True, True)): + self._t = trace + self.am_homed = list(homed) + self.am_enabled = list(enabled) + self.positions = [0.0, 0.0] + + def enable_axis(self, axis): + self._t.record("enable_axis", axis) + self.am_enabled[0 if axis == 0x21 else 1] = True + + def home_axis(self, axis, timeout=0.0): + self._t.record("home_axis", axis) + self.am_homed[0 if axis == 0x21 else 1] = True + + def set_velocity_params(self, axis, max_velocity=None, acceleration=None): + self._t.record("set_velocity_params", axis, max_velocity, acceleration) + + def set_trigger_trigout_maxv(self, axis): + self._t.record("set_trigger_trigout_maxv", axis) + + def move_axis_absolute(self, axis, pos, timeout=0.0): + self._t.record("move_axis_absolute", axis, round(pos, 6)) + self.positions[0 if axis == 0x21 else 1] = pos + + +class FakeScope: + """Stands in for TektronixOscilloscopeBase. + + Returns deterministic frame bytes so the written file can be compared + against an expected byte pattern. + """ + + def __init__(self, trace: Trace, samples_per_frame=8, n_frames=4): + self._t = trace + self.samples_per_frame = samples_per_frame + self._n_frames = n_frames + self._acq_polls = 0 + self.frame_seq = 0 + + # -- writes / queries --------------------------------------------------- + def write(self, cmd): + self._t.record("write", cmd) + + def query(self, cmd): + self._t.record("query", cmd) + if cmd == "ACQuire:STATE?": + self._acq_polls += 1 + return "0" # background average finished + if cmd == "ACQuire:NUMFRAMESACQuired?": + return str(self._n_frames) + return "" + + # -- typed setters used by core.scope_sras ------------------------------ + def set_trigger_source(self, ch): + self._t.record("set_trigger_source", ch) + + def set_trigger_slope(self, slope): + self._t.record("set_trigger_slope", slope) + + def set_trigger_level(self, ch, level): + self._t.record("set_trigger_level", ch, level) + + def set_trigger_mode(self, mode): + self._t.record("set_trigger_mode", mode) + + def set_acquire_mode(self, mode): + self._t.record("set_acquire_mode", mode) + + def set_fastframe_state(self, on): + self._t.record("set_fastframe_state", on) + + def set_fastframe_count(self, n): + self._t.record("set_fastframe_count", n) + self._n_frames = n + + def set_sample_rate(self, sr): + self._t.record("set_sample_rate", sr) + + def get_record_length(self): + return self.samples_per_frame + + def set_data_source(self, ch): + self._t.record("set_data_source", ch) + self._source = ch + + def query_wfmoutpre(self): + return f"WFMOUTPRE:CH{self._source};YMULT 1.5625E-3;YOFF -87.04;YZERO 0.0" + + def transfer_curve(self): + self._t.record("transfer_curve") + return bytes(range(self.samples_per_frame)) + + def transfer_fastframe(self, parse=True): + self._t.record("transfer_fastframe", self._source) + frames = [] + for i in range(self._n_frames): + frames.append(bytes((self.frame_seq + i + s) % 256 + for s in range(self.samples_per_frame))) + self.frame_seq += 1 + return frames + + # channel config (only used by configure_channels) + def set_channel_label_name(self, ch, name): + self._t.record("set_channel_label_name", ch, name) + + def set_channel_scale(self, ch, v): + self._t.record("set_channel_scale", ch, v) + + def set_channel_position(self, ch, v): + self._t.record("set_channel_position", ch, v) + + def set_channel_termination(self, ch, v): + self._t.record("set_channel_termination", ch, v) + + def set_channel_coupling(self, ch, v): + self._t.record("set_channel_coupling", ch, v) + + def set_channel_bandwidth(self, ch, v): + self._t.record("set_channel_bandwidth", ch, v) + + +class FakeT3R: + """Stands in for the (Qt-free) T3RDriver, for RotationAxis.""" + + GR_AXIS_CH = 3 + MOTOR_FULL_STEPS_PER_REV = 200 + GEAR_TEETH_MOTOR = 10 + GEAR_TEETH_STAGE = 125 + + def __init__(self, trace: Trace, is_open=True, motion_completes=True): + self._t = trace + self.is_open = is_open + self._motion_completes = motion_completes + + def set_microstep(self, ch, micro): + self._t.record("t3r_set_microstep", ch, micro) + + def set_current(self, ch, run_ma, hold_ma, ihold): + self._t.record("t3r_set_current", ch, run_ma, hold_ma, ihold) + + def enable(self, ch): + self._t.record("t3r_enable", ch) + + def steps_for_angle(self, angle_deg, microsteps): + ratio = self.GEAR_TEETH_STAGE / self.GEAR_TEETH_MOTOR + return round(self.MOTOR_FULL_STEPS_PER_REV * microsteps * ratio + * angle_deg / 360.0) + + def rotate_stage(self, angle_deg, microsteps, velocity, accel): + self._t.record("t3r_rotate", round(angle_deg, 6)) + + def wait_motion_done(self, ch, timeout): + self._t.record("t3r_wait_motion_done", ch) + return self._motion_completes diff --git a/tests/test_scan_engine.py b/tests/test_scan_engine.py new file mode 100644 index 0000000..1ce547d --- /dev/null +++ b/tests/test_scan_engine.py @@ -0,0 +1,282 @@ +"""Headless ScanEngine tests driven entirely by fake hardware. + +These cover what can't be checked without the rig: the command sequence, +the written file layout, and abort/pause behaviour. +""" +import threading +import time + +import pytest + +from core.rotation import RotationAxis, RotationSettings +from core.scan_engine import ( + AXIS_X, AXIS_Y, ScanAborted, ScanCallbacks, ScanEngine, ResumeState, + ResumeTarget, +) +from core.scan_geometry import ScanGeometryError, build_plan +from core.sras_format import SCAN_CHANNELS, SrasFile +from fakes import FakeScope, FakeStage, FakeT3R, Trace + +SPF = 8 + + +def make_plan(num_angles=1): + # Small ROI well inside the stage limits: 1 row, few frames per angle. + return build_plan(40.0, 30.0, 0.02, 0.005, num_angles, 0.01, + laser_freq_hz=20000.0, velocity_mm_s=100.0) + + +def build(tmp_path, num_angles=1, callbacks=None, resume=None, **kw): + trace = Trace() + stage = FakeStage(trace) + scope = FakeScope(trace, samples_per_frame=SPF) + t3r = FakeT3R(trace, **kw) + rotator = RotationAxis(t3r, RotationSettings()) + plan = make_plan(num_angles) + engine = ScanEngine(stage, scope, rotator, plan, tmp_path / "out.sras", + resume=resume, + callbacks=callbacks or ScanCallbacks()) + return engine, trace, plan + + +def test_single_angle_scan_writes_readable_file(tmp_path): + engine, trace, plan = build(tmp_path) + result = engine.run() + + assert not result.aborted + assert result.rows_written == plan.per_angle[0].n_rows + assert result.angles_acquired == [0] + + sras = SrasFile(result.path) + assert sras.header.n_angles == 1 + assert sras.header.samples_per_frame == SPF + assert sras.header.n_channels == len(SCAN_CHANNELS) + # File is complete: every declared row present on disk + assert [s.status for s in sras.angle_status()] == ["OK"] + assert len(sras.preambles) == 3 + assert sras.background == bytes(range(SPF)) + + +def test_command_sequence_order(tmp_path): + engine, trace, plan = build(tmp_path) + engine.run() + names = trace.names() + + def first(name): + return names.index(name) + + # Stage prepared, then scope configured, then rows executed + assert first("set_trigger_trigout_maxv") < first("set_sample_rate") + assert first("set_sample_rate") < first("transfer_fastframe") + # Velocity set for both axes before any scan move + assert trace.count("set_velocity_params") == 2 + # Per row: Y positioned, then X pre-ramp, then X run + moves = trace.of("move_axis_absolute") + assert moves[0][1] == AXIS_Y + assert moves[1][1] == AXIS_X and moves[2][1] == AXIS_X + assert moves[1][2] < moves[2][2] # pre-ramp start < run-off end + # Data channels transferred (CH3 is synthesized, not read) + assert [c[1] for c in trace.of("transfer_fastframe")] == [1, 4] + + +def test_multi_angle_rotates_and_returns_home(tmp_path): + engine, trace, plan = build(tmp_path, num_angles=3) + engine.run() + + rotations = [c[1] for c in trace.of("t3r_rotate")] + # Three angles at 0/-90/-180 → two moves out, then one back to 0 + assert rotations == [-90.0, -90.0, 180.0] + # Every move waits for completion instead of sleeping a guess + assert trace.count("t3r_wait_motion_done") == len(rotations) + # GR configured once, before any rotation + assert trace.names().index("t3r_set_microstep") < trace.names().index("t3r_rotate") + + sras = SrasFile(tmp_path / "out.sras") + assert [s.status for s in sras.angle_status()] == ["OK"] * 3 + + +def test_fastframe_count_rearmed_per_angle(tmp_path): + engine, trace, plan = build(tmp_path, num_angles=3) + engine.run() + counts = [c[1] for c in trace.of("set_fastframe_count")] + assert counts == [pa.n_frames for pa in plan.per_angle] + + +def test_abort_before_start_raises_and_stops_early(tmp_path): + engine, trace, _ = build(tmp_path) + engine.abort() + with pytest.raises(ScanAborted): + engine.run() + assert trace.count("transfer_fastframe") == 0 + + +def test_abort_during_prompt_unblocks(tmp_path): + """A prompt that never returns must not deadlock an aborting scan.""" + released = threading.Event() + + def prompt(title, msg): + # Simulates the GUI bridge: waits until abort flips the flag. + while not engine.aborted: + if released.wait(0.01): + return + + engine, trace, _ = build(tmp_path, callbacks=ScanCallbacks(prompt=prompt)) + + errors = [] + + def run(): + try: + engine.run() + except ScanAborted: + errors.append("aborted") + + t = threading.Thread(target=run, daemon=True) + t.start() + time.sleep(0.2) # let it reach the first prompt + engine.abort() + t.join(timeout=5) + assert not t.is_alive(), "engine deadlocked on a prompt during abort" + assert errors == ["aborted"] + + +def test_pause_and_resume_at_row_boundary(tmp_path): + states = [] + engine, trace, plan = build( + tmp_path, num_angles=1, + callbacks=ScanCallbacks(on_paused_changed=states.append)) + engine.pause() + + done = threading.Event() + + def run(): + try: + engine.run() + except ScanAborted: + pass # only reachable via the failure escape hatch below + finally: + done.set() + + t = threading.Thread(target=run, daemon=True) + t.start() + try: + # The engine's instrument-settling sleeps run before the first row, + # so poll for the pause rather than assuming a fixed delay. + deadline = time.monotonic() + 10.0 + while not states and time.monotonic() < deadline: + time.sleep(0.05) + paused = bool(states) + assert paused and states[0] is True, "engine did not report the pause" + finally: + # Always release the scan thread; if the pause never arrived, abort + # too, so a failed assertion can't leave it parked forever. + if not states: + engine.abort() + engine.resume() + t.join(timeout=10) + assert done.is_set() + assert states[-1] is False + + +def test_dc_bias_callback_reports_per_frame_means(tmp_path): + rows = [] + engine, trace, plan = build( + tmp_path, callbacks=ScanCallbacks(on_dc_bias=lambda r, m: rows.append((r, m)))) + engine.run() + + assert len(rows) == plan.per_angle[0].n_rows + row_idx, means = rows[0] + assert row_idx == 1 + assert len(means) == plan.per_angle[0].n_frames + assert all(isinstance(v, float) for v in means) + + +def test_offstage_plan_rejected_before_touching_hardware(tmp_path): + trace = Trace() + stage = FakeStage(trace) + scope = FakeScope(trace, samples_per_frame=SPF) + # X range that runs off the 110 mm stage once ramps are added + plan = build_plan(80.0, 30.0, 40.0, 5.0, 1, 0.25, + laser_freq_hz=20000.0, velocity_mm_s=100.0) + engine = ScanEngine(stage, scope, None, plan, tmp_path / "bad.sras") + with pytest.raises(ScanGeometryError): + engine.run() + assert trace.calls == [], "hardware touched despite invalid geometry" + + +def test_multi_angle_without_rotator_raises(tmp_path): + trace = Trace() + engine = ScanEngine(FakeStage(trace), FakeScope(trace, samples_per_frame=SPF), + None, make_plan(3), tmp_path / "x.sras") + with pytest.raises(RuntimeError, match="T3R rotation stage"): + engine.run() + + +def test_missing_hardware_raises(tmp_path): + trace = Trace() + with pytest.raises(RuntimeError, match="BBD202"): + ScanEngine(None, FakeScope(trace), None, make_plan(), + tmp_path / "x.sras").run() + with pytest.raises(RuntimeError, match="Oscilloscope"): + ScanEngine(FakeStage(trace), None, None, make_plan(), + tmp_path / "x.sras").run() + + +def test_resume_seeks_to_angle_offset_and_skips_others(tmp_path): + # First produce a complete 3-angle file + engine, trace, plan = build(tmp_path, num_angles=3) + engine.run() + path = tmp_path / "out.sras" + original = path.read_bytes() + + sras = SrasFile(path) + statuses = sras.angle_status() + target = statuses[1] + resume = ResumeState( + path=path, + targets=[ResumeTarget(target.index, target.data_offset, + target.n_rows, target.angle_deg)], + samples_per_frame=SPF, + ) + + engine2, trace2, _ = build(tmp_path, num_angles=3, resume=resume) + result = engine2.run() + + assert result.angles_acquired == [1] + # Only the middle angle's rows were re-acquired + assert result.rows_written == plan.per_angle[1].n_rows + rewritten = path.read_bytes() + assert len(rewritten) == len(original) + # Angle 0's block is untouched; angle 1's changed (fresh frame data) + a1_start, a1_end = target.data_offset, target.data_offset + target.row_bytes * target.n_rows + assert rewritten[:a1_start] == original[:a1_start] + assert rewritten[a1_start:a1_end] != original[a1_start:a1_end] + assert rewritten[a1_end:] == original[a1_end:] + + +def test_resume_record_length_mismatch_rejected(tmp_path): + engine, trace, plan = build(tmp_path) + engine.run() + path = tmp_path / "out.sras" + resume = ResumeState(path=path, + targets=[ResumeTarget(0, 0, 1, 0.0)], + samples_per_frame=SPF + 1) # scope changed + engine2, _, _ = build(tmp_path, resume=resume) + with pytest.raises(RuntimeError, match="record length"): + engine2.run() + + +def test_engine_imports_without_qt(): + """The engine must be usable from a non-Qt front end.""" + import subprocess + import sys + code = ( + "import sys;" + "sys.modules['PyQt6'] = None;" + "import core.scan_engine, core.rotation, core.scope_sras," + " core.scan_resume, core.sras_format, core.scan_geometry;" + "print('ok')" + ) + out = subprocess.run([sys.executable, "-c", code], capture_output=True, + text=True, cwd=str(__import__('pathlib').Path(__file__).parent.parent)) + assert out.returncode == 0, out.stderr + assert "ok" in out.stdout diff --git a/tests/test_scan_resume.py b/tests/test_scan_resume.py new file mode 100644 index 0000000..417ffb3 --- /dev/null +++ b/tests/test_scan_resume.py @@ -0,0 +1,77 @@ +"""core.scan_resume: the frontier contiguity rule, over real fixture files.""" +from pathlib import Path + +from core.scan_resume import is_compatible, plan_resume +from core.sras_format import SrasFile + +GOLDEN = Path(__file__).parent / "golden" + + +def _statuses(name): + return SrasFile(GOLDEN / name).angle_status() + + +def test_complete_file_frontier_is_past_the_end(): + st = _statuses("complete.sras") + plan = plan_resume(st, selected={0}) + assert plan.frontier_idx == len(st) + assert [t.angle_idx for t in plan.targets] == [0] + assert plan.auto_added == [] + + +def test_selecting_past_frontier_backfills_the_gap(): + # angle 0 complete, angle 1 truncated → frontier = 1 + st = _statuses("trunc_rowboundary_a1.sras") + assert st[0].complete and not st[1].complete + + plan = plan_resume(st, selected={1}) + assert plan.frontier_idx == 1 + assert [t.angle_idx for t in plan.targets] == [1] + assert plan.auto_added == [] + + +def test_selection_before_frontier_is_untouched(): + st = _statuses("trunc_rowboundary_a1.sras") + plan = plan_resume(st, selected={0}) + assert [t.angle_idx for t in plan.targets] == [0] + assert plan.auto_added == [] + + +def test_selecting_only_a_later_angle_pulls_in_the_frontier(): + """Data is one contiguous stream, so angle 1 can't be skipped to reach 2.""" + st = _statuses("header_only.sras") # nothing written: frontier = 0 + assert plan_resume(st, selected={0}).frontier_idx == 0 + + plan = plan_resume(st, selected={1}) + assert [t.angle_idx for t in plan.targets] == [0, 1] + assert plan.auto_added == [0] + + +def test_targets_carry_offsets_and_rows(): + st = _statuses("complete.sras") + plan = plan_resume(st, selected={0, 1}) + for target, status in zip(plan.targets, st, strict=True): + assert target.data_offset == status.data_offset + assert target.n_rows == status.n_rows + assert target.angle_deg == status.angle_deg + assert plan.total_rows == sum(s.n_rows for s in st) + + +def test_to_state_carries_samples_per_frame(): + sras = SrasFile(GOLDEN / "complete.sras") + state = plan_resume(sras.angle_status(), selected={0}).to_state(sras) + assert state.path == sras.path + assert state.samples_per_frame == sras.header.samples_per_frame + assert state.target_indices == {0} + + +def test_is_compatible_checks_acquisition_settings(): + sras = SrasFile(GOLDEN / "complete.sras") + h = sras.header + ok = dict(velocity=h.velocity, laser_freq=h.laser_freq, + sample_rate=h.sample_rate, n_channels=h.n_channels) + assert is_compatible(sras, **ok) + assert not is_compatible(sras, **{**ok, "velocity": h.velocity + 1}) + assert not is_compatible(sras, **{**ok, "laser_freq": h.laser_freq * 2}) + assert not is_compatible(sras, **{**ok, "sample_rate": h.sample_rate * 2}) + assert not is_compatible(sras, **{**ok, "n_channels": h.n_channels + 1})