bec/pxiii_bec
0
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Aerotech scans work nicely

Browse Source
This commit is contained in:
Unknown MX Person
2024-09-20 14:38:04 +02:00
parent 14ca9bd74a
commit b0703552f2
5 changed files with 77 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pxiii_bec/device_configs/x06da_device_config.yaml
View File

@@ -345,7 +345,7 @@ omega:
abr:
description: Aerotech ABR motion system
deviceClass: pxiii_bec.devices.AerotechAbrStage
deviceConfig: {prefix: X06DA-ES}
deviceConfig: {prefix: 'X06DA-ES'}
onFailure: buffer
enabled: true
readoutPriority: monitored

22
pxiii_bec/devices/A3200.py
View File

@@ -81,7 +81,7 @@ Examples
import time
import math
from ophyd import Component, Device, EpicsSignal, EpicsSignalRO, Kind
from ophyd import Component, Device, EpicsSignal, EpicsSignalRO, Kind, SignalRO
from ophyd.status import MoveStatus, SubscriptionStatus, DeviceStatus
@@ -185,6 +185,12 @@ class AerotechAbrStage(Device):
_var_9 = Component(EpicsSignal, "-PSO:VAR-9", put_complete=True, kind=Kind.omitted)
_var_10 = Component(EpicsSignal, "-PSO:VAR-10", put_complete=True, kind=Kind.omitted)
# Task status PVs (programs always run on task 1)
task1 = Component(EpicsSignalRO, "-AERO:TSK1-DONE", auto_monitor=True, kind=Kind.hinted)
task2 = Component(EpicsSignalRO, "-AERO:TSK2-DONE", auto_monitor=True)
task3 = Component(EpicsSignalRO, "-AERO:TSK3-DONE", auto_monitor=True)
task4 = Component(EpicsSignalRO, "-AERO:TSK4-DONE", auto_monitor=True)
# A few PVs still needed from grid
raster_scan_done = Component(EpicsSignal, "-GRD:SCAN-DONE", kind=Kind.config)
raster_num_rows = Component(EpicsSignal, "-GRD:ROW-DONE", kind=Kind.config)
@@ -255,6 +261,20 @@ class AerotechAbrStage(Device):
"""
return self.raster.complete()
def complete(self, timeout=None) -> SubscriptionStatus:
""" Waits for task execution
NOTE: Original complete was raster scanner complete...
"""
# Define wait until the busy flag goes down (excluding initial update)
def isIdle(*args, old_value, value, timestamp, **kwargs):
return bool(value==1)
# Subscribe and wait for update
status = SubscriptionStatus(self.task1, isIdle, timeout=timeout, settle_time=0.5)
return status
def reset(self, settle_time=0.1):
self.scan_command.set(CMD_NONE, settle_time=settle_time).wait()
self.set_axis_mode("direct", settle_time=settle_time)

29
pxiii_bec/devices/A3200utils.py
View File

@@ -25,10 +25,29 @@ MEASURING_MODE = 1
class A3200Axis(PVPositioner):
"""Positioner wrapper for motors on the Aerotech A3200 controller. As the
IOC does not provide a motor record, this class simply wraps axes into
a standard Ophyd positioner for the BEC. It also has some additional
functionality for diagnostics.
""" Positioner wrapper for A3200 axes
Positioner wrapper for motors on the Aerotech A3200 controller. As the IOC
does not provide a motor record, this class simply wraps axes into a
standard Ophyd positioner for the BEC. It also has some additional
functionality for error checking and diagnostics.
Examples
--------
omega = A3200Axis('X06DA-ES-DF1:OMEGA', base_pv='X06DA-ES')
class abr(Device):
omega = Component(A3200Axis, '-DF1:OMEGA')
gmx = Component(A3200Axis, '-DF1:GMX')
gmy = Component(A3200Axis, '-DF1:GMY')
Parameters
----------
prefix : str
Axis PV name root.
base_pv : str (situational)
IOC PV name root, i.e. X06DA-ES if standalone class.
"""
USER_ACCESS = ['omove']
@@ -74,7 +93,7 @@ class A3200Axis(PVPositioner):
""" Patching the parent's PVs into the axis class to check for direct/locked mode"""
if parent is None:
def maybe_add_prefix(self, instance, kw, suffix):
""" Patched not to enforce parent prefix"""
""" Patched not to enforce parent prefix when no parent"""
if kw in self.add_prefix:
return suffix
return suffix

2
pxiii_bec/scans/__init__.py
View File

@@ -1 +1 @@
from .mx_measurements import MeasureFastOmega
from .mx_measurements import MeasureFastOmega, MeasureStandardWedge, MeasureVerticalLine, MeasureScanSlits, MeasureRasterSimple, MeasureScreening, MeasureRepeatedOscillation, MeasureMSOX, MeasureGrid

52
pxiii_bec/scans/mx_measurements.py
View File

