From 46751e4fccd119855bccb6fd842b936766a13ebf Mon Sep 17 00:00:00 2001 From: Mirko Holler Date: Wed, 28 Oct 2026 17:07:25 +0100 Subject: [PATCH] feat(aerotech): aerotech_cont_grid is hardware triggered, MCS armed once scan_type HARDWARE_TRIGGERED: the MCS card and the detectors are armed once at stage for all lines x frames (existing hardware_triggered mode of mcs_card_csaxs / ddg_1) instead of per line; per line the scan only moves the axes. With trigger_source="ddg" each line is still checked: the MCS must have reached the line's last channel, so a missing line trigger aborts after the first line. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01M8zoDyJxbBLmjZCEPUo5FM --- csaxs_bec/devices/aerotech/README.md | 8 ++- .../scans/scans_v4/aerotech_cont_grid.py | 56 +++++++++++-------- .../test_aerotech_cont_grid_scan.py | 26 +++++++-- 3 files changed, 60 insertions(+), 30 deletions(-) diff --git a/csaxs_bec/devices/aerotech/README.md b/csaxs_bec/devices/aerotech/README.md index 8664e6e3..498e19a7 100644 --- a/csaxs_bec/devices/aerotech/README.md +++ b/csaxs_bec/devices/aerotech/README.md @@ -69,8 +69,12 @@ scans.aerotech_cont_grid(dev.aerox, 0, 1, 0.01, dev.aeroy, 0, 1, 0.1, exp_time=0 pre_scan. Its output rises on entering the window: at the low edge on forward lines, at the high edge on backward lines, so in both directions the first exposure starts one shutter delay after the trigger at the first point of the line -- `trigger_source`: `"ddg"` (beamline: PSO -> DDG1 EXT; DDG1/DDG2 take exp_time and - frames_per_trigger = frames per line from the scan info at stage; MCS prepared per line), +- hardware triggered scan: MCS card and detectors are armed once at stage for all + lines x frames (num_points = lines, frames_per_trigger = frames per line); per line the + scan only moves X and Y +- `trigger_source`: `"ddg"` (beamline: PSO -> DDG1 EXT, one DDG1 burst of frames_per_line + gates per trigger; DDG1/DDG2 take exp_time and frames_per_trigger from the scan info at + stage; after each line the scan checks that the MCS reached the line's last channel), `"panda"` (test setup: PandA configured and armed once, every line checked with `wait_lines_done`; `readout_time`, `shutter_delay` arguments) or `None` (no trigger hardware, motion tests) diff --git a/csaxs_bec/scans/scans_v4/aerotech_cont_grid.py b/csaxs_bec/scans/scans_v4/aerotech_cont_grid.py index 4a91de85..94487b70 100644 --- a/csaxs_bec/scans/scans_v4/aerotech_cont_grid.py +++ b/csaxs_bec/scans/scans_v4/aerotech_cont_grid.py @@ -12,15 +12,22 @@ edge on backward lines, so in both directions the first exposure starts one shut after the trigger at the first point of the line. The window is armed once per scan and the trigger receiver stays armed; between lines the scan only moves the axes. +The scan is hardware triggered: the MCS card and the detectors are armed once at stage for +all num_lines x frames_per_line frames (num_points = lines, frames_per_trigger = frames per +line) and checked for completeness in complete. + Trigger sources (trigger_source): - "ddg": PSO output -> DDG1 EXT (beamline). DDG1 runs shutter / detector / MCS, the MCS - card is prepared per line. trigger_edge: the Npaq open collector (J7-8) and + "ddg": PSO output -> DDG1 EXT (beamline). Every trigger starts one DDG1 burst of + frames_per_line gates (shutter / detectors / MCS). After each line the scan + checks that the MCS card counted the line's frames. trigger_edge: the Npaq open collector (J7-8) and differential (J7-1/2) PSO outputs are active low, so the window entry is a falling edge (default); the opto output (J7-6/7) gives a rising edge. "panda": PSO output -> PandABox TTLIN3 (device pandatrig, test setup). The PandA is configured and armed once and sends the shutter pulse and the detector gates of each line. - None: no trigger hardware (motion tests); the PSO window is still armed. + None: no trigger hardware (motion tests); the PSO window is still armed. Only for + sessions without hardware triggered detectors / MCS, they would wait for + frames that never come. With a position capture device (capture_positions=True), the encoder positions are latched at each detector trigger (Npaq UINT1) and published per line. @@ -45,6 +52,7 @@ Scan procedure: from __future__ import annotations +import time from typing import Annotated, TypedDict import numpy as np @@ -87,8 +95,9 @@ class AerotechLineParameter(TypedDict): class AerotechContGrid(ScanBase): - # One line = one point with frames_per_trigger frames, like cont_grid. - scan_type = ScanType.SOFTWARE_TRIGGERED + # One line = one point with frames_per_trigger frames; MCS and detectors are armed once + # for all frames of the scan. + scan_type = ScanType.HARDWARE_TRIGGERED scan_name = "aerotech_cont_grid" gui_config = { @@ -321,18 +330,9 @@ class AerotechContGrid(ScanBase): self.fast_axis.disarm_line_trigger() self.actions.set(self.motors, [line["line_start"], stepper_pos]).wait() self.fast_axis.arm_line_trigger(line["trigger_low"], line["trigger_high"]) - status_mcs = None - if self.trigger_source == "ddg": - self.ddg1.prepare_mcs_on_trigger().wait() - status_mcs = self.mcs.get_transition_status( - signal_name="acquiring", transition=[1, 0] - ) target = line["line_end"] if self._is_forward(line_index) else line["line_start"] self.at_each_point( - motors=[self.fast_axis], - positions=np.array([target]), - line_index=line_index, - status_mcs=status_mcs, + motors=[self.fast_axis], positions=np.array([target]), line_index=line_index ) @scan_hook @@ -341,7 +341,6 @@ class AerotechContGrid(ScanBase): motors: list[str | DeviceBase], positions: np.ndarray, line_index: int, - status_mcs=None, last_positions: np.ndarray | None = None, ): """ @@ -359,13 +358,7 @@ class AerotechContGrid(ScanBase): continue if self.trigger_source == "ddg": - try: - status_mcs.wait(timeout=3) - except TimeoutError as exc: - raise ScanAbortion( - "MCS card did not go back to DONE after the line. Was the line trigger" - " received? Check the PSO output -> DDG1 EXT wiring and the PSO interlock." - ) from exc + self._wait_mcs_frames(line_index) elif self.trigger_source == "panda": self.panda.wait_lines_done(line_index + 1, timeout=self._line_duration() + 2) @@ -408,6 +401,23 @@ class AerotechContGrid(ScanBase): def _is_forward(self, line_index: int) -> bool: return not self.snake or line_index % 2 == 0 + def _wait_mcs_frames(self, line_index: int, timeout: float = 2.0, poll: float = 0.05): + """ + After line line_index the MCS card must have advanced to the last channel of that + line (the last channel of the scan is closed in its complete). Catches a missing + line trigger right after the line instead of at the end of the scan. + """ + expected = (line_index + 1) * self.frames_per_line - 1 + deadline = time.time() + timeout + while (channel := self.mcs.current_channel.get()) < expected: + if time.time() > deadline: + raise ScanAbortion( + f"MCS card at channel {channel} after line {line_index}, expected" + f" {expected}. Was the line trigger received? Check the PSO output ->" + " DDG1 EXT wiring and trigger_edge." + ) + time.sleep(poll) + def _line_duration(self) -> float: line = self._line return abs(line["line_end"] - line["line_start"]) / line["target_velocity"] diff --git a/tests/tests_scans/test_aerotech_cont_grid_scan.py b/tests/tests_scans/test_aerotech_cont_grid_scan.py index ede79aeb..bfdf8e8c 100644 --- a/tests/tests_scans/test_aerotech_cont_grid_scan.py +++ b/tests/tests_scans/test_aerotech_cont_grid_scan.py @@ -315,17 +315,33 @@ def test_panda_lines_are_checked(v4_scan_assembler, device_manager, nth_done_sta scan.panda.arm.assert_not_called() # armed once in pre_scan -def test_ddg_scan_core_prepares_mcs_per_line( +def test_scan_is_hardware_triggered(v4_scan_assembler, device_manager): + scan = _assemble(v4_scan_assembler, device_manager) + scan.prepare_scan() + assert scan.scan_info.scan_type == "hardware_triggered" + # MCS / detectors: num_points lines x frames_per_trigger frames, armed once + assert scan.scan_info.num_points == 3 + assert scan.scan_info.frames_per_trigger == 5 + + +def test_ddg_scan_core_checks_mcs_frames_per_line( v4_scan_assembler, device_manager, nth_done_status_mock ): scan = _ready_for_core(_assemble(v4_scan_assembler, device_manager), nth_done_status_mock) scan.ddg1 = mock.MagicMock() scan.mcs = mock.MagicMock() - scan.mcs.get_transition_status.side_effect = lambda **kwargs: nth_done_status_mock( - resolve_after=1 - ) + scan.mcs.current_channel.get.side_effect = [4, 9, 14] scan.scan_core() - assert scan.ddg1.prepare_mcs_on_trigger.call_count == 3 + scan.ddg1.prepare_mcs_on_trigger.assert_not_called() # no arming per line + assert scan.mcs.current_channel.get.call_count == 3 + + +def test_ddg_missing_line_trigger_aborts(v4_scan_assembler, device_manager, nth_done_status_mock): + scan = _ready_for_core(_assemble(v4_scan_assembler, device_manager), nth_done_status_mock) + scan.mcs = mock.MagicMock() + scan.mcs.current_channel.get.return_value = 0 + with pytest.raises(ScanAbortion, match="expected 4"): + scan._wait_mcs_frames(0, timeout=0.1, poll=0.02) def test_at_each_point_reads_capture(v4_scan_assembler, device_manager, nth_done_status_mock):