From fe36f374fb16edb3f0ab94fc5ef6ac6e1728793e Mon Sep 17 00:00:00 2001 From: Mirko Holler Date: Tue, 27 Oct 2026 18:28:29 +0100 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01M8zoDyJxbBLmjZCEPUo5FM --- csaxs_bec/device_configs/sastt_aerotech.yaml | 34 +- .../simulated_sastt_aerotech.yaml | 30 +- csaxs_bec/devices/aerotech/README.md | 44 ++- csaxs_bec/devices/aerotech/a3200.py | 257 +++++++++++- csaxs_bec/devices/sim/sim_a3200.py | 194 ++++++++- csaxs_bec/scans/__init__.py | 1 + .../scans/scans_v4/aerotech_cont_grid.py | 369 ++++++++++++++++++ tests/tests_devices/test_a3200.py | 111 +++++- .../test_aerotech_cont_grid_scan.py | 185 +++++++++ 9 files changed, 1210 insertions(+), 15 deletions(-) create mode 100644 csaxs_bec/scans/scans_v4/aerotech_cont_grid.py create mode 100644 tests/tests_scans/test_aerotech_cont_grid_scan.py diff --git a/csaxs_bec/device_configs/sastt_aerotech.yaml b/csaxs_bec/device_configs/sastt_aerotech.yaml index dc5b4f88..72e2dca9 100644 --- a/csaxs_bec/device_configs/sastt_aerotech.yaml +++ b/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 diff --git a/csaxs_bec/device_configs/simulated_omny/simulated_sastt_aerotech.yaml b/csaxs_bec/device_configs/simulated_omny/simulated_sastt_aerotech.yaml index b89ffc68..d18b2971 100644 --- a/csaxs_bec/device_configs/simulated_omny/simulated_sastt_aerotech.yaml +++ b/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 diff --git a/csaxs_bec/devices/aerotech/README.md b/csaxs_bec/devices/aerotech/README.md index e1aa544d..347dae79 100644 --- a/csaxs_bec/devices/aerotech/README.md +++ b/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`. +- 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. diff --git a/csaxs_bec/devices/aerotech/a3200.py b/csaxs_bec/devices/aerotech/a3200.py index be51d51d..6d5a84dc 100644 --- a/csaxs_bec/devices/aerotech/a3200.py +++ b/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()} diff --git a/csaxs_bec/devices/sim/sim_a3200.py b/csaxs_bec/devices/sim/sim_a3200.py index 40a0f802..18f37910 100644 --- a/csaxs_bec/devices/sim/sim_a3200.py +++ b/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": "%\\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 | RAMP RATE + 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 1 INPUT | LOAD | RANGE , | 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 1 TRIGGER | INPUT | ARRAY , | 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 READ $global[] [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() diff --git a/csaxs_bec/scans/__init__.py b/csaxs_bec/scans/__init__.py index 60909788..802a5c1d 100644 --- a/csaxs_bec/scans/__init__.py +++ b/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 diff --git a/csaxs_bec/scans/scans_v4/aerotech_cont_grid.py b/csaxs_bec/scans/scans_v4/aerotech_cont_grid.py new file mode 100644 index 00000000..338f5407 --- /dev/null +++ b/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() diff --git a/tests/tests_devices/test_a3200.py b/tests/tests_devices/test_a3200.py index eada5f51..9567841c 100644 --- a/tests/tests_devices/test_a3200.py +++ b/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() + ) diff --git a/tests/tests_scans/test_aerotech_cont_grid_scan.py b/tests/tests_scans/test_aerotech_cont_grid_scan.py new file mode 100644 index 00000000..2962a448 --- /dev/null +++ b/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)