feat(aerotech): aerotech_cont_grid scan with PSO line trigger and position capture

- A3200Motor: arm_line_trigger/disarm_line_trigger (PSO window on the fast
  axis drives DDG1 EXT), acceleration (RAMP RATE before each move),
  encoder_channel
- A3200PositionCapture: Npaq DATAACQ latches the encoder counters on each
  detector trigger (UINT1), read back per line via ARRAY READ and published
  as async data; offset calibrated with a software PSO fire
- aerotech_cont_grid: cont_grid geometry, line start triggered by position
  instead of a software sleep
- simulator support for RAMP, $global, UNITSTOCOUNTS, PSO, DATAACQ, ARRAY READ
- configs, tests, README with wiring and commissioning checklist

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

csaxs_bec/device_configs/sastt_aerotech.yaml

@@ -3,7 +3,9 @@
 # (FireWire to the Npaq). BEC connects to the A3200 ASCII command interface on that PC.
 # A3200 setup: CommandSetup = 0x00010004 (enabled + Always Send EOS), CommandPort = 8000.
 #
-# TODO commissioning: host name of the A3200 PC, A3200 axis names, velocities, limits.
+# TODO commissioning: host name of the A3200 PC, A3200 axis names, encoder channels,
+# velocities (= maximum scan velocity), accelerations, limits.
+# Wiring for aerotech_cont_grid: PSO output (J7) -> DDG1 EXT, detector trigger -> UINT1 (J12).
 # UU (ADRS-150 slipring follower) follows U on the controller and is not a BEC device.
 # Limits [0, 0] = no BEC soft limits; the controller's SoftwareLimitLow/High still apply.
 ############################################################
@@ -14,9 +16,11 @@ aerox:
   deviceConfig:
     axis_Id: X
     axis_nr: 0
+    encoder_channel: 0
+    acceleration: 100.0
     host: a3200-saxstt.psi.ch
     port: 8000
-    velocity: 1.0
+    velocity: 10.0
     egu: mm
     limits:
     - 0
@@ -34,6 +38,8 @@ aeroy:
   deviceConfig:
     axis_Id: Y
     axis_nr: 1
+    encoder_channel: 1
+    acceleration: 20.0
     host: a3200-saxstt.psi.ch
     port: 8000
     velocity: 0.5
@@ -54,9 +60,11 @@ aeroz:
   deviceConfig:
     axis_Id: Z
     axis_nr: 2
+    encoder_channel: 2
+    acceleration: 100.0
     host: a3200-saxstt.psi.ch
     port: 8000
-    velocity: 1.0
+    velocity: 10.0
     egu: mm
     limits:
     - 0
@@ -74,6 +82,8 @@ aerorot:
   deviceConfig:
     axis_Id: U
     axis_nr: 3
+    encoder_channel: 3
+    acceleration: 360.0
     host: a3200-saxstt.psi.ch
     port: 8000
     velocity: 10.0
@@ -89,3 +99,21 @@ aerorot:
   readoutPriority: baseline
   deviceTags:
     - sastt_aerotech
+aerocapture:
+  description: Aerotech position capture (Npaq DATAACQ, latched on the detector trigger at UINT1)
+  deviceClass: csaxs_bec.devices.aerotech.a3200.A3200PositionCapture
+  deviceConfig:
+    host: a3200-saxstt.psi.ch
+    port: 8000
+    drive_axis: X
+    channels:
+      0: X
+      1: Y
+      2: Z
+      3: U
+  enabled: true
+  onFailure: buffer
+  readOnly: false
+  readoutPriority: async
+  deviceTags:
+    - sastt_aerotech

csaxs_bec/device_configs/simulated_omny/simulated_sastt_aerotech.yaml

@@ -9,9 +9,11 @@ aerox:
   deviceConfig:
     axis_Id: X
     axis_nr: 0
+    encoder_channel: 0
+    acceleration: 100.0
     host: sim-a3200-saxstt
     port: 8000
-    velocity: 1.0
+    velocity: 10.0
     egu: mm
     limits:
     - 0
@@ -29,6 +31,8 @@ aeroy:
   deviceConfig:
     axis_Id: Y
     axis_nr: 1
+    encoder_channel: 1
+    acceleration: 20.0
     host: sim-a3200-saxstt
     port: 8000
     velocity: 0.5
@@ -49,9 +53,11 @@ aeroz:
   deviceConfig:
     axis_Id: Z
     axis_nr: 2
+    encoder_channel: 2
+    acceleration: 100.0
     host: sim-a3200-saxstt
     port: 8000
-    velocity: 1.0
+    velocity: 10.0
     egu: mm
     limits:
     - 0
@@ -69,6 +75,8 @@ aerorot:
   deviceConfig:
     axis_Id: U
     axis_nr: 3
+    encoder_channel: 3
+    acceleration: 360.0
     host: sim-a3200-saxstt
     port: 8000
     velocity: 10.0
@@ -84,3 +92,21 @@ aerorot:
   readoutPriority: baseline
   deviceTags:
     - sastt_aerotech
+aerocapture:
+  description: Aerotech position capture (Npaq DATAACQ, latched on the detector trigger at UINT1)
+  deviceClass: csaxs_bec.devices.sim.sim_a3200.SimA3200PositionCapture
+  deviceConfig:
+    host: sim-a3200-saxstt
+    port: 8000
+    drive_axis: X
+    channels:
+      0: X
+      1: Y
+      2: Z
+      3: U
+  enabled: true
+  onFailure: buffer
+  readOnly: false
+  readoutPriority: async
+  deviceTags:
+    - sastt_aerotech

csaxs_bec/devices/aerotech/README.md

@@ -11,6 +11,7 @@ A3200 ASCII command interface on the Windows PC (TCP, default port 8000).
 | aeroz      | Z          | ATS20010 | linear                                         |
 | aerorot    | U          | ABRT-200 | rotation                                       |
 | -          | UU         | ADRS-150 | slipring follower of U, handled on the controller |
+| aerocapture | -         | -        | position capture (DATAACQ) of X, Y, Z, U       |
 
 Configs: `device_configs/sastt_aerotech.yaml` (hardware),
 `device_configs/simulated_omny/simulated_sastt_aerotech.yaml` (simulation, `sim_a3200.py`).
@@ -39,9 +40,42 @@ Documentation: A3200 6.04.004 help (`A3200.chm`, topic "ASCII Command Interface"
 7. Check `ProgramPositionFeedback` matches the position shown in Motion Composer.
 8. Check how UU follows U (gantry, slave or program) before rotating.
 
-## Next: continuous line scan
+## Scan: aerotech_cont_grid
 
-Fast axis moving continuously, slow axis stepping between lines, start of each line
-triggered by position with PSO (position synchronized output) on the fast axis:
-`PSOCONTROL`, `PSOWINDOW`/`PSODISTANCE`, `PSOOUTPUT` drive the digital output when the
-armed position is crossed. To be implemented once the hardware is reachable.
+`scans_v4/aerotech_cont_grid.py`, same arguments as `cont_grid`:
+
+```python
+scans.aerotech_cont_grid(dev.aerox, 0, 2, 0.01, dev.aeroy, 0, 1, 0.05, exp_time=0.01)
+```
+
+- fast axis moves continuously at `fast_step_size / exp_time`, the step axis steps between lines
+- line start is triggered by position: PSO window on the fast axis, the output goes high
+  at `fast_start - velocity * shutter_open_delay` and triggers DDG1 (external, rising edge),
+  so the first exposure starts at `fast_start`
+- run-up / run-out = ramp distance `v^2 / (2 * acceleration)` + 20 % + 50 ms at speed
+- `capture_positions=True` (default): device `aerocapture` latches the encoder counters of
+  all configured channels on each detector trigger (DATAACQ on UINT1, up to 16 kHz, drive
+  array of 2M elements). After each line they are read back, converted to user units and
+  published as async data `aerocapture_positions_enc<channel>`.
+- the counter-to-position offset is calibrated in pre_scan with one software PSO fire,
+  before DDG1 is switched to the external trigger
+
+Wiring:
+
+| Signal | From | To |
+|--------|------|----|
+| line trigger | Npaq PSO output (J7, differential / opto / open collector) | DDG1 EXT |
+| exposure trigger | detector trigger (DDG2 AB) | Npaq UINT1 (J12 pins 1+/2-, 5 V opto, 5-15 mA) |
+
+Commissioning of the scan:
+
+1. Encoder channel of each axis (`encoder_channel`, `aerocapture.channels`); check that
+   the counter direction matches the axis direction (calibration only fixes the offset).
+2. PSO output: select the output used (`PSOOUTPUT CONTROL`), PSO interlock driven or
+   disabled via the `IOSetup` axis parameter (Npaq manual), polarity (JP16).
+3. `acceleration` of the fast axis = the ramp rate BEC sets before each move.
+4. `velocity` of the fast axis is the maximum scan velocity.
+5. `GlobalDoubles` >= 256: BEC uses `$global[0..199]` for array reads and `$global[255]`
+   as scratch variable.
+6. Check one line with the scope: PSO edge at the trigger position, number of UINT1 edges
+   = frames per line, captured positions vs. requested positions.

csaxs_bec/devices/aerotech/a3200.py

@@ -27,6 +27,7 @@ from ophyd import Component as Cpt
 from ophyd import Device, PositionerBase, Signal
 from ophyd.status import wait as status_wait
 from ophyd.utils import LimitError, ReadOnlyError
+from ophyd_devices import AsyncMultiSignal, PSIDeviceBase
 from ophyd_devices.utils.controller import Controller, retry_once, threadlocked
 from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected
 from prettytable import PrettyTable
@@ -131,10 +132,17 @@ class A3200Controller(Controller):
         "print_command_history",
     ]
 
