fix(aerotech): PSO window open beyond the line end, range switched per line

With one closed window per scan the window exit coincided with the end of the
exposures; encoder noise at the exit edge gave extra trigger edges after the train
(2 in 11 lines on the PandA, one started an extra line). Now the window is open on the
far side of the line and only its range is switched before each line (PSOWINDOW RANGE,
no reload, so the position reference of the arm stays). Hardware test: 2 x 11 lines,
22 edges, 22 lines, 2222 gates, 0 dropped.

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

csaxs_bec/devices/aerotech/README.md

@@ -64,11 +64,16 @@ scans.aerotech_cont_grid(dev.aerox, 0, 1, 0.01, dev.aeroy, 0, 1, 0.1, exp_time=0
 
 - fast axis X moves continuously at `fast_step_size / exp_time`, the step axis Y steps between
   lines; the lines always run as a snake (alternating direction)
-- line start triggered by position: one PSO window per scan,
-  `[fast_start - v * shutter_delay, exposure_end + v * shutter_delay]`, armed once in
-  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
+- line start triggered by position: the PSO output rises on entering the window, at
+  `trigger_low = fast_start - v * shutter_delay` on forward lines and at
+  `trigger_high = exposure_end + v * shutter_delay` on backward lines, so in both directions
+  the first exposure starts one shutter delay after the trigger at the first point of the
+  line. The window is open on the far side (`[trigger_low, beyond line_end]` forward,
+  `[beyond line_start, trigger_high]` backward): the axis never leaves it during a line.
+  With a closed window the exit coincides with the end of the exposures and encoder noise
+  at the exit edge started extra lines (2026-10-29, PandA: 2 extra edges in 11 lines).
+  Armed once in pre_scan (one position reference); before each line only the range is
+  switched (`PSOWINDOW RANGE`, no reload), with the axis outside both windows
 - 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

csaxs_bec/devices/aerotech/a3200.py

@@ -146,6 +146,7 @@ class A3200Controller(Controller):
         if not hasattr(self, "_rx_buffer"):
             self._rx_buffer = ""
             self._counts_per_unit = {}
+            self._pso_reference = {}  # axis -> (position at PSOWINDOW LOAD, PSO counts per unit)
 
     @threadlocked
     def socket_put(self, val: str) -> None:
@@ -357,10 +358,8 @@ class A3200Controller(Controller):
         the EmulatedQuadratureDivider for the MXR emulated quadrature inputs 6-8.
         """
         cpu = self.counts_per_unit(axis) / counts_divider
-        current = self.get_position(axis)
-        low_counts, high_counts = sorted(
-            (int(round((low - current) * cpu)), int(round((high - current) * cpu)))
-        )
+        self._pso_reference[axis] = (self.get_position(axis), cpu)
+        low_counts, high_counts = self._pso_window_counts(axis, low, high)
         self.command(f"PSOCONTROL {axis} RESET")
         self.command(f"PSOWINDOW {axis} 1 INPUT {pso_input}")
         self.command(f"PSOWINDOW {axis} 1 LOAD 0")
@@ -368,9 +367,27 @@ class A3200Controller(Controller):
         self.command(f"PSOOUTPUT {axis} WINDOW")
         self.command(f"PSOCONTROL {axis} ARM")
 
+    def pso_window_range(self, axis: str, low: float, high: float) -> None:
+        """
+        Move the armed window to [low, high] (user units, absolute). The window counter is
+        not reloaded: the position reference of pso_arm_window stays valid, the axis may be
+        in motion.
+        """
+        if axis not in self._pso_reference:
+            raise A3200Error(f"PSO window of axis {axis} is not armed.")
+        low_counts, high_counts = self._pso_window_counts(axis, low, high)
+        self.command(f"PSOWINDOW {axis} 1 RANGE {low_counts}, {high_counts}")
+
+    def _pso_window_counts(self, axis: str, low: float, high: float) -> tuple[int, int]:
+        reference, cpu = self._pso_reference[axis]
+        return tuple(
+            sorted((int(round((low - reference) * cpu)), int(round((high - reference) * cpu))))
+        )
+
     def pso_off(self, axis: str) -> None:
         self.command(f"PSOWINDOW {axis} 1 OFF")
         self.command(f"PSOCONTROL {axis} OFF")
+        self._pso_reference.pop(axis, None)
 
     # --- DATAACQ (position capture into the drive array) -----------------------
 
@@ -743,6 +760,10 @@ class A3200Motor(Device, PositionerBase):
             counts_divider=self.pso_counts_divider,
         )
 
+    def set_line_trigger_window(self, start: float, end: float) -> None:
+        """Move the window of the armed line trigger to [start, end], keeping its reference."""
+        self.controller.pso_window_range(self.axis_Id, start * self.sign, end * self.sign)
+
     def disarm_line_trigger(self) -> None:
         self.controller.pso_off(self.axis_Id)
 

csaxs_bec/scans/scans_v4/aerotech_cont_grid.py

@@ -3,14 +3,20 @@ Continuous grid scan for Aerotech A3200 stages with a position triggered line st
 
 The fast axis moves continuously, the step axis steps between lines. The lines always run
 as a snake (alternating direction). The start of each line is triggered in hardware by
-the A3200 PSO output of the fast axis: one PSO window per scan,
+the A3200 PSO output of the fast axis, which goes high when the axis enters the PSO window.
+The trigger positions are
 
-    [fast_start - v * shutter_delay, exposure_end + v * shutter_delay]
+    trigger_low = fast_start - v * shutter_delay, trigger_high = exposure_end + v * shutter_delay
 
-goes high when the axis enters it, i.e. at the low edge on forward lines and 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. The window is armed once per scan and
-the trigger receiver stays armed; between lines the scan only moves the axes.
+and the window is open on the far side of the line: [trigger_low, beyond line_end] on
+forward lines, [beyond line_start, trigger_high] on backward lines, so in both directions
+the first exposure starts one shutter delay after the trigger at the first point of the
+line. The axis never leaves the window during a line: with a closed window the exit
+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.
 
 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
@@ -330,7 +336,7 @@ class AerotechContGrid(ScanBase):
         self._fast_axis_configured = True
         self.fast_axis.motor_velocity.set(line["target_velocity"]).wait(timeout=5)
         self.fast_axis.wait_in_position.set(False).wait(timeout=5)
-        self.fast_axis.arm_line_trigger(line["trigger_low"], line["trigger_high"])
+        self.fast_axis.arm_line_trigger(*self._trigger_window(0))
         if self.trigger_source == "ddg":
             self.ddg1.set_trigger(DDG_TRIGGER_EDGES[self.trigger_edge])
         elif self.trigger_source == "panda":
@@ -347,6 +353,8 @@ class AerotechContGrid(ScanBase):
         for line_index in range(line["num_lines"]):
             stepper_pos = self.positions[line_index][1]
             if line_index > 0:
+                # fast axis at the turnaround, outside the window of the next line
+                self.fast_axis.set_line_trigger_window(*self._trigger_window(line_index))
                 self.actions.set([self.stepper_axis], [stepper_pos]).wait()
             target = line["line_end"] if self._is_forward(line_index) else line["line_start"]
             self.at_each_point(
@@ -419,6 +427,14 @@ class AerotechContGrid(ScanBase):
     def _is_forward(self, line_index: int) -> bool:
         return line_index % 2 == 0
 
+    def _trigger_window(self, line_index: int) -> tuple[float, float]:
+        """PSO window of a line: entered at the trigger position, open beyond the line end."""
+        line = self._line
+        span = line["line_end"] - line["line_start"]
+        if self._is_forward(line_index):
+            return line["trigger_low"], line["line_end"] + span
+        return line["line_start"] - span, line["trigger_high"]
+
     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

tests/tests_devices/test_a3200.py

@@ -262,6 +262,28 @@ def test_line_trigger_window_is_relative_to_current_position():
     assert not _sim_state().pso["armed"]
 
 
+def test_line_trigger_window_range_keeps_the_reference():
+    x, _, _ = _capture_setup()
+    x.move(-1.0, wait=True)
+    x.arm_line_trigger(0.0, 1.0)
+    x.move(2.0, wait=True)
+    x.set_line_trigger_window(-3.0, 1.5)
+    # relative to the position at arm (-1.0), not to the current one (2.0)
+    assert _sim_state().pso["range"] == (-40000.0, 50000.0)
+    assert "[PUT]: PSOWINDOW X 1 LOAD 0" in x.controller.command_history
+    assert sum("LOAD" in c for c in x.controller.command_history) == 1
+
+
+def test_line_trigger_window_range_needs_armed_window():
+    x, _, _ = _capture_setup()
+    with pytest.raises(A3200Error, match="not armed"):
+        x.set_line_trigger_window(0.0, 1.0)
+    x.arm_line_trigger(0.5, 1.0)
+    x.disarm_line_trigger()
+    with pytest.raises(A3200Error, match="not armed"):
+        x.set_line_trigger_window(0.0, 1.0)
+
+
 def test_line_trigger_uses_emulated_quadrature_divider():
     u = _sim_motor(
         "U", axis_nr=3, egu="deg", pso_input=7, pso_counts_divider=40, sim_counts_per_unit=560000

tests/tests_scans/test_aerotech_cont_grid_scan.py

@@ -39,6 +39,7 @@ def _add_devices(
     )
     aerox.arm_line_trigger = mock.MagicMock()
     aerox.disarm_line_trigger = mock.MagicMock()
+    aerox.set_line_trigger_window = mock.MagicMock()
     # signal writes go through RPC; mock them, keep the configured values readable
     for signal_name, value in (
         ("motor_velocity", 10.0),
@@ -256,7 +257,9 @@ def test_pre_scan_arms_once_after_calibration(
         mock.call.capture.calibrate(),
         mock.call.fast_axis.motor_velocity.set(line["target_velocity"]),
         mock.call.fast_axis.wait_in_position.set(False),
-        mock.call.fast_axis.arm_line_trigger(line["trigger_low"], line["trigger_high"]),
+        mock.call.fast_axis.arm_line_trigger(
+            line["trigger_low"], 2 * line["line_end"] - line["line_start"]
+        ),
         mock.call.ddg1.set_trigger(2),  # falling edge: PSO open collector / differential
     ]
 
@@ -315,9 +318,13 @@ 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
+    # the PSO window is armed once in pre_scan, not per line; only its range is switched
     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"]
+    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]
 
 
 def test_panda_lines_are_checked(v4_scan_assembler, device_manager, nth_done_status_mock):