bec/csaxs_bec
0
0 Fork 3
Code Issues 17 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

revert(aerotech): remove the RIO burst box trigger source
CI for csaxs_bec / test (push) Successful in 3m22s
Details

Browse Source 
This reverts commit bf77d6f6fe.

The RIO timing (1 ms steps, ~50 Hz opto outputs) has too much jitter to judge
whether the position sampling works, so the test setup will not use it. The
DMC program stays in the OMNY Galil repository (RioDDG).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01M8zoDyJxbBLmjZCEPUo5FM
This commit is contained in:
Mirko HollerandClaude Opus 5.5 committed 2026-10-28 06:16:12 +01:00
1 parent bf77d6f6fe
commit 669a7ea02b
9 files changed
+22 -797
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

csaxs_bec/device_configs/sastt_aerotech_test.yaml
-134
View File
@@ -1,134 +0,0 @@
############################################################
# SAXS-TT Aerotech stages, TEST SETUP without beamline timing (no DDG / MCS / shutter):
# Galil RIO-47120 trigger box (program RioDDG/rio_ddg.dmc, OMNY Galil repo) as line
# burst generator. Scan: aerotech_cont_grid(..., trigger_source="rio")
#
# Wiring: Npaq PSO output (J7) -> RIO analog input 1 (TTL ok, threshold 3 V)
# RIO DO1 (12-24 V, series resistor!) or AO5 (0/5 V) -> Npaq UINT1 (J12)
#
# A3200 setup: CommandSetup = 0x00010004 (enabled + Always Send EOS), CommandPort = 8000.
# TODO: host names of the A3200 PC and the RIO, axis names, encoder channels,
# velocities (= maximum scan velocity), accelerations, limits.
############################################################
aerox:
description: Aerotech X, ALS25020 linear stage
deviceClass: csaxs_bec.devices.aerotech.a3200.A3200Motor
deviceConfig:
axis_Id: X
axis_nr: 0
encoder_channel: 0
acceleration: 100.0
host: a3200-saxstt.psi.ch
port: 8000
velocity: 10.0
egu: mm
limits:
- 0
- 0
sign: 1
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: baseline
deviceTags:
- sastt_aerotech
aeroy:
description: Aerotech Y, AVSI125 vertical lift
deviceClass: csaxs_bec.devices.aerotech.a3200.A3200Motor
deviceConfig:
axis_Id: Y
axis_nr: 1
encoder_channel: 1
acceleration: 20.0
host: a3200-saxstt.psi.ch
port: 8000
velocity: 0.5
egu: mm
limits:
- 0
- 0
sign: 1
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: baseline
deviceTags:
- sastt_aerotech
aeroz:
description: Aerotech Z, ATS20010 linear stage
deviceClass: csaxs_bec.devices.aerotech.a3200.A3200Motor
deviceConfig:
axis_Id: Z
axis_nr: 2
encoder_channel: 2
acceleration: 100.0
host: a3200-saxstt.psi.ch
port: 8000
velocity: 10.0
egu: mm
limits:
- 0
- 0
sign: 1
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: baseline
deviceTags:
- sastt_aerotech
aerorot:
description: Aerotech U, ABRT-200 rotation stage (UU follows on the controller)
deviceClass: csaxs_bec.devices.aerotech.a3200.A3200Motor
deviceConfig:
axis_Id: U
axis_nr: 3
encoder_channel: 3
acceleration: 360.0
host: a3200-saxstt.psi.ch
port: 8000
velocity: 10.0
egu: deg
tolerance: 0.001
limits:
- 0
- 0
sign: 1
enabled: true
onFailure: buffer
readOnly: false
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
rioddg:
description: Galil RIO-47120 trigger box, line burst generator (rio_ddg.dmc)
deviceClass: csaxs_bec.devices.omny.galil.galil_rio_ddg.GalilRIOBurst
deviceConfig:
host: rio-saxstt.psi.ch
port: 23
min_period_ms: 20
threshold: 3.0
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: baseline
deviceTags:
- sastt_aerotech
csaxs_bec/device_configs/simulated_omny/simulated_sastt_aerotech.yaml
-18
View File
@@ -1,8 +1,6 @@
############################################################
# Simulated SAXS-TT Aerotech stages (sim_a3200), same devices as sastt_aerotech.yaml.
# Axes start enabled and homed (sim_enabled / sim_homed in deviceConfig to change).
# rioddg: simulated RIO burst box, triggered by the PSO output of the simulated A3200.
# Scan without hardware: aerotech_cont_grid(..., trigger_source="rio")
############################################################
aerox:
@@ -112,19 +110,3 @@ aerocapture:
readoutPriority: async
deviceTags:
- sastt_aerotech
rioddg:
description: Galil RIO-47120 trigger box, line burst generator (simulated)
deviceClass: csaxs_bec.devices.sim.sim_rio_ddg.SimGalilRIOBurst
deviceConfig:
host: sim-rio-saxstt
port: 23
min_period_ms: 20
threshold: 3.0
sim_trigger_host: sim-a3200-saxstt
sim_trigger_port: 8000
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: baseline
deviceTags:
- sastt_aerotech
csaxs_bec/devices/aerotech/README.md
+2 -22
View File
@@ -13,13 +13,8 @@ A3200 ASCII command interface on the Windows PC (TCP, default port 8000).
| - | UU | ADRS-150 | slipring follower of U, handled on the controller |
| aerocapture | - | - | position capture (DATAACQ) of X, Y, Z, U |
Configs:
| Config | Use |
|--------|-----|
| `device_configs/sastt_aerotech.yaml` | beamline: A3200 + position capture, timing by DDG1/DDG2/MCS |
| `device_configs/sastt_aerotech_test.yaml` | test setup: A3200 + position capture + Galil RIO burst box (`rioddg`) |
| `device_configs/simulated_omny/simulated_sastt_aerotech.yaml` | simulation of the test setup (`sim_a3200.py`, `sim_rio_ddg.py`) |
Configs: `device_configs/sastt_aerotech.yaml` (hardware),
`device_configs/simulated_omny/simulated_sastt_aerotech.yaml` (simulation, `sim_a3200.py`).
Documentation: A3200 6.04.004 help (`A3200.chm`, topic "ASCII Command Interface"), copy at
`~/gitea/SAXS-TT_docs` on the workstation.
@@ -84,18 +79,3 @@ Commissioning of the scan:
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.
## Test setup without beamline timing: trigger_source="rio"
`aerotech_cont_grid(..., trigger_source="rio")` replaces DDG1/DDG2/MCS by the Galil
RIO-47120 trigger box (device `rioddg`, class `GalilRIOBurst`, program
`RioDDG/rio_ddg.dmc` in the OMNY Galil repository, branch `feat/rio-ddg`):
- pre_scan: capture calibration, then the RIO is switched to burst mode (`HX0`, `XQ#BURST,0`)
- per line: `arm_burst(frames_per_line, exp_time in ms)`, PSO line trigger -> RIO analog
input 1 -> pulses on DO1 / AO5 -> Npaq UINT1 (position capture); BEC waits until the
RIO reports all pulses sent
- post_scan / exception: RIO back to the legacy delay generator (`XQ#DDG,0`)
- exp_time in whole ms, >= 20 ms (opto outputs ~50 Hz); no shutter delay compensation
In the simulation the RIO is triggered by the PSO window of the simulated A3200.
csaxs_bec/devices/omny/galil/galil_rio_ddg.py
-129
View File
@@ -1,129 +0,0 @@
"""
Galil RIO-47120 trigger box used as a line burst generator (stand-in for DDG1/DDG2 in
test setups without the beamline timing).
Runs the program RioDDG/rio_ddg.dmc of the OMNY Galil repository. Thread 0 is either in
the legacy delay generator mode (#DDG, default after power up) or in the burst mode
(#BURST). In burst mode, per line BEC sets the number of pulses, period and width and arms
the box; on the rising edge of analog input 1 (line trigger, e.g. the Aerotech PSO output)
the program sends the pulses on digital output 1 and analog output 5.
"""
from __future__ import annotations
import time
from bec_lib.logger import bec_logger
from csaxs_bec.devices.omny.galil.galil_rio import GalilRIO
logger = bec_logger.logger
class GalilRIOBurstError(Exception):
pass
class GalilRIOBurst(GalilRIO):
"""
Galil RIO trigger box with the line burst generator of rio_ddg.dmc.
Args:
min_period_ms: shortest allowed pulse period; the opto outputs (DO1) have ~50 Hz
bandwidth, so 20 ms by default.
threshold: trigger threshold of analog input 1 in V.
"""
USER_ACCESS = [
"start_burst_mode",
"stop_burst_mode",
"arm_burst",
"disarm_burst",
"burst_status",
"burst_done",
"wait_burst_done",
]
def __init__(
self,
*,
name: str,
host: str,
device_manager,
port: int | None = None,
min_period_ms: int = 20,
threshold: float = 3.0,
**kwargs,
):
super().__init__(name=name, host=host, device_manager=device_manager, port=port, **kwargs)
self.min_period_ms = int(min_period_ms)
self.threshold = float(threshold)
def _get_var(self, var: str) -> float:
return float(self.controller.socket_put_and_receive(f"MG {var}").strip())
def _set_var(self, var: str, value) -> None:
self.controller.socket_put_confirmed(f"{var}={value}")
def start_burst_mode(self) -> None:
"""Switch thread 0 from the legacy delay generator to the burst generator."""
self.controller.socket_put_confirmed("HX0")
self.controller.socket_put_confirmed("XQ#BURST,0")
time.sleep(0.05)
def stop_burst_mode(self) -> None:
"""Back to the legacy delay generator (#DDG)."""
self.controller.socket_put_confirmed("HX0")
self.controller.socket_put_confirmed("XQ#DDG,0")
def arm_burst(self, n_pulses: int, period_ms: int, width_ms: int | None = None) -> None:
"""Arm one burst of n_pulses, started by the next rising edge of the trigger input."""
period_ms = int(round(period_ms))
if width_ms is None:
width_ms = period_ms // 2
width_ms = int(round(width_ms))
if period_ms < self.min_period_ms:
raise GalilRIOBurstError(
f"Pulse period {period_ms} ms is below the minimum of {self.min_period_ms} ms."
)
if not 1 <= width_ms < period_ms:
raise GalilRIOBurstError(f"Pulse width {width_ms} ms must be in [1, {period_ms}).")
if n_pulses < 1:
raise GalilRIOBurstError(f"Number of pulses must be at least 1, got {n_pulses}.")
if self._get_var("busy"):
raise GalilRIOBurstError("RIO is still sending the pulses of the previous line.")
self._set_var("nfr", int(n_pulses))
self._set_var("per", period_ms)
self._set_var("wid", width_ms)
self._set_var("thr", self.threshold)
self._set_var("armed", 1)
def disarm_burst(self) -> None:
self._set_var("armed", 0)
def burst_status(self) -> dict:
reply = self.controller.socket_put_and_receive("MG armed, busy, nsent, nline")
armed, busy, nsent, nline = (float(v) for v in reply.split())
return {"armed": bool(armed), "busy": bool(busy), "nsent": int(nsent), "nline": int(nline)}
def burst_done(self) -> bool:
"""True once the armed burst was triggered and all its pulses are sent."""
status = self.burst_status()
return not status["armed"] and not status["busy"]
def wait_burst_done(self, n_pulses: int, timeout: float) -> dict:
"""Wait for the burst to finish, check that n_pulses were sent."""
start = time.time()
while True:
status = self.burst_status()
if not status["armed"] and not status["busy"]:
break
if time.time() - start > timeout:
raise GalilRIOBurstError(
f"Burst not finished within {timeout} s: {status}. Was the line trigger"
" received on analog input 1?"
)
time.sleep(0.05)
if status["nsent"] != n_pulses:
raise GalilRIOBurstError(f"Sent {status['nsent']} pulses instead of {n_pulses}.")
return status
csaxs_bec/devices/sim/sim_a3200.py
+2 -9
View File
@@ -123,8 +123,6 @@ class SimA3200State:
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)
# callables(trigger_time) receiving the PSO output edge, e.g. a simulated trigger box
self.line_trigger_listeners = []
self.dataacq = {"armed": False, "trigger": None, "inputs": set(), "n_elements": 0}
def counter(self, channel: int) -> int:
@@ -142,9 +140,7 @@ class SimA3200State:
value = self.counter(ch)
self.drive_array[index * len(inputs) + i] = float(value)
def on_move_start(
self, ax: _SimA3200Axis, start: float, target: float, velocity: float
) -> None:
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):
@@ -156,9 +152,6 @@ class SimA3200State:
if not (min(start, target) <= low and high <= max(start, target)):
return
self.pso_window_entries += 1
trigger_time = time.time() + abs(low - start) / abs(velocity)
for listener in list(self.line_trigger_listeners):
listener(trigger_time)
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)
@@ -254,7 +247,7 @@ class SimA3200Socket(SimSocketBase):
ax = self._ready_axis(name)
start = ax.position()
target = value if cmd == "MOVEABS" else start + value
state.on_move_start(ax, start, target, velocity)
state.on_move_start(ax, start, target)
ax.start_move(target, velocity)
return ""
if cmd == "HOMEASYNC":
csaxs_bec/devices/sim/sim_rio_ddg.py
-156
View File
@@ -1,156 +0,0 @@
"""
Simulated Galil RIO trigger box running RioDDG/rio_ddg.dmc (OMNY Galil repository).
Implements the commands used by `GalilRIOBurst` and `GalilRIO`: variable assignments,
"MG var, var, ...", HX0, XQ#BURST,0, XQ#DDG,0, SB/CB, MG@OUT[n], MG@AN[...].
The line trigger (analog input 1) comes from a simulated A3200 controller: the device is
registered as listener of the PSO output of `SimA3200State` (host/port given in the
device config). A burst is evaluated lazily from the trigger time: armed is cleared at the
trigger, nsent counts the periods elapsed since then, busy is set until nfr * per ms.
"""
from __future__ import annotations
import re
import threading
import time
from bec_lib.logger import bec_logger
from csaxs_bec.devices.omny.galil.galil_rio_ddg import GalilRIOBurst
from csaxs_bec.devices.sim.sim_a3200 import SimA3200State
from csaxs_bec.devices.sim.sim_socket import SimSocketBase, SimStateRegistry
logger = bec_logger.logger
_ASSIGNMENT_RE = re.compile(r"^([A-Za-z][A-Za-z0-9]*)\s*=\s*(-?[0-9.]+)$")
class SimRioDDGState:
"""Shared state of one simulated RIO with the rio_ddg.dmc program."""
def __init__(self, host, port):
self.host = host
self.port = port
self._lock = threading.RLock()
self.mode = "ddg"
self.vars = {}
self.digital_outputs = [0.0] * 16
self.analog_inputs = [0.0] * 8
self._burst = None # (trigger_time, nfr, per)
self._reset_burst_vars()
self.vars.update(nfr=1.0, per=100.0, wid=10.0, thr=3.0, delay=100.0)
def _reset_burst_vars(self):
self.vars.update(armed=0.0, busy=0.0, nsent=0.0, nline=0.0)
self._burst = None
def start(self, label: str):
with self._lock:
self.mode = "burst" if label == "#BURST" else "ddg"
if self.mode == "burst":
self._reset_burst_vars()
def halt(self):
with self._lock:
self.mode = None
def line_trigger(self, trigger_time: float):
"""Rising edge on analog input 1 at trigger_time."""
with self._lock:
self._refresh()
if self.mode != "burst" or not self.vars["armed"] or self._burst is not None:
return
self._burst = (trigger_time, int(self.vars["nfr"]), self.vars["per"])
def _refresh(self):
if self._burst is None:
return
trigger_time, nfr, per = self._burst
elapsed_ms = (time.time() - trigger_time) * 1000.0
if elapsed_ms < 0:
return
self.vars["armed"] = 0.0
if elapsed_ms >= nfr * per:
self.vars.update(busy=0.0, nsent=float(nfr), nline=self.vars["nline"] + 1)
self._burst = None
else:
self.vars.update(busy=1.0, nsent=float(min(nfr, int(elapsed_ms // per) + 1)))
def get(self, var: str) -> float:
with self._lock:
self._refresh()
return self.vars.get(var, 0.0)
def set(self, var: str, value: float):
with self._lock:
self._refresh()
self.vars[var] = value
class SimRioDDGSocket(SimSocketBase):
state_cls = SimRioDDGState
def handle_command(self, line: str):
state: SimRioDDGState = self.state
if line == "HX0":
state.halt()
return ":"
if line.startswith("XQ"):
state.start(line[2:].split(",")[0].strip())
return ":"
if line.startswith("SB") or line.startswith("CB"):
state.digital_outputs[int(line[2:])] = 1.0 if line.startswith("SB") else 0.0
return ":"
if line.startswith("MG"):
items = [item.strip() for item in line[2:].split(",")]
values = []
for item in items:
if item.startswith("@OUT["):
values.append(state.digital_outputs[int(item[5:].split("]")[0])])
elif item.startswith("@AN["):
values.append(state.analog_inputs[int(item[4:].split("]")[0])])
else:
values.append(state.get(item))
return " ".join(f"{v:.4f}" for v in values)
match = _ASSIGNMENT_RE.match(line)
if match:
state.set(match.group(1), float(match.group(2)))
return ":"
logger.warning(f"[sim rio ddg] {self.host}:{self.port} unhandled command '{line}'")
return "?"
class SimGalilRIOBurst(GalilRIOBurst):
"""GalilRIOBurst connected to a simulated RIO; its trigger input follows the PSO output
of the simulated A3200 controller at (sim_trigger_host, sim_trigger_port)."""
def __init__(
self,
*,
name: str,
host: str,
device_manager,
port: int | None = None,
min_period_ms: int = 20,
threshold: float = 3.0,
sim_trigger_host: str | None = None,
sim_trigger_port: int = 8000,
**kwargs,
):
super().__init__(
name=name,
host=host,
device_manager=device_manager,
port=port,
min_period_ms=min_period_ms,
threshold=threshold,
socket_cls=SimRioDDGSocket,
**kwargs,
)
state = SimStateRegistry.get(SimRioDDGState, host, port or 23)
if sim_trigger_host is not None:
a3200 = SimStateRegistry.get(SimA3200State, sim_trigger_host, sim_trigger_port)
if state.line_trigger not in a3200.line_trigger_listeners:
a3200.line_trigger_listeners.append(state.line_trigger)
csaxs_bec/scans/scans_v4/aerotech_cont_grid.py
+18 -71
View File
@@ -8,16 +8,10 @@ which runs the shutter / detector / MCS sequence. With a position capture device
encoder positions of the axes are latched on each detector trigger (Npaq UINT1 input)
and published per line.
Wiring (trigger_source="ddg", beamline):
Wiring:
Npaq PSO output (J7) -> DDG1 EXT/EN
detector trigger (DDG2 AB) -> Npaq UINT1 (J12 pins 1/2, 5 V opto input)
Wiring (trigger_source="rio", test setup without beamline timing):
Npaq PSO output (J7) -> RIO analog input 1
RIO digital output 1 / analog output 5 -> Npaq UINT1 (and detector, if any)
The RIO (program RioDDG/rio_ddg.dmc) sends frames_per_line pulses with period exp_time
(1 ms steps, >= 20 ms) on each line trigger; no DDG, MCS or shutter is used.
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
@@ -58,8 +52,6 @@ logger = bec_logger.logger
DDG_EXT_RISING_EDGE = 1
DDG_SINGLE_SHOT = 5
TRIGGER_SOURCES = ("ddg", "rio")
# 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
@@ -72,7 +64,6 @@ class AerotechLineParameter(TypedDict):
target_velocity: float | None
acceleration: float | None
shutter_open_delay: float | None
rio_period_ms: int | None
trigger_position: float | None
exposure_end: float | None
line_start: float | None
@@ -97,7 +88,7 @@ class AerotechContGrid(ScanBase):
"stepper_stop",
"stepper_step_size",
],
"Acquisition Parameters": ["exp_time", "relative", "capture_positions", "trigger_source"],
"Acquisition Parameters": ["exp_time", "relative", "capture_positions"],
}
def __init__(
@@ -115,8 +106,6 @@ class AerotechContGrid(ScanBase):
relative: DefaultArgType.Relative = False,
capture_positions: bool = True,
capture_device: str = "aerocapture",
trigger_source: str = "ddg",
rio_device: str = "rioddg",
**kwargs,
# fmt: on
):
@@ -138,8 +127,6 @@ class AerotechContGrid(ScanBase):
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".
trigger_source (str): "ddg" (DDG1/DDG2 + MCS, beamline) or "rio" (RIO burst box, test setup). Default is "ddg".
rio_device (str): Name of the GalilRIOBurst device for trigger_source="rio". Default is "rioddg".
Returns:
ScanReport
@@ -160,19 +147,8 @@ class AerotechContGrid(ScanBase):
self.relative = relative
self.exp_time = exp_time
self.capture_positions = capture_positions
if trigger_source not in TRIGGER_SOURCES:
raise ScanAbortion(
f"Unknown trigger_source '{trigger_source}', use one of {TRIGGER_SOURCES}."
)
self.trigger_source = trigger_source
self.ddg1 = self.mcs = self.rio = None
if trigger_source == "ddg":
self.ddg1 = self.device_manager.devices["ddg1"]
self.mcs = self.device_manager.devices["mcs"]
else:
if rio_device not in self.device_manager.devices:
raise ScanAbortion(f"RIO burst device '{rio_device}' not found.")
self.rio = self.device_manager.devices[rio_device]
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:
@@ -228,8 +204,7 @@ class AerotechContGrid(ScanBase):
num_lines=self._line["num_lines"],
)
if self.mcs is not None:
self.actions.add_scan_report_instruction_device_progress(self.mcs)
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
@@ -248,16 +223,13 @@ class AerotechContGrid(ScanBase):
@scan_hook
def pre_scan(self):
"""
Calibrate the position capture while DDG1 / the RIO still ignore the line trigger
(the calibration fires the PSO output once), then enable the trigger receiver.
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()
if self.rio is not None:
self.rio.start_burst_mode()
else:
self.ddg1.set_trigger(DDG_EXT_RISING_EDGE)
self.ddg1.set_trigger(DDG_EXT_RISING_EDGE)
self.actions.pre_scan_all_devices()
@scan_hook
@@ -268,13 +240,11 @@ class AerotechContGrid(ScanBase):
self.motors, [self._line["line_start"], self.positions[line_index][1]]
)
status_motor.wait()
mcs_acquiring_status = None
if self.ddg1 is not None:
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]
)
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"]]),
@@ -295,8 +265,6 @@ class AerotechContGrid(ScanBase):
"""
if self.capture is not None:
self.capture.arm(self.frames_per_line)
if self.rio is not None:
self.rio.arm_burst(self.frames_per_line, self._line["rio_period_ms"])
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:
@@ -311,18 +279,6 @@ class AerotechContGrid(ScanBase):
self.fast_axis.motor_velocity.set(self._line["original_velocity"]).wait(timeout=5)
self.fast_axis.disarm_line_trigger()
if self.rio is not None:
self.rio.wait_burst_done(
self.frames_per_line,
timeout=3 + self.frames_per_line * self._line["rio_period_ms"] / 1000,
)
else:
self._wait_mcs(status_mcs)
if self.capture is not None:
self.capture.read_line()
def _wait_mcs(self, status_mcs):
try:
status_mcs.wait(timeout=3)
except TimeoutError as exc:
@@ -331,6 +287,9 @@ class AerotechContGrid(ScanBase):
" 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."""
@@ -377,16 +336,7 @@ class AerotechContGrid(ScanBase):
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() if self.ddg1 is not None else 0.0
)
if self.rio is not None:
period_ms = self.exp_time * 1000
if not np.isclose(period_ms, round(period_ms)):
raise ScanAbortion(
f"With the RIO trigger box exp_time must be a multiple of 1 ms, got {self.exp_time}."
)
line["rio_period_ms"] = int(round(period_ms))
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()
@@ -412,10 +362,7 @@ class AerotechContGrid(ScanBase):
def _restore(self):
try:
if self.rio is not None:
self.rio.stop_burst_mode()
else:
self.ddg1.set_trigger(DDG_SINGLE_SHOT)
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"])
tests/tests_devices/test_galil_rio_ddg.py
-142
View File
@@ -1,142 +0,0 @@
import time
import pytest
from bec_server.device_server.tests.utils import DMMock
from ophyd_devices.tests.utils import SocketMock
from csaxs_bec.devices.aerotech.a3200 import A3200Controller
from csaxs_bec.devices.omny.galil.galil_rio import GalilRIOController
from csaxs_bec.devices.omny.galil.galil_rio_ddg import GalilRIOBurst, GalilRIOBurstError
from csaxs_bec.devices.sim.sim_a3200 import SimA3200Motor
from csaxs_bec.devices.sim.sim_rio_ddg import SimGalilRIOBurst
from csaxs_bec.devices.sim.sim_socket import SimStateRegistry
@pytest.fixture(autouse=True)
def _reset():
GalilRIOController._reset_controller()
A3200Controller._reset_controller()
SimStateRegistry.reset()
yield
SimStateRegistry.reset()
GalilRIOController._reset_controller()
A3200Controller._reset_controller()
@pytest.fixture
def rio():
device = GalilRIOBurst(
name="rioddg", host="rio", device_manager=DMMock(), socket_cls=SocketMock
)
device.wait_for_connection()
device.controller.sock.flush_buffer()
return device
def test_arm_burst_sets_variables_in_order(rio):
rio.controller.sock.buffer_recv = [b" 0.0000\r\n", b":", b":", b":", b":", b":"]
rio.arm_burst(10, 50)
assert rio.controller.sock.buffer_put == [
b"MG busy\r",
b"nfr=10\r",
b"per=50\r",
b"wid=25\r",
b"thr=3.0\r",
b"armed=1\r",
]
@pytest.mark.parametrize(
"n_pulses, period, width, match",
[(5, 10, None, "below the minimum"), (5, 40, 40, "width"), (0, 40, None, "at least 1")],
)
def test_arm_burst_rejects_bad_parameters(rio, n_pulses, period, width, match):
with pytest.raises(GalilRIOBurstError, match=match):
rio.arm_burst(n_pulses, period, width)
def test_arm_burst_refused_while_busy(rio):
rio.controller.sock.buffer_recv = [b" 1.0000\r\n"]
with pytest.raises(GalilRIOBurstError, match="still sending"):
rio.arm_burst(5, 40)
def test_burst_mode_switching(rio):
rio.controller.sock.buffer_recv = [b":", b":", b":", b":"]
rio.start_burst_mode()
rio.stop_burst_mode()
assert rio.controller.sock.buffer_put == [b"HX0\r", b"XQ#BURST,0\r", b"HX0\r", b"XQ#DDG,0\r"]
def test_wait_burst_done_checks_pulse_count(rio):
rio.controller.sock.buffer_recv = [b" 0.0000 0.0000 4.0000 1.0000\r\n"]
with pytest.raises(GalilRIOBurstError, match="Sent 4 pulses instead of 5"):
rio.wait_burst_done(5, timeout=1)
def test_wait_burst_done_times_out_without_trigger(rio):
rio.controller.sock.buffer_recv = b" 1.0000 0.0000 0.0000 0.0000\r\n"
with pytest.raises(GalilRIOBurstError, match="line trigger"):
rio.wait_burst_done(5, timeout=0.1)
def test_sim_burst_triggered_by_aerotech_pso():
dm = DMMock()
x = SimA3200Motor(
"X",
name="aerox",
host="a3200",
port=8000,
device_manager=dm,
velocity=10,
encoder_channel=0,
)
rio = SimGalilRIOBurst(
name="rioddg",
host="rio",
device_manager=dm,
sim_trigger_host="a3200",
sim_trigger_port=8000,
)
x.wait_for_connection()
rio.wait_for_connection()
rio.start_burst_mode()
x.move(-0.5, wait=True)
rio.arm_burst(5, 20)
x.arm_line_trigger(0.0, 1.0)
start = time.time()
status = x.move(2.0, wait=False)
result = rio.wait_burst_done(5, timeout=5)
status.wait(timeout=5)
# trigger after 0.5 mm at 10 mm/s, then 5 x 20 ms: cannot be done earlier
assert 0.14 <= time.time() - start < 2.0
assert result == {"armed": False, "busy": False, "nsent": 5, "nline": 1}
def test_sim_burst_not_armed_ignores_trigger():
dm = DMMock()
x = SimA3200Motor(
"X",
name="aerox",
host="a3200",
port=8000,
device_manager=dm,
velocity=10,
encoder_channel=0,
)
rio = SimGalilRIOBurst(
name="rioddg",
host="rio",
device_manager=dm,
sim_trigger_host="a3200",
sim_trigger_port=8000,
)
x.wait_for_connection()
rio.wait_for_connection()
rio.start_burst_mode()
x.move(-0.5, wait=True)
x.arm_line_trigger(0.0, 1.0)
x.move(2.0, wait=True)
assert rio.burst_status() == {"armed": False, "busy": False, "nsent": 0, "nline": 0}
tests/tests_scans/test_aerotech_cont_grid_scan.py
-116
View File
@@ -183,119 +183,3 @@ def test_on_exception_restores_ddg_and_pso(v4_scan_assembler, device_manager):
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)
# --- trigger_source="rio" (RIO burst box instead of DDG1/DDG2 + MCS) ----------
def _add_rio(device_manager):
rio = MockCustomDevice(
"rioddg",
device_info={"signals": {"readback": _signal("rioddg", "hinted")}},
signal_read_values={"rioddg": 0.0},
)
for method in ("start_burst_mode", "stop_burst_mode", "arm_burst", "wait_burst_done"):
setattr(rio, method, mock.MagicMock())
device_manager.add_device(rio, replace=True)
def _assemble_rio(v4_scan_assembler, device_manager, exp_time=0.1):
_add_devices(device_manager)
_add_rio(device_manager)
return v4_scan_assembler(
"aerotech_cont_grid",
"aerox",
-1.0,
1.0,
0.5,
"aeroy",
-2.0,
2.0,
2,
exp_time=exp_time,
trigger_source="rio",
)
@pytest.mark.parametrize(("hook_name", "hook_tests"), AEROTECH_CONT_GRID_HOOK_TESTS)
def test_aerotech_cont_grid_rio_default_hooks(
v4_scan_assembler, device_manager, nth_done_status_mock, hook_name, hook_tests
):
scan = _assemble_rio(v4_scan_assembler, device_manager)
run_scan_tests(scan, [(hook_name, hook_tests)], nth_done_status_mock=nth_done_status_mock)
def test_rio_geometry_has_no_shutter_delay(v4_scan_assembler, device_manager):
scan = _assemble_rio(v4_scan_assembler, device_manager)
scan.prepare_scan()
assert scan.ddg1 is None and scan.mcs is None
assert scan._line["shutter_open_delay"] == 0.0
assert scan._line["trigger_position"] == pytest.approx(-1.0)
assert scan._line["rio_period_ms"] == 100
def test_rio_exp_time_must_be_whole_ms(v4_scan_assembler, device_manager):
scan = _assemble_rio(v4_scan_assembler, device_manager, exp_time=0.0255)
with pytest.raises(ScanAbortion, match="multiple of 1 ms"):
scan.prepare_scan()
def test_unknown_trigger_source_aborts(v4_scan_assembler, device_manager):
_add_devices(device_manager)
with pytest.raises(ScanAbortion, match="trigger_source"):
v4_scan_assembler(
"aerotech_cont_grid",
"aerox",
-1.0,
1.0,
0.5,
"aeroy",
-2.0,
2.0,
2,
exp_time=0.1,
trigger_source="laser",
)
def test_rio_pre_scan_calibrates_before_burst_mode(
v4_scan_assembler, device_manager, nth_done_status_mock
):
scan = _assemble_rio(v4_scan_assembler, device_manager)
scan.prepare_scan()
order = mock.MagicMock()
scan.capture = order.capture
scan.rio = order.rio
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.rio.start_burst_mode()]
def test_rio_at_each_point_arms_and_waits_for_burst(
v4_scan_assembler, device_manager, nth_done_status_mock
):
scan = _assemble_rio(v4_scan_assembler, device_manager)
scan.prepare_scan()
scan.capture = mock.MagicMock()
scan.rio = 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=None)
scan.rio.arm_burst.assert_called_once_with(5, 100)
scan.rio.wait_burst_done.assert_called_once_with(5, timeout=3 + 5 * 0.1)
scan.capture.read_line.assert_called_once_with()
def test_rio_on_exception_stops_burst_mode(v4_scan_assembler, device_manager):
scan = _assemble_rio(v4_scan_assembler, device_manager)
scan.prepare_scan()
scan.rio = mock.MagicMock()
scan.fast_axis = mock.MagicMock()
scan.on_exception(RuntimeError("test"))
scan.rio.stop_burst_mode.assert_called_once_with()