+    # $global variables used by BEC: one scratch variable for evaluate() and a block
+    # for drive array reads. The controller has GlobalDoubles (default 256) of them.
+    scratch_global = 255
+    array_global_base = 0
+    array_read_chunk = 200
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         if not hasattr(self, "_rx_buffer"):
             self._rx_buffer = ""
+            self._counts_per_unit = {}
 
     @threadlocked
     def socket_put(self, val: str) -> None:
@@ -304,6 +312,81 @@ class A3200Controller(Controller):
         if axes:
             self.command(f"ABORT {' '.join(axes)}")
 
+    def set_ramp_rate(self, axis: str, rate: float) -> None:
+        """Rate-based ramping for axis-based moves (MOVEABS), rate in user units/s^2."""
+        self.command(f"RAMP MODE RATE {axis}")
+        self.command(f"RAMP RATE {axis} {rate:.6f}")
+
+    # --- global variables (scratch space for functions and array reads) -----------
+
+    def get_globals(self, start: int, count: int) -> list[float]:
+        """Read $global[start] ... $global[start + count - 1] in one ~STATUS round trip."""
+        return self.get_status_items([(f"$global[{i}]",) for i in range(start, start + count)])
+
+    def evaluate(self, expression: str) -> float:
+        """Evaluate an AeroBasic expression on the controller, e.g. UNITSTOCOUNTS(X, 1)."""
+        self.command(f"$global[{self.scratch_global}] = {expression}")
+        return self.get_globals(self.scratch_global, 1)[0]
+
+    def counts_per_unit(self, axis: str) -> float:
+        """Encoder counts per user unit of an axis (cached, it is a fixed parameter)."""
+        if axis not in self._counts_per_unit:
+            self._counts_per_unit[axis] = self.evaluate(f"UNITSTOCOUNTS({axis}, 1)")
+        return self._counts_per_unit[axis]
+
+    # --- PSO (position synchronized output) -----------------------------------
+
+    def pso_arm_window(self, axis: str, encoder_channel: int, low: float, high: float) -> None:
+        """
+        PSO output high while the axis is inside [low, high] (user units, absolute).
+
+        The window counter is loaded with the current position (axis at rest), so the
+        range is given relative to it in encoder counts.
+        """
+        cpu = self.counts_per_unit(axis)
+        current = self.get_position(axis)
+        low_counts, high_counts = sorted(
+            (int(round((low - current) * cpu)), int(round((high - current) * cpu)))
+        )
+        self.command(f"PSOCONTROL {axis} RESET")
+        self.command(f"PSOWINDOW {axis} 1 INPUT {encoder_channel}")
+        self.command(f"PSOWINDOW {axis} 1 LOAD 0")
+        self.command(f"PSOWINDOW {axis} 1 RANGE {low_counts}, {high_counts}")
+        self.command(f"PSOOUTPUT {axis} WINDOW")
+        self.command(f"PSOCONTROL {axis} ARM")
+
+    def pso_off(self, axis: str) -> None:
+        self.command(f"PSOWINDOW {axis} 1 OFF")
+        self.command(f"PSOCONTROL {axis} OFF")
+
+    # --- DATAACQ (position capture into the drive array) -----------------------
+
+    def dataacq_arm(self, axis: str, trigger: int, channels: list[int], n_triggers: int) -> None:
+        """
+        Capture the encoder counters of `channels` on each trigger into the drive array,
+        starting at index 0. Npaq trigger 0 = Input 12 (UINT1) rising edge, 6 = PSO firing.
+        """
+        self.command(f"DATAACQ {axis} 1 OFF")
+        self.command(f"DATAACQ {axis} 1 TRIGGER {trigger}")
+        for channel in sorted(channels):
+            self.command(f"DATAACQ {axis} 1 INPUT {channel}")
+        self.command(f"DATAACQ {axis} 1 ARRAY 0, {n_triggers * len(channels)}")
+
+    def dataacq_off(self, axis: str) -> None:
+        self.command(f"DATAACQ {axis} 1 OFF")
+
+    def read_drive_array(self, axis: str, start: int, count: int) -> list[float]:
+        """Read drive array elements through $global[array_global_base ...] in chunks."""
+        values = []
+        chunk = self.array_read_chunk
+        for offset in range(0, count, chunk):
+            n = min(chunk, count - offset)
+            self.command(
+                f"ARRAY {axis} READ $global[{self.array_global_base}] {start + offset} {n} FAST"
+            )
+            values.extend(self.get_globals(self.array_global_base, n))
+        return values
+
     def describe(self) -> None:
         t = PrettyTable()
         t.title = f"{self.__class__.__name__} on {self.sock.host}:{self.sock.port}"
@@ -365,6 +448,9 @@ class A3200SetpointSignal(A3200SignalBase):
     @threadlocked
     def _socket_set(self, val: float) -> None:
         self.setpoint = val
+        acceleration = self.parent.motor_acceleration.get()
+        if acceleration:
+            self.controller.set_ramp_rate(self.parent.axis_Id, acceleration)
         self.controller.move_absolute(
             self.parent.axis_Id, val * self.parent.sign, self.parent.motor_velocity.get()
         )
@@ -398,14 +484,28 @@ class A3200Motor(Device, PositionerBase):
         velocity: move speed in user units per second (A3200 user units: mm or deg).
         egu: engineering units shown in BEC.
         require_homed: refuse moves while the axis is not homed.
