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

A3200 cleanup

Browse Source
This commit is contained in:
Unknown MX Person
2024-12-13 17:44:52 +01:00
parent 78c75b1769
commit add46d8b0d
4 changed files with 202 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

144
pxiii_bec/devices/A3200.py
View File

@@ -80,56 +80,20 @@ Examples
"""
import time
from ophyd import Component, Device, EpicsSignal, EpicsSignalRO, Kind
from ophyd.status import MoveStatus, SubscriptionStatus, DeviceStatus
from ophyd.status import SubscriptionStatus
from .A3200utils import A3200Axis, A3200RasterScanner, A3200Oscillator
from .A3200utils import A3200RasterScanner, A3200Oscillator
from .A3200enums import *
try:
from bec_lib import bec_logger
logger = bec_logger.logger
except ModuleNotFoundError:
import logging
logger = logging.getLogger("A3200")
ABR_DONE = 0
ABR_READY = 1
ABR_BUSY = 2
# pylint: disable=logging-fstring-interpolation
GRID_SCAN_BUSY = 0
GRID_SCAN_DONE = 1
DIRECT_MODE = 0
MEASURING_MODE = 1
SHUTTER_CLOSE = 0
SHUTTER_OPEN = 1
FULL_PERIOD = 0
HALF_PERIOD = 1
CMD_NONE = 0
CMD_RASTER_SCAN_SIMPLE = 1
CMD_MEASURE_STANDARD = 2
CMD_VERTICAL_LINE_SCAN = 3
CMD_SCREENING = 4
CMD_SUPER_FAST_OMEGA = 5
CMD_STILL_WEDGE = 6
CMD_STILLS = 7
CMD_REPEAT_SINGLE_OSCILLATION = 8
CMD_SINGLE_OSCILLATION = 9
CMD_OLD_FASHIONED = 10
CMD_RASTER_SCAN = 11
CMD_JET_ROTATION = 12
CMD_X_HELICAL = 13
CMD_X_RUNSEQ = 14
CMD_JUNGFRAU = 15
CMD_MSOX = 16
CMD_SLIT_SCAN = 17
CMD_RASTER_SCAN_STILL = 18
CMD_SCAN_SASTT = 19
CMD_SCAN_SASTT_V2 = 20
CMD_SCAN_SASTT_V3 = 21
AXIS_OMEGA = 1
AXIS_GMX = 2
AXIS_GMY = 3
AXIS_GMZ = 4
AXIS_STY = 5
AXIS_STZ = 6
class AerotechAbrStage(Device):
@@ -165,11 +129,13 @@ class AerotechAbrStage(Device):
raster = Component(A3200RasterScanner, "", kind=Kind.config)
osc = Component(A3200Oscillator, "", kind=Kind.config)
# Positioners for the motor axes
omega = Component(A3200Axis, "-DF1:OMEGA", kind=Kind.hinted)
gmx = Component(A3200Axis, "-DF1:GMX", kind=Kind.hinted)
gmy = Component(A3200Axis, "-DF1:GMY", kind=Kind.hinted)
gmz = Component(A3200Axis, "-DF1:GMZ", kind=Kind.hinted)
# Status flags for all axes
omega_done = Component(EpicsSignalRO, "-DF1:OMEGA-DONE", kind=Kind.normal)
gmx_done = Component(EpicsSignalRO, "-DF1:GMX-DONE", kind=Kind.normal)
gmy_done = Component(EpicsSignalRO, "-DF1:GMY-DONE", kind=Kind.normal)
gmz_done = Component(EpicsSignalRO, "-DF1:GMZ-DONE", kind=Kind.normal)
scan_command = Component(
EpicsSignal, "-PSO:CMD", put_complete=True, kind=Kind.omitted)
@@ -270,19 +236,13 @@ class AerotechAbrStage(Device):
new = self.read_configuration()
return old, new
def complete(self, timeout=None) -> DeviceStatus:
""" ToDo: WTF was this device doing here? Whatever...
"""
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 is_idle(*args, old_value, value, timestamp, **kwargs):
def is_idle(*args, value, _, **kwargs):
return bool(value==1)
# Subscribe and wait for update
@@ -290,46 +250,14 @@ class AerotechAbrStage(Device):
return status
def reset(self, settle_time=0.1):
"""Attempts to reset the currently running measurement task on the A3200"""
self.scan_command.set(CMD_NONE, settle_time=settle_time).wait()
self.set_axis_mode("direct", settle_time=settle_time)
def is_ioc_ok(self):
"""Checks execution status"""
return 0 == self.status.get()
def is_ready_and_willing(self):
return self.is_ioc_ok() and self.omega.is_homed()
def is_omega_ok(self):
return 0 == self.self.omega.status.get()
def is_homed(self):
"""Are the motors homed?"""
return 1 == self.omega.is_homed.get()
def do_homing(self, wait=True):
"""Execute the homing procedure.
Executes the homing procedure and waits (default) until it is completed.
TODO: Return a status object to do this wwith futures and monitoring.
PARAMETERS
`wait` true / false if the routine is to wait for the homing to finish.
"""
self.omega.home.set(1, settle_time=1).wait()
if not wait:
return
while not self.omega.is_homed():
time.sleep(0.2)
@property
def velocity(self):
return self.omega.velocity.get()
@velocity.setter
def velocity(self, value):
self.omega.velocity.set(value).wait()
@property
def exp_time(self):
return self.osc.exp_time.get()
@@ -396,12 +324,11 @@ class AerotechAbrStage(Device):
self.osc.get_ready.set(1, settle_time=0.1).wait()
if wait:
for n in range(10):
for _ in range(10):
try:
self.wait_status(ABR_READY, timeout=5).wait()
except RuntimeWarning as e:
print(str(e), end=" ")
print(" --- trying ready again.")
except RuntimeWarning as ex:
logger.warning(f"{ex} --- trying ready again.")
self.osc.get_ready.set(1).wait()
@@ -418,8 +345,8 @@ class AerotechAbrStage(Device):
time.sleep(0.1)
try:
moving = self.is_moving()
except:
print("Aerotech() Failed to retrieve moving state for axes omega, gmx, gmy, gmz")
except Exception as ex:
logger.error(f"Failed to retrieve moving state for axes: {ex}")
moving = True
timeisup = timeout < time.time()
@@ -427,9 +354,10 @@ class AerotechAbrStage(Device):
raise RuntimeWarning("timeout waiting for all axis to stop moving")
def is_moving(self):
return (
self.omega.moving
and self.gmx.moving and self.gmy.moving and self.gmz.moving )
"""Chechs if all axes are DONE"""
return not (
self.omega_done.get()
and self.gmx_done.get() and self.gmy_done.get() and self.gmz_done.get() )
def stop(self, wait_after_reload=1.0):
"""Stops current motions; reloads aerobasic programs; goes DIRECT
@@ -449,21 +377,13 @@ class AerotechAbrStage(Device):
"""Starts the previously configured exposure."""
self.wait_for_movements()
self.osc.taskStart.set(1).wait()
for n in range(10):
for _ in range(10):
try:
self.osc.wait_status(ABR_BUSY, timeout=1)
except RuntimeWarning as e:
print("%s --- trying start again.", str(e))
except RuntimeWarning as ex:
logger.error(f"{ex} --- trying start again.")
self.osc.kickoff()
def move(self, position, wait=False, velocity=None, relative=None, direct=False,
**kwargs) -> MoveStatus:
""" Move the omega axis
"""
return self.omega.move(position, wait=wait, velocity=velocity, relative=relative,
direct=direct, **kwargs)
if __name__ == "__main__":
abr = AerotechAbrStage(prefix="X06DA-ES", name="abr")

