fix(aerotech): park the PSO window once the line is triggered

At the beamline (DDG1 trigger) every line started two bursts: the window was
switched to the next line when the axis passed trigger_high, just after the DDG1
cycle (as long as the travel trigger_low -> trigger_high), and the edge started a
second burst. MCS and detectors counted twice the frames, so their frames no
longer matched the captured positions (progress bar done at half the scan).

The window is now moved to a parking range beyond the line end as soon as the axis
has entered it, while DDG1 / PandA run the line and ignore edges; the next line's
window is set beyond the exposures as before, outside both windows. The per-line
MCS check also fails on more frames than the line has.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017NrH3poJsYBCTVq2Y9fuHZ

csaxs_bec/scans/scans_v4/aerotech_cont_grid.py

@@ -23,8 +23,13 @@ line. The axis never leaves the window during a line: with a closed window the e
 coincides with the end of the exposures, and encoder noise at the exit edge (seen
 2026-10-29: 2 extra edges in 11 lines) could start an extra line in the trigger receiver.
 Noise at the entry edge falls into the running line, which DDG and PandA ignore. The window
-is armed once per scan (one position reference) and only its range is switched before each
-line, while the axis is outside both windows; the trigger receiver stays armed.
+is armed once per scan (one position reference) and only its range is switched, the trigger
+receiver stays armed. As soon as the line is triggered, the window is moved to a parking
+range the axis never reaches: the output goes low while DDG1 / PandA run the line and ignore
+edges. Switching from the line's window to the next one beyond the exposures instead gave a
+second trigger per line at the beamline (2026-11-03, DDG1: the switch came just after the
+DDG1 cycle, which lasts as long as the axis needs from trigger_low to trigger_high). The next
+line's window is set once the axis has left the exposure range, outside both windows.
 
 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
@@ -429,14 +434,19 @@ class AerotechContGrid(ScanBase):
     ):
         """
         Run one line: arm the position capture, move the fast axis through the PSO window,
-        start the step to the next line once the fast axis has left the exposure range,
-        then check that the line was triggered and read the captured positions.
+        park the window once the line is triggered, start the step to the next line once the
+        fast axis has left the exposure range, then check that the line was triggered and
+        read the captured positions.
         """
         line_start_time = time.time()
         self._line_starts.append(line_start_time)
         self.capture.arm(self.frames_per_line)
         move_status = self.actions.set(motors, positions, wait=False)
         next_line = line_index + 1 if line_index + 1 < self._line["num_lines"] else None
+        # the PSO output goes low while the trigger receiver runs the line and ignores edges
+        while not move_status.done and not self._past_trigger(line_index):
+            time.sleep(STEP_POLL_INTERVAL)
+        self.fast_axis.set_line_trigger_window(*self._parking_window())
         while next_line is not None and not move_status.done:
             if self._past_exposures(line_index):
                 break
@@ -498,6 +508,19 @@ class AerotechContGrid(ScanBase):
     def _is_forward(self, line_index: int) -> bool:
         return line_index % 2 == 0
 
+    def _past_trigger(self, line_index: int) -> bool:
+        """The fast axis has entered the PSO window of the line (the line is triggered)."""
+        position = self.fast_axis.readback.get()
+        if self._is_forward(line_index):
+            return position > self._line["trigger_low"]
+        return position < self._line["trigger_high"]
+
+    def _parking_window(self) -> tuple[float, float]:
+        """PSO window beyond the line end, never reached by the fast axis."""
+        line = self._line
+        span = line["line_end"] - line["line_start"]
+        return line["line_end"] + span, line["line_end"] + 2 * span
+
     def _past_exposures(self, line_index: int) -> bool:
         """The fast axis has left the exposure range of the line (beyond its trigger position)."""
         position = self.fast_axis.readback.get()
@@ -603,7 +626,9 @@ class AerotechContGrid(ScanBase):
         """
         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.
+        line trigger right after the line instead of at the end of the scan, and an extra
+        one: the detectors would count it as well, their frames would no longer match the
+        captured positions.
         """
         expected = (line_index + 1) * self.frames_per_line - 1
         deadline = time.time() + timeout
@@ -615,6 +640,12 @@ class AerotechContGrid(ScanBase):
                     " DDG1 EXT wiring and trigger_edge."
                 )
             time.sleep(poll)
+        if channel > expected:
+            raise RuntimeError(
+                f"MCS card at channel {channel} after line {line_index}, expected"
+                f" {expected}: more frames than the line has, the trigger receiver was"
+                " triggered more than once. The detector frames do not match the positions."
+            )
 
     def _line_duration(self) -> float:
         line = self._line