+        acceleration: ramp rate for moves in user units/s^2; None keeps the controller's
+            ramp settings (DefaultRampRate).
+        encoder_channel: Npaq local encoder counter of this axis, needed for the PSO line
+            trigger (arm_line_trigger).
     """
 
-    USER_ACCESS = ["controller", "enable", "disable", "home", "fault_ack", "status"]
+    USER_ACCESS = [
+        "controller",
+        "enable",
+        "disable",
+        "home",
+        "fault_ack",
+        "status",
+        "arm_line_trigger",
+        "disarm_line_trigger",
+    ]
     readback = Cpt(A3200ReadbackSignal, signal_name="readback", kind="hinted")
     user_setpoint = Cpt(A3200SetpointSignal, signal_name="setpoint")
     motor_is_moving = Cpt(A3200MotorIsMoving, signal_name="motor_is_moving", kind="normal")
     homed = Cpt(A3200AxisHomed, signal_name="homed", kind="config")
     motor_velocity = Cpt(Signal, value=1.0, kind="config")
+    motor_acceleration = Cpt(Signal, value=0.0, kind="config")
     high_limit_travel = Cpt(Signal, value=0, kind="omitted")
     low_limit_travel = Cpt(Signal, value=0, kind="omitted")
 
@@ -432,6 +532,8 @@ class A3200Motor(Device, PositionerBase):
         egu="mm",
         tolerance=0.001,
         require_homed=True,
+        acceleration=None,
+        encoder_channel=None,
         socket_cls=SocketIO,
         device_manager=None,
         **kwargs,
@@ -445,6 +547,7 @@ class A3200Motor(Device, PositionerBase):
         self.sign = sign
         self.tolerance = tolerance
         self.require_homed = require_homed
+        self.encoder_channel = encoder_channel
         self._egu = egu
         self.device_manager = device_manager
 
@@ -459,6 +562,7 @@ class A3200Motor(Device, PositionerBase):
         )
         self.readback.name = self.name
         self.motor_velocity.put(velocity)
+        self.motor_acceleration.put(acceleration or 0.0)
         self.controller.subscribe(
             self._update_connection_state, event_type=self.SUB_CONNECTION_CHANGE
         )
@@ -566,5 +670,156 @@ class A3200Motor(Device, PositionerBase):
     def fault_ack(self) -> None:
         self.controller.fault_ack(self.axis_Id)
 
+    def arm_line_trigger(self, start: float, end: float) -> None:
+        """
+        Arm the PSO output to go high when the axis enters [start, end] and low when it
+        leaves it. The rising edge starts the exposure sequence of a scan line. The axis
+        must be at rest outside the window.
+        """
+        if self.encoder_channel is None:
+            raise A3200Error(f"{self.name}: encoder_channel is not configured, no PSO trigger.")
+        self.controller.pso_arm_window(
+            self.axis_Id, self.encoder_channel, start * self.sign, end * self.sign
+        )
+
+    def disarm_line_trigger(self) -> None:
+        self.controller.pso_off(self.axis_Id)
+
     def status(self) -> dict:
         return self.controller.axis_status(self.axis_Id)
+
+
+NPAQ_ENCODER_CHANNELS = 6
+DATAACQ_TRIGGER_UINT1_RISING = 0
+DATAACQ_TRIGGER_PSO_FIRING_EVENT = 6
+
+
+class A3200PositionCapture(PSIDeviceBase):
+    """
+    Latch axis positions on an external trigger (Npaq DATAACQ, drive array).
+
+    Each rising edge on the trigger input (default: Input 12 / UINT1, the detector
+    exposure trigger) stores the encoder counters of the configured channels in the
+    drive array. After a scan line, read_line() reads them back, converts them to user
+    units and publishes them as async data (signals enc0 ... enc5).
+
+    Args:
+        channels: Npaq local encoder counter -> A3200 axis name, e.g. {0: "X", 1: "Y"}.
+            Values are in controller coordinates (the BEC motor sign is not applied).
+        drive_axis: any axis of the Npaq, used to address the drive (shared array).
+        trigger_input: DATAACQ trigger, 0 = Input 12 (UINT1) rising edge.
+    """
+
+    USER_ACCESS = ["controller", "arm", "read_line", "calibrate", "disarm"]
+
+    positions = Cpt(
+        AsyncMultiSignal,
+        name="positions",
+        signals=[f"enc{i}" for i in range(NPAQ_ENCODER_CHANNELS)],
+        ndim=1,
+        async_update={"type": "add", "max_shape": [None]},
+        max_size=1000,
+        kind="normal",
+        doc="Positions latched at each trigger, in user units, one sub-signal per channel.",
+    )
+
+    def __init__(
+        self,
+        prefix="",
+        *,
+        name,
+        host="localhost",
+        port=8000,
+        channels=None,
+        drive_axis="X",
+        trigger_input=DATAACQ_TRIGGER_UINT1_RISING,
+        socket_cls=SocketIO,
+        device_manager=None,
+        scan_info=None,
+        **kwargs,
+    ):
+        self.controller = A3200Controller(
+            socket_cls=socket_cls, socket_host=host, socket_port=port, device_manager=device_manager
+        )
+        self.channels = {int(ch): str(axis) for ch, axis in (channels or {}).items()}
+        for ch in self.channels:
+            if not 0 <= ch < NPAQ_ENCODER_CHANNELS:
+                raise ValueError(f"Encoder channel {ch} out of range 0..{NPAQ_ENCODER_CHANNELS-1}")
+        self.drive_axis = drive_axis
+        self.trigger_input = trigger_input
+        self._offsets = {ch: 0.0 for ch in self.channels}
+        self._n_triggers = 0
+        self._correlation_group = "monitored"
+        super().__init__(
+            prefix, name=name, scan_info=scan_info, device_manager=device_manager, **kwargs
+        )
+        if len(self.channels) < 2:
+            raise ValueError("Configure at least two encoder channels for position capture.")
+        # only the configured channels are published (strict signal validation)
+        self.positions.signal_info.signals = self.positions._unify_signals(
+            [f"enc{ch}" for ch in sorted(self.channels)]
+        )
+
+    def wait_for_connection(self, all_signals=False, timeout: float = 30.0) -> None:
+        self.controller.on(timeout=timeout)
+
+    def destroy(self):
+        self.controller.off(update_config=False)
+        return super().destroy()
+
+    def on_stage(self) -> None:
+        frames = getattr(getattr(self, "scan_info", None), "msg", None)
+        frames_per_trigger = getattr(frames, "frames_per_trigger", 1) if frames else 1
+        self._correlation_group = "monitored" if frames_per_trigger == 1 else "burst_group"
+
+    def calibrate(self) -> dict[int, float]:
+        """
+        Determine the offset between encoder counts and program position of each channel:
+        one capture triggered by a software PSO fire with the axes at rest.
+
+        Fires the PSO output once: the line trigger receiver (DDG) must not be armed.
+        """
+        channels = sorted(self.channels)
+        self.controller.dataacq_arm(
+            self.drive_axis, DATAACQ_TRIGGER_PSO_FIRING_EVENT, channels, n_triggers=1
+        )
+        self.controller.command(f"PSOCONTROL {self.drive_axis} RESET")
+        self.controller.command(f"PSOCONTROL {self.drive_axis} FIRE")
+        time.sleep(0.05)
+        raw = self.controller.read_drive_array(self.drive_axis, 0, len(channels))
+        self.controller.dataacq_off(self.drive_axis)
+        for ch, counts in zip(channels, raw):
+            axis = self.channels[ch]
+            position = self.controller.get_position(axis)
+            self._offsets[ch] = position - counts / self.controller.counts_per_unit(axis)
+        return dict(self._offsets)
+
+    def arm(self, n_triggers: int) -> None:
+        """Arm the capture of the next n_triggers trigger edges (start of the array)."""
+        self._n_triggers = int(n_triggers)
+        self.controller.dataacq_arm(
+            self.drive_axis, self.trigger_input, sorted(self.channels), self._n_triggers
+        )
+
+    def disarm(self) -> None:
+        self.controller.dataacq_off(self.drive_axis)
+
+    def read_line(self) -> dict[str, list[float]]:
+        """Read the positions of the armed triggers and publish them. Returns them too."""
+        channels = sorted(self.channels)
+        raw = self.controller.read_drive_array(self.drive_axis, 0, self._n_triggers * len(channels))
+        # Npaq interleaves the inputs per trigger, in ascending channel order
+        data = np.asarray(raw, dtype=float).reshape(self._n_triggers, len(channels))
+        timestamp = time.time()
+        result = {}
+        for i, ch in enumerate(channels):
+            cpu = self.controller.counts_per_unit(self.channels[ch])
+            result[f"enc{ch}"] = data[:, i] / cpu + self._offsets[ch]
+        self.positions.put(
+            {
+                f"{self.positions.name}_{key}": {"value": value, "timestamp": timestamp}
+                for key, value in result.items()
+            },
+            correlation_group=self._correlation_group,
+        )
+        return {key: value.tolist() for key, value in result.items()}

csaxs_bec/devices/sim/sim_a3200.py

@@ -2,7 +2,9 @@
 Simulated Aerotech A3200 controller (ASCII command interface).
 
 Implements the command subset used by `A3200Controller`: ~VERSION, ~LASTERROR, ~STATUS,
-ENABLE, DISABLE, MOVEABS, MOVEINC, ABORT, HOMEASYNC, FAULTACK, ACKNOWLEDGEALL.
+ENABLE, DISABLE, MOVEABS, MOVEINC, ABORT, HOMEASYNC, FAULTACK, ACKNOWLEDGEALL, RAMP,
+$global assignments (numbers, UNITSTOCOUNTS), PSOCONTROL/PSOWINDOW/PSOOUTPUT (window mode),
+DATAACQ and ARRAY READ.
 Replies follow the real interface with "Always Send EOS": "%<data>\\n", "!\\n" or "#\\n".
 
 Simulated behavior:
@@ -10,6 +12,11 @@ Simulated behavior:
 - HOMEASYNC moves the axis to 0 at the home speed and sets the Homed bit
 - moving a disabled axis is a task fault ("#"), like on the real controller
 - faults can be injected via `SimA3200State.axis(name).fault`, FAULTACK clears them
+- encoder counter of a channel = position * counts_per_unit + counter_offset
+- detector exposures: when a move starts with the PSO window armed and DATAACQ armed on
+  UINT1, the armed number of triggers is latched immediately, evenly spread over the
+  PSO window of the moving axis (other channels at their current position)
+- PSOCONTROL FIRE latches one capture if DATAACQ is armed on the PSO firing event
 """
 
 from __future__ import annotations