113
pxiii_bec/devices/A3200enums.py Normal file
View File

@@ -0,0 +1,113 @@
# -*- coding: utf-8 -*-
"""
Enumerations for the MX specific Aerotech A3200 stage.
@author: mohacsi_i
"""
# pylint: disable=too-few-public-methods
class AbrStatus:
"""ABR measurement task status"""
DONE = 0
READY = 1
BUSY = 2
ABR_DONE = 0
ABR_READY = 1
ABR_BUSY = 2
class AbrGridStatus:
"""ABR grid scan status"""
BUSY = 0
DONE = 1
GRID_SCAN_BUSY = 0
GRID_SCAN_DONE = 1
class AbrMode:
"""ABR mode status"""
DIRECT = 0
MEASURING = 1
DIRECT_MODE = 0
MEASURING_MODE = 1
class AbrShutterStatus:
"""ABR shutter status"""
CLOSE = 0
OPEN = 1
SHUTTER_CLOSE = 0
SHUTTER_OPEN = 1
class AbrGridPeriod:
"""ABR grid period"""
FULL = 0
HALF = 1
FULL_PERIOD = 0
HALF_PERIOD = 1
class AbrCmd:
"""ABR command table"""
NONE = 0
RASTER_SCAN_SIMPLE = 1
MEASURE_STANDARD = 2
VERTICAL_LINE_SCAN = 3
SCREENING = 4
SUPER_FAST_OMEGA = 5
STILL_WEDGE = 6
STILLS = 7
REPEAT_SINGLE_OSCILLATION = 8
SINGLE_OSCILLATION = 9
OLD_FASHIONED = 10
RASTER_SCAN = 11
JET_ROTATION = 12
X_HELICAL = 13
X_RUNSEQ = 14
JUNGFRAU = 15
MSOX = 16
SLIT_SCAN = 17
RASTER_SCAN_STILL = 18
SCAN_SASTT = 19
SCAN_SASTT_V2 = 20
SCAN_SASTT_V3 = 21
CMD_NONE = 0
CMD_RASTER_SCAN_SIMPLE = 1
CMD_MEASURE_STANDARD = 2
CMD_VERTICAL_LINE_SCAN = 3
CMD_SCREENING = 4
CMD_SUPER_FAST_OMEGA = 5
CMD_STILL_WEDGE = 6
CMD_STILLS = 7
CMD_REPEAT_SINGLE_OSCILLATION = 8
CMD_SINGLE_OSCILLATION = 9
CMD_OLD_FASHIONED = 10
CMD_RASTER_SCAN = 11
CMD_JET_ROTATION = 12
CMD_X_HELICAL = 13
CMD_X_RUNSEQ = 14
CMD_JUNGFRAU = 15
CMD_MSOX = 16
CMD_SLIT_SCAN = 17
CMD_RASTER_SCAN_STILL = 18
CMD_SCAN_SASTT = 19
CMD_SCAN_SASTT_V2 = 20
CMD_SCAN_SASTT_V3 = 21
class AbrAxis:
"""ABR axis index"""
OMEGA = 1
GMX = 2
GMY = 3
GMZ = 4
STY = 5
STZ = 6
AXIS_OMEGA = 1
AXIS_GMX = 2
AXIS_GMY = 3
AXIS_GMZ = 4
AXIS_STY = 5
AXIS_STZ = 6

