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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01M8zoDyJxbBLmjZCEPUo5FM

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)

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"]

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):