@@ -27,12 +34,15 @@ from csaxs_bec.devices.aerotech.a3200 import (
     AXISSTATUS_MOVE_DONE,
     DRIVESTATUS_ENABLED,
     A3200Motor,
+    A3200PositionCapture,
 )
 from csaxs_bec.devices.sim.sim_socket import SimSocketBase, SimStateRegistry
 
 logger = bec_logger.logger
 
 _STATUS_ITEM_RE = re.compile(r"\(([^)]*)\)")
+_GLOBAL_RE = re.compile(r"^\$global\[(\d+)\]$", re.IGNORECASE)
+_UNITSTOCOUNTS_RE = re.compile(r"^UNITSTOCOUNTS\(\s*(\w+)\s*,\s*([-+0-9.eE]+)\s*\)$", re.I)
 
 
 class _SimA3200Axis:
@@ -45,6 +55,9 @@ class _SimA3200Axis:
         self.homing = False
         self.fault = 0
         self.home_velocity = 10.0
+        self.counts_per_unit = 10000.0
+        self.counter_offset = 0
+        self.ramp_rate = None
         self.pos = 0.0
         self._move_start = 0.0
         self._move_target = 0.0
@@ -105,6 +118,46 @@ class SimA3200State:
         self.port = port
         self._axes = {}
         self._lock = threading.RLock()
+        self.globals = {}
+        self.channel_axis = {}  # Npaq encoder channel -> axis name
+        self.drive_array = [0.0] * 4096
+        self.pso = {"armed": False, "mode": None, "channel": None, "range": None, "load": 0}
+        self.pso_window_entries = 0  # rising edges of the PSO window output (line triggers)
+        self.dataacq = {"armed": False, "trigger": None, "inputs": set(), "n_elements": 0}
+
+    def counter(self, channel: int) -> int:
+        ax = self.axis(self.channel_axis[channel])
+        return int(round(ax.position() * ax.counts_per_unit)) + ax.counter_offset
+
+    def latch(self, positions: dict[str, float] | None = None, index: int = 0) -> None:
+        """Store one DATAACQ sample (all inputs, ascending channel) at trigger `index`."""
+        inputs = sorted(self.dataacq["inputs"])
+        for i, ch in enumerate(inputs):
+            ax = self.axis(self.channel_axis[ch])
+            if positions and ax.name in positions:
+                value = int(round(positions[ax.name] * ax.counts_per_unit)) + ax.counter_offset
+            else:
+                value = self.counter(ch)
+            self.drive_array[index * len(inputs) + i] = float(value)
+
+    def on_move_start(self, ax: _SimA3200Axis, start: float, target: float) -> None:
+        """Simulate the line trigger and the detector exposures of a scan line."""
+        pso, acq = self.pso, self.dataacq
+        if not (pso["armed"] and pso["mode"] == "window" and pso["range"] is not None):
+            return
+        if self.channel_axis.get(pso["channel"]) != ax.name:
+            return
+        low = start + pso["range"][0] / ax.counts_per_unit
+        high = start + pso["range"][1] / ax.counts_per_unit
+        if not (min(start, target) <= low and high <= max(start, target)):
+            return
+        self.pso_window_entries += 1
+        if acq["armed"] and acq["trigger"] == 0 and acq["inputs"]:
+            n_triggers = acq["n_elements"] // len(acq["inputs"])
+            step = (high - low) / max(n_triggers, 1)
+            for k in range(n_triggers):
+                self.latch({ax.name: low + k * step}, index=k)
+            acq["armed"] = False
 
     def axis(self, name: str) -> _SimA3200Axis:
         with self._lock:
@@ -114,9 +167,23 @@ class SimA3200State:
             return self._axes[key]
 
     def configure_axis(
-        self, name, initial_position=None, enabled=None, homed=None, home_velocity=None
+        self,
+        name,
+        initial_position=None,
+        enabled=None,
+        homed=None,
+        home_velocity=None,
+        counts_per_unit=None,
+        counter_offset=None,
+        encoder_channel=None,
     ):
         ax = self.axis(name)
+        if counts_per_unit is not None:
+            ax.counts_per_unit = float(counts_per_unit)
+        if counter_offset is not None:
+            ax.counter_offset = int(counter_offset)
+        if encoder_channel is not None:
+            self.channel_axis[int(encoder_channel)] = ax.name
         if initial_position is not None:
             ax.pos = float(initial_position)
         if enabled is not None:
@@ -178,7 +245,9 @@ class SimA3200Socket(SimSocketBase):
             if velocity == 0:
                 raise _TaskFault("Speed of zero is not allowed.")
             ax = self._ready_axis(name)
-            target = value if cmd == "MOVEABS" else ax.position() + value
+            start = ax.position()
+            target = value if cmd == "MOVEABS" else start + value
+            state.on_move_start(ax, start, target)
             ax.start_move(target, velocity)
             return ""
         if cmd == "HOMEASYNC":
@@ -200,9 +269,82 @@ class SimA3200Socket(SimSocketBase):
             for ax in state._axes.values():
                 ax.fault = 0
             return ""
+        if cmd == "RAMP":
+            # RAMP MODE RATE <axis> | RAMP RATE <axis> <value>
+            if args[0].upper() == "RATE" and len(args) == 3:
+                state.axis(args[1]).ramp_rate = float(args[2])
+            return ""
+        if _GLOBAL_RE.match(words[0]) and len(words) >= 3 and words[1] == "=":
+            index = int(_GLOBAL_RE.match(words[0]).group(1))
+            state.globals[index] = self._evaluate(line.split("=", 1)[1].strip())
+            return ""
+        if cmd == "PSOCONTROL":
+            return self._psocontrol(args)
+        if cmd == "PSOWINDOW":
+            # PSOWINDOW <axis> 1 INPUT <ch> | LOAD <v> | RANGE <lo>, <hi> | OFF | ON
+            sub = args[2].upper()
+            if sub == "INPUT":
+                state.pso["channel"] = int(args[3])
+            elif sub == "LOAD":
+                state.pso["load"] = int(args[3])
+            elif sub == "RANGE":
+                low, high = (float(v) for v in " ".join(args[3:]).split(","))
+                state.pso["range"] = (low, high)
+            elif sub == "OFF":
+                state.pso["range"] = None
+            return ""
+        if cmd == "PSOOUTPUT":
+            state.pso["mode"] = args[1].lower()
+            return ""
+        if cmd == "DATAACQ":
+            # DATAACQ <axis> 1 TRIGGER <t> | INPUT <ch> | ARRAY <start>, <n> | OFF
+            sub = args[2].upper()
+            acq = state.dataacq
+            if sub == "OFF":
+                acq.update(armed=False, trigger=None, inputs=set(), n_elements=0)
+            elif sub == "TRIGGER":
+                acq["trigger"] = int(args[3])
+            elif sub == "INPUT":
+                ch = int(args[3])
+                if ch not in state.channel_axis:
+                    raise _TaskFault(f"No simulated axis on encoder channel {ch}.")
+                acq["inputs"].add(ch)
+            elif sub == "ARRAY":
+                _, n_elements = (int(v) for v in " ".join(args[3:]).split(","))
+                acq.update(armed=True, n_elements=n_elements)
+            return ""
+        if cmd == "ARRAY":
+            # ARRAY <axis> READ $global[<i>] <start> <n> [FAST]
+            base = int(_GLOBAL_RE.match(args[2]).group(1))
+            start, count = int(args[3]), int(args[4])
+            for i in range(count):
+                state.globals[base + i] = state.drive_array[start + i]
+            return ""
         logger.warning(f"[sim a3200] {self.host}:{self.port} unhandled command '{line}'")
         return None
 