65
pxiii_bec/devices/A3200utils.py
View File

@@ -5,6 +5,7 @@ Created on Tue Jun 11 11:28:38 2024
@author: mohacsi_i
"""
import types
import math
from ophyd import (Component, PVPositioner, Device, Signal, EpicsSignal, EpicsSignalRO, Kind,
PositionerBase)
from ophyd.status import Status, MoveStatus, SubscriptionStatus, DeviceStatus
@@ -14,6 +15,11 @@ from bec_lib import bec_logger
logger = bec_logger.logger
from .A3200enums import *
ABR_DONE = 0
ABR_READY = 1
ABR_BUSY = 2
@@ -118,8 +124,8 @@ class A3200Axis(PVPositioner):
self.vmin.set(vmin).wait()
self.vmax.set(vmax).wait()
def move(self, position, velocity=None, wait=True, relative=False, direct=False, timeout=None,
**kwargs) -> MoveStatus:
def omove(self, position, wait=True, timeout=None, moved_cb=None, velocity=None, relative=False,
direct=False, **kwargs) -> MoveStatus:
""" Native absolute/relative movement on the A3200 """
# Check if we're in direct movement mode
@@ -164,10 +170,9 @@ class A3200Axis(PVPositioner):
return status
def omove(self, position, velocity=None, wait=True, relative=False, direct=False,
**kwargs) -> MoveStatus:
def move(self, position, wait=True, timeout=None, moved_cb=None, **kwargs) -> MoveStatus:
""" Exposes the ophyd move command through BEC abstraction"""
return self.move(position, velocity, wait, relative, direct, **kwargs)
return self.omove(position, wait=wait, timeout=timeout, moved_cb=moved_cb, **kwargs)
def rock(self, distance, counts: int, velocity=None, wait=True) -> Status:
""" Repeated single axis zigzag scan I guess PSO should be configured for this"""
@@ -186,6 +191,38 @@ class A3200Axis(PVPositioner):
return status
# def is_omega_ok(self):
# """Checks omega axis status"""
# return 0 == self.self.omega.status.get()
# def is_homed(self):
# """Checks if omega is homed"""
# return 1 == self.omega.is_homed.get()
# def do_homing(self, wait=True):
# """Execute the homing procedure.
# Executes the homing procedure on omega and waits (default) until it is completed.
# TODO: Return a status object to do this wwith futures and monitoring.
# PARAMETERS
# `wait` true / false if the routine is to wait for the homing to finish.
# """
# self.omega.home.set(1, settle_time=1).wait()
# if not wait:
# return
# while not self.omega.is_homed():
# time.sleep(0.2)
@@ -305,7 +342,7 @@ class A3200RasterScanner(Device):
try:
# Define wait until the busy flag goes down (excluding initial update)
timestamp_ = 0
def isReady(*args, old_value, value, timestamp, **kwargs):
def is_ready(*args, old_value, value, timestamp, **kwargs):
nonlocal timestamp_
result = False if (timestamp_== 0) else bool(value==ABR_READY)
print(f"Old {old_value}\tNew: {value}\tResult: {result}")
@@ -313,7 +350,7 @@ class A3200RasterScanner(Device):
return result
# Subscribe and wait for update
status = SubscriptionStatus(self.grid_done, isReady, timeout=2, settle_time=0.5)
status = SubscriptionStatus(self.grid_done, is_ready, timeout=2, settle_time=0.5)
status.wait()
except TimeoutError as e:
print(str(e), end=" ")
@@ -333,7 +370,7 @@ class A3200RasterScanner(Device):
def wait_scan_done(self, timeout=60.0) -> SubscriptionStatus:
# Define wait until the busy flag goes down (excluding initial update)
timestamp_ = 0
def isReady(*args, old_value, value, timestamp, **kwargs):
def is_ready(*args, old_value, value, timestamp, **kwargs):
nonlocal timestamp_
result = False if (timestamp_== 0) else bool(value==1)
print(f"Old {old_value}\tNew: {value}\tResult: {result}")
@@ -341,7 +378,7 @@ class A3200RasterScanner(Device):
return result
# Subscribe and wait for update
status = SubscriptionStatus(self.scan_done, isReady, timeout=timeout, settle_time=0.5)
status = SubscriptionStatus(self.scan_done, is_ready, timeout=timeout, settle_time=0.5)
status.wait()
return status
@@ -373,7 +410,7 @@ class A3200RasterScanner(Device):
if timeout is not None:
# Define wait until the busy flag goes down (excluding initial update)
timestamp_ = 0
def isReady(*args, old_value, value, timestamp, **kwargs):
def is_ready(*args, old_value, value, timestamp, **kwargs):
nonlocal timestamp_
result = False if (timestamp_== 0) else bool(value==GRID_SCAN_DONE)
print(f"Old {old_value}\tNew: {value}\tResult: {result}")
@@ -381,7 +418,7 @@ class A3200RasterScanner(Device):
return result
# Subscribe and wait for update
status = SubscriptionStatus(self.scan_done, isReady, timeout=timeout, settle_time=0.5)
status = SubscriptionStatus(self.scan_done, is_ready, timeout=timeout, settle_time=0.5)
return status
status = DeviceStatus(self, settle_time=0.5)
status.set_finished()
@@ -424,13 +461,13 @@ class A3200Oscillator(Device):
SubscriptionStatus : A waitable status subscribed on 'phase' updates.
"""
# Define wait until the busy flag goes down (excluding initial update)
def inStatus(*args, old_value, value, timestamp, **kwargs):
def in_status(*args, old_value, value, _, **kwargs):
result = bool(value==target)
print(f"Old {old_value}\tNew: {value}\tResult: {result}")
# print(f"Old {old_value}\tNew: {value}\tResult: {result}")
return result
# Subscribe and wait for update
status = SubscriptionStatus(self.phase, inStatus, timeout=timeout, settle_time=0.5)
status = SubscriptionStatus(self.phase, in_status, timeout=timeout, settle_time=0.5)
return status

6
pxiii_bec/devices/__init__.py
View File

@@ -1,2 +1,8 @@
# -*- coding: utf-8 -*-
"""
Ophyd devices for the PX III beamline, including the MX specific Aerotech A3200 stage.
@author: mohacsi_i
"""
from .A3200 import AerotechAbrStage
from .A3200utils import A3200Axis