tests/tests_scans/test_aerotech_cont_grid_scan.py

@@ -348,13 +348,21 @@ def test_snake_scan_core(v4_scan_assembler, device_manager, nth_done_status_mock
         ([stepper], [2.0]),
         ([fast], [line["line_end"]]),
     ]
-    # the PSO window is armed once in pre_scan, not per line; only its range is switched
+    # the PSO window is armed once in pre_scan, not per line; only its range is switched:
+    # parked once each line is triggered, then the window of the next line
     scan.fast_axis.arm_line_trigger.assert_not_called()
     scan.fast_axis.disarm_line_trigger.assert_not_called()
     span = line["line_end"] - line["line_start"]
+    park = mock.call(line["line_end"] + span, line["line_end"] + 2 * span)
     backward = mock.call(line["line_start"] - span, line["trigger_high"])
     forward = mock.call(line["trigger_low"], line["line_end"] + span)
-    assert scan.fast_axis.set_line_trigger_window.call_args_list == [backward, forward]
+    assert scan.fast_axis.set_line_trigger_window.call_args_list == [
+        park,
+        backward,
+        park,
+        forward,
+        park,
+    ]
 
 
 def test_panda_lines_are_checked(v4_scan_assembler, device_manager, nth_done_status_mock):
@@ -489,8 +497,15 @@ def test_step_starts_when_fast_axis_left_the_exposures(
         return nth_done_status_mock(resolve_after=1)
 
     scan.actions.set = mock.MagicMock(side_effect=set_side_effect)
-    # beyond the high trigger position after forward lines, below the low one after backward
-    scan.fast_axis.readback.get.side_effect = [line["line_end"], line["line_start"]]
+    # per line: inside the window (triggered), then beyond the exposures (forward: beyond
+    # trigger_high, backward: below trigger_low); the last line only parks the window
+    scan.fast_axis.readback.get.side_effect = [
+        line["line_end"],
+        line["line_end"],
+        line["line_start"],
+        line["line_start"],
+        line["line_end"],
+    ]
 
     scan.scan_core()
 
@@ -502,12 +517,64 @@ def test_step_starts_when_fast_axis_left_the_exposures(
         ("fast", line["line_end"]),
     ]
     span = line["line_end"] - line["line_start"]
+    park = mock.call(line["line_end"] + span, line["line_end"] + 2 * span)
     assert scan.fast_axis.set_line_trigger_window.call_args_list == [
+        park,
         mock.call(line["line_start"] - span, line["trigger_high"]),
+        park,
         mock.call(line["trigger_low"], line["line_end"] + span),
+        park,
     ]
 
 
+def test_window_parked_once_line_is_triggered(
+    v4_scan_assembler, device_manager, nth_done_status_mock
+):
+    # the window must be parked while the trigger receiver runs the line (it ignores the
+    # edge then), not at the end of the exposures: that gave a second trigger per line
+    scan = _ready_for_core(
+        _assemble(v4_scan_assembler, device_manager, trigger_source=None), nth_done_status_mock
+    )
+    line = scan._line
+    status = mock.MagicMock()
+    status.done = False
+    scan.actions.set = mock.MagicMock(return_value=status)
+    windows = []
+    scan.fast_axis.set_line_trigger_window.side_effect = lambda *args: windows.append(args)
+    # forward line: run-up (before the window), entered, then beyond the exposures
+    positions = [line["line_start"], line["trigger_low"] + 1e-3, line["trigger_high"] + 1e-3]
+    readbacks = iter(positions)
+
+    def readback():
+        value = next(readbacks)
+        readback.seen.append(value)
+        return value
+
+    readback.seen = []
+    scan.fast_axis.readback.get.side_effect = readback
+    status.wait.side_effect = lambda *a, **k: setattr(status, "done", True)
+    scan._line["num_lines"] = 2
+
+    scan.at_each_point(
+        motors=[scan.fast_axis], positions=np.array([line["line_end"]]), line_index=0
+    )
+
+    span = line["line_end"] - line["line_start"]
+    assert windows == [
+        (line["line_end"] + span, line["line_end"] + 2 * span),
+        (line["line_start"] - span, line["trigger_high"]),
+    ]
+    assert readback.seen == positions
+
+
+def test_ddg_extra_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 = 9  # two bursts of 5 frames in line 0
+    with pytest.raises(RuntimeError, match="triggered more than once"):
+        scan._wait_mcs_frames(0, timeout=0.1, poll=0.02)
+
+
 @pytest.mark.parametrize(
     ("attribute", "device_cls"),
     [