+    def _psocontrol(self, args):
+        state = self.state
+        mode = args[1].upper()
+        if mode == "RESET":
+            state.pso.update(armed=False, mode=None, channel=None, range=None, load=0)
+        elif mode == "ARM":
+            state.pso["armed"] = True
+        elif mode == "OFF":
+            state.pso["armed"] = False
+        elif mode == "FIRE":
+            acq = state.dataacq
+            if acq["armed"] and acq["trigger"] == 6:
+                state.latch()
+                acq["armed"] = False
+        return ""
+
+    def _evaluate(self, expression: str) -> float:
+        match = _UNITSTOCOUNTS_RE.match(expression)
+        if match:
+            return float(match.group(2)) * self.state.axis(match.group(1)).counts_per_unit
+        return float(expression)
+
     @staticmethod
     def _require_axes(args):
         if not args:
@@ -219,6 +361,8 @@ class SimA3200Socket(SimSocketBase):
 
     def _status_item(self, item: str) -> str:
         parts = [p.strip() for p in item.split(",")]
+        if len(parts) == 1 and _GLOBAL_RE.match(parts[0]):
+            return f"{self.state.globals.get(int(_GLOBAL_RE.match(parts[0]).group(1)), 0.0):.6f}"
         if len(parts) < 2:
             raise ValueError(f"Unsupported status item ({item})")
         ax = self.state.axis(parts[0])
@@ -262,11 +406,15 @@ class SimA3200Motor(A3200Motor):
         egu="mm",
         tolerance=0.001,
         require_homed=True,
+        acceleration=None,
+        encoder_channel=None,
         device_manager=None,
         sim_initial_position=None,
         sim_enabled=True,
         sim_homed=True,
         sim_home_velocity=None,
+        sim_counts_per_unit=None,
+        sim_counter_offset=None,
         **kwargs,
     ):
         super().__init__(
@@ -282,6 +430,8 @@ class SimA3200Motor(A3200Motor):
             egu=egu,
             tolerance=tolerance,
             require_homed=require_homed,
+            acceleration=acceleration,
+            encoder_channel=encoder_channel,
             socket_cls=SimA3200Socket,
             device_manager=device_manager,
             **kwargs,
@@ -295,4 +445,42 @@ class SimA3200Motor(A3200Motor):
             enabled=sim_enabled,
             homed=sim_homed,
             home_velocity=sim_home_velocity,
+            counts_per_unit=sim_counts_per_unit,
+            counter_offset=sim_counter_offset,
+            encoder_channel=encoder_channel,
         )
+
+
+class SimA3200PositionCapture(A3200PositionCapture):
+    """A3200PositionCapture connected to the simulated A3200 controller."""
+
+    def __init__(
+        self,
+        prefix="",
+        *,
+        name,
+        host="localhost",
+        port=8000,
+        channels=None,
+        drive_axis="X",
+        trigger_input=0,
+        device_manager=None,
+        scan_info=None,
+        **kwargs,
+    ):
+        super().__init__(
+            prefix,
+            name=name,
+            host=host,
+            port=port,
+            channels=channels,
+            drive_axis=drive_axis,
+            trigger_input=trigger_input,
+            socket_cls=SimA3200Socket,
+            device_manager=device_manager,
+            scan_info=scan_info,
+            **kwargs,
+        )
+        state = SimStateRegistry.get(SimA3200State, host, port)
+        for channel, axis in self.channels.items():
+            state.channel_axis[channel] = axis.upper()

csaxs_bec/scans/__init__.py

@@ -2,4 +2,5 @@ from .flomni_fermat_scan import FlomniFermatScan
 from .lamni_fermat_scan import LamniFermatScan
 from .lamni_move_to_scan_center import LamniMoveToScanCenter
 from .omny_fermat_scan import OmnyFermatScan
+from .scans_v4.aerotech_cont_grid import AerotechContGrid
 from .scans_v4.cont_grid import ContGrid

csaxs_bec/scans/scans_v4/aerotech_cont_grid.py