@@ -45,12 +45,19 @@ class AbrCmd:
class AerotechFlyscanBase(AsyncFlyScanBase):
""" Base class for MX flyscans
Low-level base class for standard scans at the PX beamlines at SLS. Theese scans use the
A3200 rotation stage and the actual experiment is performed using an AeroBasic script
controlled via global variables.
Low-level base class for standard scans at the PX beamlines at SLS. Theese
scans use the A3200 rotation stage and the actual experiment is performed
using an AeroBasic script controlled via global variables. The base class
has some basic safety features like checking status then sets globals and
fires off the scan. Implementations can choose to set the corresponding
configurations in child classes or pass it as command line parameters.
IMPORTANT: The AeroBasic scripts take care of the PSO configuration.
Parameters:
-----------
abr_complete : bool
Wait for the launched ABR task to complete.
"""
scan_report_hint = "table"
arg_input = {}
@@ -59,7 +66,6 @@ class AerotechFlyscanBase(AsyncFlyScanBase):
# Aerotech stage config
abr_command = None
abr_globals = {}
abr_reset = False
abr_raster_reset =False
abr_complete = False
abr_timeout = None
@@ -76,8 +82,6 @@ class AerotechFlyscanBase(AsyncFlyScanBase):
self.abr_command = self.caller_kwargs.get("abr_command")
if "abr_globals" in self.caller_kwargs:
self.abr_globals = self.caller_kwargs.get("abr_globals")
if "abr_reset" in self.caller_kwargs:
self.abr_reset = self.caller_kwargs.get("abr_reset")
if "abr_raster_reset" in self.caller_kwargs:
self.abr_raster_reset = self.caller_kwargs.get("abr_raster_reset")
if "abr_complete" in self.caller_kwargs:
@@ -86,17 +90,19 @@ class AerotechFlyscanBase(AsyncFlyScanBase):
self.abr_timeout = self.caller_kwargs.get("abr_timeout")
def pre_scan(self):
# ToDo: Move roughly to start position
# TODO: Move roughly to start position???
# ABR status checking
stat = yield from self.stubs.send_rpc_and_wait("abr","status.get")
if stat not in (0, "OK"):
raise RuntimeError("Aerotech ABR seems to be in error state {stat}, please reset")
task = yield from self.stubs.send_rpc_and_wait("abr","task1.get")
# From what I got values are copied to local vars at the start of scan,
# so only kickoff should be forbidden.
if task not in (1, "OK"):
raise RuntimeError("Aerotech ABR task #1 seems to busy")
if self.abr_reset:
yield from self.stubs.send_rpc_and_wait("abr", "reset")
# Reset the scan engine
if self.abr_reset:
yield from self.stubs.send_rpc_and_wait("abr", "reset")
# Reset the raster scan engine
if self.abr_raster_reset:
yield from self.stubs.command("abr", "raster_scan_done.set", 0)
@@ -123,9 +129,13 @@ class AerotechFlyscanBase(AsyncFlyScanBase):
logger.info("Measurement launched on the ABR stage...")
# Wait for grid scanner to finish
if self.abr_complete and self.abr_timeout is not None:
st = yield from self.stubs.send_rpc_and_wait("abr", "complete", self.abr_timeout)
st.wait()
if self.abr_complete:
if self.abr_timeout is not None:
st = yield from self.stubs.send_rpc_and_wait("abr", "complete", self.abr_timeout)
st.wait()
else:
st = yield from self.stubs.send_rpc_and_wait("abr", "complete")
st.wait()
def cleanup(self):
"""Set scan progress to 1 to finish the scan"""
@@ -175,6 +185,7 @@ class MeasureStandardWedge(AerotechFlyscanBase):
"var_3" : self.scan_move_time,
"var_4" : self.scan_ready_rate,
}
logger.info(self.abr_globals)
# Call base class
super().__init__(parameter=parameter, **kwargs)
@@ -204,7 +215,7 @@ class MeasureVerticalLine(AerotechFlyscanBase):
exp_time : float
Eeffective exposure time per step [s]
"""
scan_name = "measure_vline"
scan_name = "vertical_line"
required_kwargs = ["exp_time", "range_y", "steps_y"]
def __init__(self, *args, parameter: dict = None, **kwargs):
@@ -515,6 +526,7 @@ class MeasureFastOmega(AerotechFlyscanBase):
"var_3" : self.scan_move_time,
"var_4" : self.scan_acceleration,
}
logger.info(self.abr_globals)
# Call base class
super().__init__(parameter=parameter, **kwargs)
@@ -750,7 +762,6 @@ class MeasureMSOX(AerotechFlyscanBase):
self.scan_stepnum = parameter['kwargs'].get("steps", 1)
self.scan_settle_time = parameter['kwargs'].get("settle_time", 0)
self.abr_command = AbrCmd.MSOX
self.abr_reset = True
self.abr_globals = {
"var_1" : self.scan_start,
"var_2" : self.scan_range,
@@ -814,7 +825,6 @@ class MeasureGrid(AerotechFlyscanBase):
self.axis = ['abr.omega', 'abr.gmx', 'abr.gmy']
# Set parameters from meaningful inputs
self.abr_command = AbrCmd.RASTER_SCAN
self.abr_reset = True
self.scan_positions = parameter['kwargs'].get("positions")
self.scan_stepnum_x = parameter['kwargs'].get("steps_x")
self.scan_stepnum_y = parameter['kwargs'].get("steps_y")
@@ -913,7 +923,6 @@ class MeasureGridStill(AerotechFlyscanBase):
self.axis = ['abr.gmx', 'abr.gmy', 'abr.omega']
# Set parameters from meaningful inputs
self.abr_command = AbrCmd.RASTER_SCAN_STILL
self.abr_reset = True
self.scan_positions = parameter['kwargs'].get("positions")
self.scan_stepnum_x = parameter['kwargs'].get("steps_x")
self.scan_stepnum_y = parameter['kwargs'].get("steps_y")
@@ -1005,7 +1014,6 @@ class MeasureJungfrau(AerotechFlyscanBase):
self.scan_sleep = parameter['kwargs'].get("sleep", 0.1)
self.abr_timeout = None if self.scan_sleep <= 0 else 5.0 + (self.scan_num_rows * self.scan_sleep) + (self.scan_num_cols * self.scan_num_rows * self.scan_exp_time)
self.abr_command = AbrCmd.JUNGFRAU
self.abr_reset = True
self.abr_complete = True
self.abr_globals = {
"var_1" : self.scan_stepnum_x,