@@ -0,0 +1,369 @@
+"""
+Continuous grid scan for Aerotech A3200 stages with a position triggered line start.
+
+Same geometry as cont_grid (fast axis moving continuously, step axis stepping between
+lines), but the start of each line is triggered in hardware: the A3200 PSO output goes
+high when the fast axis enters the line window and triggers DDG1 (EXT input, rising edge),
+which runs the shutter / detector / MCS sequence. With a position capture device, the
+encoder positions of the axes are latched on each detector trigger (Npaq UINT1 input)
+and published per line.
+
+Wiring:
+    Npaq PSO output (J7) -> DDG1 EXT/EN
+    detector trigger (DDG2 AB) -> Npaq UINT1 (J12 pins 1/2, 5 V opto input)
+
+Timing per line:
+    trigger position = fast_start - velocity * shutter_open_delay, so the first exposure
+    starts at fast_start; the fast axis starts one ramp distance (plus margin) before the
+    trigger position and stops one ramp distance after the last exposure.
+
+Scan procedure:
+    - prepare_scan
+    - open_scan
+    - stage
+    - pre_scan
+    - scan_core
+        - at_each_point (called by scan_core for each line)
+    - post_scan
+    - unstage
+    - close_scan
+    - on_exception (called if any exception is raised during the scan)
+"""
+
+from __future__ import annotations
+
+from copy import deepcopy
+from typing import Annotated, TypedDict
+
+import numpy as np
+from bec_lib.device import DeviceBase
+from bec_lib.logger import bec_logger
+from bec_lib.scan_args import DefaultArgType, ScanArgument
+from bec_server.scan_server.errors import ScanAbortion
+from bec_server.scan_server.scans import position_generators
+from bec_server.scan_server.scans.scan_base import ScanBase, ScanType
+from bec_server.scan_server.scans.scan_modifier import scan_hook
+
+from csaxs_bec.scans.scan_customization.scan_components import CsaxsBecScanComponents
+
+logger = bec_logger.logger
+
+# DDG trigger sources (csaxs_bec.devices.epics.delay_generator_csaxs.TRIGGERSOURCE)
+DDG_EXT_RISING_EDGE = 1
+DDG_SINGLE_SHOT = 5
+
+# Extra run-up / run-out beyond the ramp distance, as a fraction of it, and as settling time
+RAMP_MARGIN_FRACTION = 0.2
+SETTLE_TIME = 0.05
+
+
+class AerotechLineParameter(TypedDict):
+    """Line geometry of the fast axis, computed in prepare_scan."""
+
+    original_velocity: float | None
+    target_velocity: float | None
+    acceleration: float | None
+    shutter_open_delay: float | None
+    trigger_position: float | None
+    exposure_end: float | None
+    line_start: float | None
+    line_end: float | None
+    num_lines: int | None
+
+
+class AerotechContGrid(ScanBase):
+    # One DDG1 sequence per line, like cont_grid: the MCS card and DDG1 count each line as
+    # one point with frames_per_trigger frames.
+    scan_type = ScanType.SOFTWARE_TRIGGERED
+    scan_name = "aerotech_cont_grid"
+
+    gui_config = {
+        "Scan Parameters": [
+            "fast_axis",
+            "fast_start",
+            "fast_end",
+            "fast_step_size",
+            "stepper_axis",
+            "stepper_start",
+            "stepper_stop",
+            "stepper_step_size",
+        ],
+        "Acquisition Parameters": ["exp_time", "relative", "capture_positions"],
+    }
+
+    def __init__(
+        # fmt: off
+        self,
+        fast_axis: Annotated[DeviceBase, ScanArgument(display_name="Fast Axis", description="A3200 axis with continuous motion and PSO line trigger.")],
+        fast_start: Annotated[float, ScanArgument(display_name="Fast Start", description="Start position for measurement points of the fast axis.")],
+        fast_end: Annotated[float, ScanArgument(display_name="Fast End", description="End position for measurement points of the fast axis.")],
+        fast_step_size: Annotated[float, ScanArgument(display_name="Fast Step Size", description="Distance travelled by the fast axis during one exposure.")],
+        stepper_axis: Annotated[DeviceBase, ScanArgument(display_name="Step Axis", description="Step axis of the grid scan, stepping through the lines.")],
+        stepper_start: Annotated[float, ScanArgument(display_name="Step Start", description="Start position of the step axis.")],
+        stepper_stop: Annotated[float, ScanArgument(display_name="Step Stop", description="End position of the step axis.")],
+        stepper_step_size: Annotated[float, ScanArgument(display_name="Step Step Size", description="Step size of the step axis in units of the motor.")],
+        exp_time: DefaultArgType.ExposureTime,
+        relative: DefaultArgType.Relative = False,
+        capture_positions: bool = True,
+        capture_device: str = "aerocapture",
+        **kwargs,
+        # fmt: on
+    ):
+        """
+        Continuous grid scan for A3200 stages: the PSO output of the fast axis triggers
+        DDG1 at the start of each line, optionally the axis positions are latched at each
+        detector trigger.
+
+        Args:
+            fast_axis (DeviceBase): A3200 axis with continuous motion (needs encoder_channel and acceleration configured).
+            fast_start (float): Start position for measurement points of the fast axis.
+            fast_end (float): End position for measurement points of the fast axis.
+            fast_step_size (float): Distance travelled by the fast axis during one exposure.
+            stepper_axis (DeviceBase): Step axis of the grid scan, stepping through the lines.
+            stepper_start (float): Start position of the step axis.
+            stepper_stop (float): End position of the step axis.
+            stepper_step_size (float): Step size of the step axis in units of the motor.
+            exp_time (float): Exposure time in seconds
+            relative (bool): Whether the positions are relative to the current position. Default is False.
+            capture_positions (bool): Latch the axis positions at each detector trigger. Default is True.
+            capture_device (str): Name of the A3200PositionCapture device. Default is "aerocapture".
+
+        Returns:
+            ScanReport
+        """
+        super().__init__(**kwargs)
+        self.components = CsaxsBecScanComponents(self)
+        self._baseline_readout_status = None
+        self.fast_axis = fast_axis
+        if fast_end < fast_start:  # the line always runs in positive direction
+            fast_start, fast_end = fast_end, fast_start
+        self.fast_start = fast_start
+        self.fast_end = fast_end
+        self.fast_step_size = fast_step_size
+        self.stepper_axis = stepper_axis
+        self.stepper_start = stepper_start
+        self.stepper_stop = stepper_stop
+        self.stepper_step_size = stepper_step_size
+        self.relative = relative
+        self.exp_time = exp_time
+        self.capture_positions = capture_positions
+        self.ddg1 = self.device_manager.devices["ddg1"]
+        self.mcs = self.device_manager.devices["mcs"]
+        self.capture = None
+        if capture_positions:
+            if capture_device not in self.device_manager.devices:
+                raise ScanAbortion(
+                    f"Position capture device '{capture_device}' not found."
+                    " Use capture_positions=False to scan without it."
+                )
+            self.capture = self.device_manager.devices[capture_device]
+
+        self.motors = [fast_axis, stepper_axis]
+        self.frames_per_line = 0
+        self._line: AerotechLineParameter = AerotechLineParameter()
+        self.update_scan_info(relative=relative, exp_time=exp_time)
+
+    @scan_hook
+    def prepare_scan(self):
+        """Compute positions and line geometry, start the premove."""
+        self._check_motor_inputs()
+        self.frames_per_line = (
+            int(np.ceil(np.abs(self.fast_end - self.fast_start) / self.fast_step_size)) + 1
+        )  # include last point
+        self._line["num_lines"] = (
+            int(np.ceil(np.abs(self.stepper_stop - self.stepper_start) / self.stepper_step_size))
+            + 1
+        )  # include last point
+        positions = position_generators.nd_grid_positions(
+            [
+                (self.fast_start, self.fast_end, self.frames_per_line),
+                (self.stepper_start, self.stepper_stop, self._line["num_lines"]),
+            ],
+            snaked=False,
+        )
+
+        if self.relative:
+            self.start_positions = self.components.get_start_positions(self.motors)
+            positions += self.start_positions
+            self.fast_start += self.start_positions[0]
+            self.fast_end += self.start_positions[0]
+            self.stepper_start += self.start_positions[1]
+            self.stepper_stop += self.start_positions[1]
+
+        self._compute_line_geometry()
+
+        # One readout per line: the end point of each line, as in cont_grid
+        self.positions = deepcopy(positions[(self.frames_per_line - 1) :: self.frames_per_line, :])
+
+        self.actions.set_device_readout_priority(self.motors, priority="monitored")
+        self.update_scan_info(
+            positions=self.positions,
+            num_points=len(self.positions),
+            frames_per_trigger=self.frames_per_line,
+            computed_positions=positions,
+            num_lines=self._line["num_lines"],
+        )
+
+        self.actions.add_scan_report_instruction_device_progress(self.mcs)
+        self._baseline_readout_status = self.actions.read_baseline_devices(wait=False)
+        self._premove_motor_status = self.actions.set(
+            self.motors, [self._line["line_start"], self.stepper_start], wait=False
+        )
+
+    @scan_hook
+    def open_scan(self):
+        """Open the scan."""
+        self.actions.open_scan()
+
+    @scan_hook
+    def stage(self):
+        """Stage all devices. DDG1 is set to its default (single shot) trigger here."""
+        self.actions.stage_all_devices()
+
+    @scan_hook
+    def pre_scan(self):
+        """
+        Calibrate the position capture while DDG1 still ignores external triggers (the
+        calibration fires the PSO output once), then switch DDG1 to the external trigger.
+        """
+        self._premove_motor_status.wait()
+        if self.capture is not None:
+            self.capture.calibrate()
+        self.ddg1.set_trigger(DDG_EXT_RISING_EDGE)
+        self.actions.pre_scan_all_devices()
+
+    @scan_hook
+    def scan_core(self):
+        """One continuous line per step axis position."""
+        for line_index in range(self._line["num_lines"]):
+            status_motor = self.actions.set(
+                self.motors, [self._line["line_start"], self.positions[line_index][1]]
+            )
+            status_motor.wait()
+            status_mcs = self.ddg1.prepare_mcs_on_trigger()
+            status_mcs.wait()
+            mcs_acquiring_status = self.mcs.get_transition_status(
+                signal_name="acquiring", transition=[1, 0]
+            )
+            self.at_each_point(
+                motors=[self.fast_axis],
+                positions=np.array([self._line["line_end"]]),
+                status_mcs=mcs_acquiring_status,
+            )
+
+    @scan_hook
+    def at_each_point(
+        self,
+        motors: list[str | DeviceBase],
+        positions: np.ndarray,
+        status_mcs,
+        last_positions: np.ndarray | None = None,
+    ):
+        """
+        Arm the line trigger (and the position capture), run the line at scan velocity,
+        wait for the MCS card, then read the captured positions.
+        """
+        if self.capture is not None:
+            self.capture.arm(self.frames_per_line)
+        self.fast_axis.arm_line_trigger(self._line["trigger_position"], self._line["exposure_end"])
+        self.fast_axis.motor_velocity.set(self._line["target_velocity"]).wait(timeout=5)
+        try:
+            move_status = self.actions.set(motors, positions, wait=False)
+            while not move_status.done:
+                self.actions.read_monitored_devices(wait=True)
+                try:
+                    move_status.wait(timeout=0.5)
+                except TimeoutError:
+                    continue
+        finally:
+            self.fast_axis.motor_velocity.set(self._line["original_velocity"]).wait(timeout=5)
+            self.fast_axis.disarm_line_trigger()
+
+        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
+
+        if self.capture is not None:
+            self.capture.read_line()
+
+    @scan_hook
+    def post_scan(self):
+        """Restore DDG1 trigger and fast axis velocity, complete devices."""
+        self._restore()
+        status = self.actions.complete_all_devices(wait=False)
+        if self.relative:
+            self.components.move_and_wait(self.motors, self.start_positions)
+        status.wait()
+
+    @scan_hook
+    def unstage(self):
+        """Unstage all devices."""
+        self.actions.unstage_all_devices()
+
+    @scan_hook
+    def close_scan(self):
+        """Close the scan."""
+        if self._baseline_readout_status is not None:
+            self._baseline_readout_status.wait()
+        self.actions.close_scan()
+        self.actions.check_for_unchecked_statuses()
+
+    @scan_hook
+    def on_exception(self, exception: Exception):
+        """Return to a safe state: no external DDG1 trigger, PSO off, velocity restored."""
+        self._restore()
+        if self.relative:
+            self.components.move_and_wait(self.motors, self.start_positions)
+
+    #######################################################
+    ######### Helper methods for the scan logic ###########
+    #######################################################
+
+    def _check_motor_inputs(self):
+        if np.isclose(self.fast_start, self.fast_end, atol=self.fast_step_size):
+            raise ScanAbortion(
+                f"Fast stop {self.fast_end} and fast start {self.fast_start} positions are too close for the given step size {self.fast_step_size}."
+            )
+        if np.isclose(self.stepper_start, self.stepper_stop, atol=self.stepper_step_size):
+            raise ScanAbortion(
+                f"Stepper stop {self.stepper_stop} and stepper start {self.stepper_start} positions are too close for the given step size {self.stepper_step_size}."
+            )
+
+    def _compute_line_geometry(self):
+        """Velocity, trigger position and run-up / run-out of the fast axis."""
+        line = self._line
+        line["shutter_open_delay"] = self.ddg1.get_shutter_to_open_delay()
+        line["original_velocity"] = self.fast_axis.motor_velocity.get()
+        line["acceleration"] = self.fast_axis.motor_acceleration.get()
+
+        velocity = self.fast_step_size / self.exp_time
+        if velocity > line["original_velocity"]:
+            raise ScanAbortion(
+                f"Requested velocity of {velocity} exceeds the velocity {line['original_velocity']}"
+                f" configured for {self.fast_axis.name}."
+            )
+        if not line["acceleration"] or line["acceleration"] <= 0:
+            raise ScanAbortion(
+                f"No acceleration configured for {self.fast_axis.name}; it is needed for the"
+                " run-up distance of the line."
+            )
+        ramp_distance = velocity**2 / (2 * line["acceleration"])
+        margin = RAMP_MARGIN_FRACTION * ramp_distance + velocity * SETTLE_TIME
+
+        line["target_velocity"] = velocity
+        line["trigger_position"] = self.fast_start - velocity * line["shutter_open_delay"]
+        line["exposure_end"] = self.fast_start + self.frames_per_line * self.fast_step_size
+        line["line_start"] = line["trigger_position"] - ramp_distance - margin
+        line["line_end"] = line["exposure_end"] + ramp_distance + margin
+
+    def _restore(self):
+        try:
+            self.ddg1.set_trigger(DDG_SINGLE_SHOT)
+        finally:
+            if self._line.get("original_velocity") is not None:
+                self.fast_axis.motor_velocity.put(self._line["original_velocity"])
+            self.fast_axis.disarm_line_trigger()

tests/tests_devices/test_a3200.py

@@ -10,7 +10,7 @@ from csaxs_bec.devices.aerotech.a3200 import (
     A3200Motor,
     decode_axis_fault,
 )
-from csaxs_bec.devices.sim.sim_a3200 import SimA3200Motor, SimA3200State
+from csaxs_bec.devices.sim.sim_a3200 import SimA3200Motor, SimA3200PositionCapture, SimA3200State
 from csaxs_bec.devices.sim.sim_socket import SimStateRegistry
 
 _HOST = "test-a3200.psi.ch"
@@ -205,3 +205,112 @@ def test_sim_two_axes_share_one_controller():
     assert x.readback.get() == pytest.approx(1.0)
     assert y.readback.get() == pytest.approx(2.0)
     x.controller.describe()
+
+
+# --- line trigger and position capture (simulated controller) ----------------
+
+
+def _sim_state():
+    return SimStateRegistry.get(SimA3200State, _HOST, _PORT)
+
+
+def _capture_setup():
+    x = _sim_motor(
+        "X", axis_nr=0, encoder_channel=0, sim_counts_per_unit=20000, sim_counter_offset=777
+    )
+    y = _sim_motor(
+        "Y", axis_nr=1, encoder_channel=1, sim_initial_position=0.3, sim_counter_offset=-55
+    )
+    capture = SimA3200PositionCapture(
+        name="aerocapture",
+        host=_HOST,
+        port=_PORT,
+        channels={0: "X", 1: "Y"},
+        device_manager=DMMock(),
+    )
+    capture.wait_for_connection()
+    return x, y, capture
+
+
+def test_counts_per_unit_is_read_once():
+    x, _, _ = _capture_setup()
+    assert x.controller.counts_per_unit("X") == 20000
+    x.controller.command_history.clear()
+    assert x.controller.counts_per_unit("X") == 20000
+    assert not x.controller.command_history
+
+
+def test_ramp_rate_is_set_before_the_move():
+    motor = _sim_motor(acceleration=50.0)
+    motor.move(1.0, wait=True)
+    history = list(motor.controller.command_history)
+    ramp = history.index("[PUT]: RAMP RATE X 50.000000")
+    move = next(i for i, c in enumerate(history) if c.startswith("[PUT]: MOVEABS X"))
+    assert ramp < move
+    assert _sim_state().axis("X").ramp_rate == 50.0
+
+
+def test_line_trigger_window_is_relative_to_current_position():
+    x, _, _ = _capture_setup()
+    x.move(-1.0, wait=True)
+    x.arm_line_trigger(0.0, 1.0)
+    assert _sim_state().pso["range"] == (20000.0, 40000.0)
+    assert _sim_state().pso["armed"]
+    x.move(2.0, wait=True)
+    assert _sim_state().pso_window_entries == 1
+    x.disarm_line_trigger()
+    assert not _sim_state().pso["armed"]
+
+
+def test_line_trigger_needs_encoder_channel():
+    motor = _sim_motor()
+    with pytest.raises(A3200Error, match="encoder_channel"):
+        motor.arm_line_trigger(0.0, 1.0)
+
+
+def test_capture_calibration_removes_counter_offset():
+    _, _, capture = _capture_setup()
+    offsets = capture.calibrate()
+    assert offsets[0] == pytest.approx(-777 / 20000)
+    assert offsets[1] == pytest.approx(55 / 10000)
+
+
+def test_capture_line_publishes_positions():
+    x, _, capture = _capture_setup()
+    capture.calibrate()
+    x.move(-1.0, wait=True)
+    capture.arm(5)
+    x.arm_line_trigger(0.0, 1.0)
+    x.move(2.0, wait=True)
+
+    result = capture.read_line()
+
+    assert result["enc0"] == pytest.approx([0.0, 0.2, 0.4, 0.6, 0.8])
+    assert result["enc1"] == pytest.approx([0.3] * 5)
+    msg = capture.positions.get()
+    assert set(msg.signals) == {"aerocapture_positions_enc0", "aerocapture_positions_enc1"}
+    assert msg.metadata["async_update"] == {"type": "add", "max_shape": [None]}
+
+
+def test_capture_reads_in_chunks():
+    x, _, capture = _capture_setup()
+    capture.controller.array_read_chunk = 3
+    x.move(-1.0, wait=True)
+    capture.arm(4)
+    x.arm_line_trigger(0.0, 1.0)
+    x.move(2.0, wait=True)
+    result = capture.read_line()
+    assert len(result["enc0"]) == 4
+    reads = [c for c in capture.controller.command_history if "ARRAY X READ" in c]
+    assert reads == [
+        "[PUT]: ARRAY X READ $global[0] 0 3 FAST",
+        "[PUT]: ARRAY X READ $global[0] 3 3 FAST",
+        "[PUT]: ARRAY X READ $global[0] 6 2 FAST",
+    ]
+
+
+def test_capture_needs_two_channels():
+    with pytest.raises(ValueError, match="two encoder channels"):
+        SimA3200PositionCapture(
+            name="aerocapture", host=_HOST, port=_PORT, channels={0: "X"}, device_manager=DMMock()
+        )

tests/tests_scans/test_aerotech_cont_grid_scan.py

@@ -0,0 +1,185 @@
+from unittest import mock
+
+import numpy as np
+import pytest
+from bec_server.scan_server.errors import ScanAbortion
+from bec_server.scan_server.tests.scan_fixtures import MockCustomDevice
+from bec_server.scan_server.tests.scan_hook_tests import (
+    DEFAULT_HOOK_TESTS,
+    PREMOVE_HOOK_TESTS,
+    run_scan_tests,
+)
+
+AEROTECH_CONT_GRID_HOOK_TESTS = [*DEFAULT_HOOK_TESTS, *PREMOVE_HOOK_TESTS]
+
+
+def _signal(obj_name, kind="config"):
+    return {"obj_name": obj_name, "kind_str": kind, "describe": {"precision": 3}}
+
+
+def _add_devices(device_manager, capture=True, acceleration=100.0):
+    aerox = MockCustomDevice(
+        "aerox",
+        device_info={
+            "signals": {
+                "readback": _signal("aerox", "hinted"),
+                "motor_velocity": _signal("aerox_motor_velocity"),
+                "motor_acceleration": _signal("aerox_motor_acceleration"),
+            }
+        },
+        signal_read_values={
+            "aerox": 0.0,
+            "aerox_motor_velocity": 10.0,
+            "aerox_motor_acceleration": acceleration,
+        },
+    )
+    aerox.arm_line_trigger = mock.MagicMock()
+    aerox.disarm_line_trigger = mock.MagicMock()
+    aeroy = MockCustomDevice(
+        "aeroy",
+        device_info={"signals": {"readback": _signal("aeroy", "hinted")}},
+        signal_read_values={"aeroy": 0.0},
+    )
+    ddg1 = MockCustomDevice(
+        "ddg1",
+        device_info={"signals": {"readback": _signal("ddg1", "hinted")}},
+        signal_read_values={"ddg1": 0.0},
+    )
+    ddg1.get_shutter_to_open_delay = mock.MagicMock(return_value=2e-3)
+    ddg1.set_trigger = mock.MagicMock()
+    mcs = MockCustomDevice(
+        "mcs",
+        device_info={"signals": {"readback": _signal("mcs", "hinted")}},
+        signal_read_values={"mcs": 0.0},
+    )
+    devices = [aerox, aeroy, ddg1, mcs]
+    if capture:
+        aerocapture = MockCustomDevice(
+            "aerocapture",
+            device_info={"signals": {"readback": _signal("aerocapture", "hinted")}},
+            signal_read_values={"aerocapture": 0.0},
+        )
+        for method in ("calibrate", "arm", "read_line", "disarm"):
+            setattr(aerocapture, method, mock.MagicMock())
+        devices.append(aerocapture)
+    for device in devices:
+        device_manager.add_device(device, replace=True)
+
+
+def _assemble(v4_scan_assembler, device_manager, **kwargs):
+    _add_devices(
+        device_manager, **{k: kwargs.pop(k) for k in ("capture", "acceleration") if k in kwargs}
+    )
+    return v4_scan_assembler(
+        "aerotech_cont_grid", "aerox", -1.0, 1.0, 0.5, "aeroy", -2.0, 2.0, 2, exp_time=0.1, **kwargs
+    )
+
+
+@pytest.mark.parametrize(("hook_name", "hook_tests"), AEROTECH_CONT_GRID_HOOK_TESTS)
+def test_aerotech_cont_grid_default_hooks(
+    v4_scan_assembler, device_manager, nth_done_status_mock, hook_name, hook_tests
+):
+    scan = _assemble(v4_scan_assembler, device_manager)
+    run_scan_tests(scan, [(hook_name, hook_tests)], nth_done_status_mock=nth_done_status_mock)
+
+
+def test_line_geometry(v4_scan_assembler, device_manager):
+    scan = _assemble(v4_scan_assembler, device_manager)
+    scan.prepare_scan()
+    line = scan._line
+
+    velocity = 0.5 / 0.1
+    ramp = velocity**2 / (2 * 100.0)
+    margin = 0.2 * ramp + velocity * 0.05
+    assert scan.frames_per_line == 5
+    assert line["num_lines"] == 3
+    assert line["target_velocity"] == pytest.approx(velocity)
+    # first exposure starts at fast_start: trigger one shutter delay earlier
+    assert line["trigger_position"] == pytest.approx(-1.0 - velocity * 2e-3)
+    assert line["exposure_end"] == pytest.approx(-1.0 + 5 * 0.5)
+    assert line["line_start"] == pytest.approx(line["trigger_position"] - ramp - margin)
+    assert line["line_end"] == pytest.approx(line["exposure_end"] + ramp + margin)
+    assert np.array_equal(scan.positions, np.array([[1.0, -2.0], [1.0, 0.0], [1.0, 2.0]]))
+    assert scan.scan_info.frames_per_trigger == 5
+
+
+def test_velocity_above_configured_velocity_aborts(v4_scan_assembler, device_manager):
+    _add_devices(device_manager)
+    scan = v4_scan_assembler(
+        "aerotech_cont_grid", "aerox", -10.0, 10.0, 2.0, "aeroy", -2.0, 2.0, 2, exp_time=0.1
+    )
+    with pytest.raises(ScanAbortion, match="exceeds the velocity"):
+        scan.prepare_scan()
+
+
+def test_missing_acceleration_aborts(v4_scan_assembler, device_manager):
+    scan = _assemble(v4_scan_assembler, device_manager, acceleration=0.0)
+    with pytest.raises(ScanAbortion, match="No acceleration"):
+        scan.prepare_scan()
+
+
+def test_missing_capture_device_aborts(v4_scan_assembler, device_manager):
+    device_manager.devices.pop("aerocapture", None)
+    with pytest.raises(ScanAbortion, match="aerocapture"):
+        _assemble(v4_scan_assembler, device_manager, capture=False)
+
+
+def test_scan_without_capture(v4_scan_assembler, device_manager):
+    device_manager.devices.pop("aerocapture", None)
+    scan = _assemble(v4_scan_assembler, device_manager, capture=False, capture_positions=False)
+    assert scan.capture is None
+
+
+def test_pre_scan_calibrates_before_ddg_goes_external(
+    v4_scan_assembler, device_manager, nth_done_status_mock
+):
+    scan = _assemble(v4_scan_assembler, device_manager)
+    scan.prepare_scan()
+    order = mock.MagicMock()
+    scan.capture = order.capture
+    scan.ddg1 = order.ddg1
+    scan._premove_motor_status = nth_done_status_mock(resolve_after=1)
+    scan.actions.pre_scan_all_devices = mock.MagicMock()
+
+    scan.pre_scan()
+
+    assert order.mock_calls[:2] == [mock.call.capture.calibrate(), mock.call.ddg1.set_trigger(1)]
+
+
+def test_at_each_point_arms_trigger_and_reads_capture(
+    v4_scan_assembler, device_manager, nth_done_status_mock
+):
+    scan = _assemble(v4_scan_assembler, device_manager)
+    scan.prepare_scan()
+    scan.capture = mock.MagicMock()
+    scan.fast_axis = mock.MagicMock()
+    scan.actions.set = mock.MagicMock(return_value=nth_done_status_mock(resolve_after=1))
+    scan.actions.read_monitored_devices = mock.MagicMock()
+
+    scan.at_each_point(
+        [scan.fast_axis],
+        np.array([scan._line["line_end"]]),
+        status_mcs=nth_done_status_mock(resolve_after=1),
+    )
+
+    scan.capture.arm.assert_called_once_with(5)
+    scan.fast_axis.arm_line_trigger.assert_called_once_with(
+        scan._line["trigger_position"], scan._line["exposure_end"]
+    )
+    velocity_sets = [c.args[0] for c in scan.fast_axis.motor_velocity.set.call_args_list]
+    assert velocity_sets == [scan._line["target_velocity"], scan._line["original_velocity"]]
+    scan.fast_axis.disarm_line_trigger.assert_called_once_with()
+    scan.capture.read_line.assert_called_once_with()
+
+
+def test_on_exception_restores_ddg_and_pso(v4_scan_assembler, device_manager):
+    scan = _assemble(v4_scan_assembler, device_manager)
+    scan.prepare_scan()
+    scan.ddg1 = mock.MagicMock()
+    scan.fast_axis = mock.MagicMock()
+
+    scan.on_exception(RuntimeError("test"))
+
+    scan.ddg1.set_trigger.assert_called_once_with(5)
+    scan.fast_axis.disarm_line_trigger.assert_called_once_with()
+    scan.fast_axis.motor_velocity.put.assert_called_once_with(10.0)