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

Mono scan works

Browse Source
This commit is contained in:
gac-x06da
2025-01-28 15:45:10 +01:00
parent 4b76d1b191
commit 22c46f8f8e
8 changed files with 439 additions and 540 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pxiii_bec/device_configs/x06da_device_config.yaml
+18 -18
View File
@@ -79,24 +79,24 @@ dccm_xbpm:
readoutPriority: monitored readoutPriority: monitored
readOnly: true readOnly: true
softwareTrigger: false softwareTrigger: false
# dccm_energy: dccm_energy:
# description: Monochromator energy using ECMC virtual motors description: Monochromator energy using ECMC virtual motors
# deviceClass: ophyd.EpicsMotor deviceClass: ophyd.EpicsMotor
# deviceConfig: {prefix: 'X06DA-OP-DCCM:_ENERGY'} deviceConfig: {prefix: 'X06DA-OP-DCCM:_ENERGY'}
# onFailure: buffer onFailure: buffer
# enabled: true enabled: true
# readoutPriority: monitored readoutPriority: monitored
# readOnly: false readOnly: false
# softwareTrigger: false softwareTrigger: false
# dccm_eoffset: dccm_offset:
# description: Monochromator energy offset for ECMC virtual motors description: Monochromator energy offset for ECMC virtual motors
# deviceClass: ophyd.EpicsMotor deviceClass: ophyd.EpicsMotor
# deviceConfig: {prefix: 'X06DA-OP-DCCM:_EOFFSET'} deviceConfig: {prefix: 'X06DA-OP-DCCM:_OFFSET'}
# onFailure: buffer onFailure: buffer
# enabled: true enabled: true
# readoutPriority: monitored readoutPriority: monitored
# readOnly: false readOnly: false
# softwareTrigger: false softwareTrigger: false
ssxbpm_trx: ssxbpm_trx:
description: XBPM motion before secondary source description: XBPM motion before secondary source
deviceClass: ophyd.EpicsMotor deviceClass: ophyd.EpicsMotor
pxiii_bec/devices/A3200.py
+7 -49
View File
@@ -82,10 +82,7 @@ Examples
import time import time
from ophyd import Component, EpicsSignal, EpicsSignalRO, Kind from ophyd import Component, EpicsSignal, EpicsSignalRO, Kind
from ophyd.status import SubscriptionStatus from ophyd.status import SubscriptionStatus
from ophyd_devices.interfaces.base_classes.psi_detector_base import PSIDetectorBase as PsiDeviceBase from ophyd_devices.interfaces.base_classes.bec_device_base import BECDeviceBase, CustomPrepare
from ophyd_devices.interfaces.base_classes.psi_detector_base import (
CustomDetectorMixin as CustomDeviceMixin,
)
try: try:
from .A3200enums import AbrCmd, AbrMode from .A3200enums import AbrCmd, AbrMode
@@ -93,18 +90,14 @@ except ImportError:
from A3200enums import AbrCmd, AbrMode from A3200enums import AbrCmd, AbrMode
try: from bec_lib import bec_logger
from bec_lib import bec_logger
logger = bec_logger.logger logger = bec_logger.logger
except ModuleNotFoundError:
import logging
logger = logging.getLogger("A3200")
# pylint: disable=logging-fstring-interpolation # pylint: disable=logging-fstring-interpolation
class AerotechAbrMixin(CustomDeviceMixin): class AerotechAbrMixin(CustomPrepare):
"""Configuration class for the Aerotech A3200 controller for the ABR stage""" """Configuration class for the Aerotech A3200 controller for the ABR stage"""
def on_stage(self): def on_stage(self):
@@ -202,7 +195,7 @@ class AerotechAbrMixin(CustomDeviceMixin):
self.parent.blueunstage() self.parent.blueunstage()
class AerotechAbrStage(PsiDeviceBase): class AerotechAbrStage(BECDeviceBase):
"""Standard PX stage on A3200 controller """Standard PX stage on A3200 controller
This is the wrapper class for the standard rotation stage layout for the PX This is the wrapper class for the standard rotation stage layout for the PX
@@ -214,7 +207,7 @@ class AerotechAbrStage(PsiDeviceBase):
""" """
custom_prepare_cls = AerotechAbrMixin custom_prepare_cls = AerotechAbrMixin
USER_ACCESS = ["reset", "kickoff", "complete"] USER_ACCESS = ["reset", "kickoff", "complete", "set_axis_mode"]
taskStop = Component(EpicsSignal, "-AERO:TSK-STOP", put_complete=True, kind=Kind.omitted) taskStop = Component(EpicsSignal, "-AERO:TSK-STOP", put_complete=True, kind=Kind.omitted)
status = Component(EpicsSignal, "-AERO:STAT", put_complete=True, kind=Kind.omitted) status = Component(EpicsSignal, "-AERO:STAT", put_complete=True, kind=Kind.omitted)
@@ -345,7 +338,6 @@ class AerotechAbrStage(PsiDeviceBase):
Since configuration synchronization is not guaranteed, this does Since configuration synchronization is not guaranteed, this does
nothing. The script launched by kickoff(). nothing. The script launched by kickoff().
""" """
pass
def bluekickoff(self, timeout=1) -> SubscriptionStatus: def bluekickoff(self, timeout=1) -> SubscriptionStatus:
"""Kick off the set program""" """Kick off the set program"""
@@ -405,6 +397,7 @@ class AerotechAbrStage(PsiDeviceBase):
# Go to direct mode # Go to direct mode
self.set_axis_mode("direct", settle_time=settle_time) self.set_axis_mode("direct", settle_time=settle_time)
# pylint: disable=arguments-differ
def stop(self, settle_time=1.0) -> None: def stop(self, settle_time=1.0) -> None:
"""Stops current motions""" """Stops current motions"""
# Disarm commands # Disarm commands
@@ -417,30 +410,6 @@ class AerotechAbrStage(PsiDeviceBase):
"""Checks execution status""" """Checks execution status"""
return 0 == self.status.get() return 0 == self.status.get()
# @property
# def exp_time(self):
# return self.osc.exp_time.get()
# @exp_time.setter
# def exp_time(self, value):
# self.osc.etime.set(value).wait()
# @property
# def start_angle(self):
# return self.osc.ostart_pos.get()
# @start_angle.setter
# def start_angle(self, value):
# self.osc.ostart_pos(value).wait()
# @property
# def measurement_state(self):
# return self.osc.phase.get()
# @measurement_state.setter
# def measurement_state(self, value):
# self.osc.phase.set(value).wait()
@property @property
def axis_mode(self): def axis_mode(self):
return self.axisAxesMode.get() return self.axisAxesMode.get()
@@ -485,17 +454,6 @@ class AerotechAbrStage(PsiDeviceBase):
and self.gmz_done.get() and self.gmz_done.get()
) )
# def start_exposure(self):
# """Starts the previously configured exposure."""
# self.wait_for_movements()
# self.osc.taskStart.set(1).wait()
# for _ in range(10):
# try:
# self.osc.wait_status(ABR_BUSY, timeout=1)
# except RuntimeWarning as ex:
# logger.error(f"{ex} --- trying start again.")
# self.osc.kickoff()
if __name__ == "__main__": if __name__ == "__main__":
abr = AerotechAbrStage(prefix="X06DA-ES", name="abr") abr = AerotechAbrStage(prefix="X06DA-ES", name="abr")
pxiii_bec/devices/A3200enums.py
-60
View File
@@ -15,11 +15,6 @@ class AbrStatus:
BUSY = 2 BUSY = 2
ABR_DONE = 0
ABR_READY = 1
ABR_BUSY = 2
class AbrGridStatus: class AbrGridStatus:
"""ABR grid scan status""" """ABR grid scan status"""
@@ -27,10 +22,6 @@ class AbrGridStatus:
DONE = 1 DONE = 1
GRID_SCAN_BUSY = 0
GRID_SCAN_DONE = 1
class AbrMode: class AbrMode:
"""ABR mode status""" """ABR mode status"""
@@ -38,10 +29,6 @@ class AbrMode:
MEASURING = 1 MEASURING = 1
DIRECT_MODE = 0
MEASURING_MODE = 1
class AbrShutterStatus: class AbrShutterStatus:
"""ABR shutter status""" """ABR shutter status"""
@@ -49,21 +36,6 @@ class AbrShutterStatus:
OPEN = 1 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: class AbrCmd:
"""ABR command table""" """ABR command table"""
@@ -91,30 +63,6 @@ class AbrCmd:
SCAN_SASTT_V3 = 21 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: class AbrAxis:
"""ABR axis index""" """ABR axis index"""
@@ -124,11 +72,3 @@ class AbrAxis:
GMZ = 4 GMZ = 4
STY = 5 STY = 5
STZ = 6 STZ = 6
AXIS_OMEGA = 1
AXIS_GMX = 2
AXIS_GMY = 3
AXIS_GMZ = 4
AXIS_STY = 5
AXIS_STZ = 6
pxiii_bec/devices/A3200utils.py
+1 -6
View File
@@ -39,11 +39,6 @@ class A3200Axis(PVPositioner):
-------- --------
omega = A3200Axis('X06DA-ES-DF1:OMEGA', base_pv='X06DA-ES') 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 Parameters
---------- ----------
prefix : str prefix : str
@@ -246,5 +241,5 @@ class A3200Axis(PVPositioner):
# Automatically start an axis if directly invoked # Automatically start an axis if directly invoked
if __name__ == "__main__": if __name__ == "__main__":
omega = A3200Axis(prefix="X06DA-ES-DF1:OMEGA", name="omega") omega = A3200Axis(prefix="X06DA-ES-DF1:OMEGA", base_pv='X06DA-ES', name="omega")
omega.wait_for_connection() omega.wait_for_connection()
pxiii_bec/devices/NDArrayPreview.py
+1 -3
View File
@@ -66,10 +66,8 @@ class NDArrayPreview(Device):
if array_size[-1] == 0: if array_size[-1] == 0:
array_size = array_size[:-1] array_size = array_size[:-1]
pixel_count = np.prod(array_size)
image = self.array_data.get() image = self.array_data.get()
if image.size == pixel_count: return np.array(image).reshape(array_size)
return np.array(image).reshape(array_size)
pxiii_bec/devices/SmarGon.py
+17 -13
View File
@@ -1,7 +1,7 @@
import time import time
import requests
from threading import Thread from threading import Thread
from ophyd import Component, Device, Kind, Signal, SignalRO, PVPositioner import requests
from ophyd import Component, Kind, Signal, PVPositioner
from ophyd.status import SubscriptionStatus from ophyd.status import SubscriptionStatus
try: try:
@@ -38,6 +38,7 @@ class SmarGonSignal(Signal):
timestamp = time.time() timestamp = time.time()
# Perform the actual write to SmargoPolo # Perform the actual write to SmargoPolo
#pylint: disable=protected-access
r = self.parent._go_n_put(f"{self.write_addr}?{self.addr.upper()}={value}") r = self.parent._go_n_put(f"{self.write_addr}?{self.addr.upper()}={value}")
old_value = self._readback old_value = self._readback
@@ -66,6 +67,7 @@ class SmarGonSignal(Signal):
raise ValueError(f"Target {value} outside of limits {self.limits}") raise ValueError(f"Target {value} outside of limits {self.limits}")
def get(self, *args, **kwargs): def get(self, *args, **kwargs):
#pylint: disable=protected-access
r = self.parent._go_n_get(self.write_addr) r = self.parent._go_n_get(self.write_addr)
# print(r) # print(r)
if isinstance(r, dict): if isinstance(r, dict):
@@ -92,6 +94,7 @@ class SmarGonSignalRO(Signal):
self._mon.start() self._mon.start()
def get(self, *args, **kwargs): def get(self, *args, **kwargs):
#pylint: disable=protected-access
r = self.parent._go_n_get(self.read_addr) r = self.parent._go_n_get(self.read_addr)
if isinstance(r, dict): if isinstance(r, dict):
@@ -100,14 +103,16 @@ class SmarGonSignalRO(Signal):
self.put(r, force=True) self.put(r, force=True)
return self._readback return self._readback
def poll(self):
def poll(self, *args, **kwargs):
""" Fooo""" """ Fooo"""
time.sleep(2) time.sleep(2)
while True: while True:
time.sleep(0.2) time.sleep(0.2)
self.get() try:
self.get()
except requests.ConnectTimeout as ex:
logger.error(f"[{self.name}] {ex}")
class SmarGonAxis(PVPositioner): class SmarGonAxis(PVPositioner):
"""SmarGon client deice """SmarGon client deice
@@ -167,13 +172,16 @@ class SmarGonAxis(PVPositioner):
print(r) print(r)
def move(self, position, wait=True, timeout=None, moved_cb=None): def move(self, position, wait=True, timeout=None, moved_cb=None):
""" Move command that's masked by BEC"""
return self.omove(position, wait, timeout, moved_cb)
def omove(self, position, wait=True, timeout=None, moved_cb=None):
""" Original move command without the BEC wrappers"""
status = self.setpoint.set(position, settle_time=0.1) status = self.setpoint.set(position, settle_time=0.1)
if not wait: if not wait:
return status return status
else:
status.wait() status.wait()
def on_target(*, value, **_): def on_target(*, value, **_):
distance = abs(value-position) distance = abs(value-position)
@@ -184,10 +192,6 @@ class SmarGonAxis(PVPositioner):
) )
return status return status
def omove(self, position, wait=True, timeout=None, moved_cb=None):
""" Original move command without the BEC wrappers"""
return self.move(position, wait, timeout, moved_cb)
def _pos_changed(self, timestamp=None, value=None, **kwargs): def _pos_changed(self, timestamp=None, value=None, **kwargs):
pass pass
pxiii_bec/devices/SmarGon_orig.py
+390 -390
View File
@@ -1,393 +1,393 @@
#!/usr/bin/env python3 # #!/usr/bin/env python3
from time import sleep, time # from time import sleep, time
from typing import Tuple # from typing import Tuple
from requests import get, put # from requests import get, put
from beamline import beamline # from beamline import beamline
from mx_redis import SMARGON # from mx_redis import SMARGON
try: # try:
from mx_preferences import get_config # from mx_preferences import get_config
host = get_config(beamline)["smargon"]["host"] # host = get_config(beamline)["smargon"]["host"]
port = get_config(beamline)["smargon"]["port"] # port = get_config(beamline)["smargon"]["port"]
except Exception: # except Exception:
host = "x06da-smargopolo.psi.ch" # host = "x06da-smargopolo.psi.ch"
port = 3000 # port = 3000
base = f"http://{host}:{port}" # base = f"http://{host}:{port}"
def gonget(thing: str, **kwargs) -> dict: # def gonget(thing: str, **kwargs) -> dict:
"""issue a GET for some API component on the smargopolo server""" # """issue a GET for some API component on the smargopolo server"""
cmd = f"{base}/{thing}" # cmd = f"{base}/{thing}"
if kwargs.get("verbose", False): # if kwargs.get("verbose", False):
print(cmd) # print(cmd)
r = get(cmd) # r = get(cmd)
if not r.ok: # if not r.ok:
raise Exception(f"error getting {thing}; server returned {r.status_code} => {r.reason}") # raise Exception(f"error getting {thing}; server returned {r.status_code} => {r.reason}")
return r.json() # return r.json()
def gonput(thing: str, **kwargs): # def gonput(thing: str, **kwargs):
"""issue a PUT for some API component on the smargopolo server""" # """issue a PUT for some API component on the smargopolo server"""
cmd = f"{base}/{thing}" # cmd = f"{base}/{thing}"
if kwargs.get("verbose", False): # if kwargs.get("verbose", False):
print(cmd) # print(cmd)
put(cmd) # put(cmd)
def scsput(**kwargs): # def scsput(**kwargs):
""" # """
Issue a new absolute target in the SH coordinate system. # Issue a new absolute target in the SH coordinate system.
The key "verbose" may be passed in kwargs with any true # The key "verbose" may be passed in kwargs with any true
value for verbose behaviour. # value for verbose behaviour.
:param kwargs: a dict containing keys ("shx", "shy", "shz", "chi", "phi") # :param kwargs: a dict containing keys ("shx", "shy", "shz", "chi", "phi")
:type kwargs: dict # :type kwargs: dict
:return: # :return:
:rtype: # :rtype:
""" # """
xyz = { # xyz = {
k.upper(): v for k, v in kwargs.items() if k.lower() in ("shx", "shy", "shz", "chi", "phi") # k.upper(): v for k, v in kwargs.items() if k.lower() in ("shx", "shy", "shz", "chi", "phi")
} # }
thing = "&".join([f"{k.upper()}={float(v):.5f}" for k, v in xyz.items()]) # thing = "&".join([f"{k.upper()}={float(v):.5f}" for k, v in xyz.items()])
cmd = f"{base}/targetSCS?{thing}" # cmd = f"{base}/targetSCS?{thing}"
if kwargs.get("verbose", False): # if kwargs.get("verbose", False):
print(cmd) # print(cmd)
put(cmd) # put(cmd)
def bcsput(**kwargs): # def bcsput(**kwargs):
""" # """
Issue a new absolute target in the beamline coordinate system. # Issue a new absolute target in the beamline coordinate system.
The key "verbose" may be passed in kwargs with any true # The key "verbose" may be passed in kwargs with any true
value for verbose behaviour. # value for verbose behaviour.
:param kwargs: a dict containing keys ("bx", "by", "bz", "chi", "phi") # :param kwargs: a dict containing keys ("bx", "by", "bz", "chi", "phi")
:return: # :return:
:rtype: # :rtype:
""" # """
xyz = {k.upper(): v for k, v in kwargs.items() if k.lower() in ("bx", "by", "bz", "chi", "phi")} # xyz = {k.upper(): v for k, v in kwargs.items() if k.lower() in ("bx", "by", "bz", "chi", "phi")}
thing = "&".join([f"{k.upper()}={float(v):.5f}" for k, v in xyz.items()]) # thing = "&".join([f"{k.upper()}={float(v):.5f}" for k, v in xyz.items()])
cmd = f"{base}/targetBCS?{thing}" # cmd = f"{base}/targetBCS?{thing}"
if kwargs.get("verbose", False): # if kwargs.get("verbose", False):
print(cmd) # print(cmd)
put(cmd) # put(cmd)
def scsrelput(**kwargs) -> None: # def scsrelput(**kwargs) -> None:
""" # """
Issue relative increments to current SH coordinate system. # Issue relative increments to current SH coordinate system.
The key "verbose" may be passed in kwargs with any true # The key "verbose" may be passed in kwargs with any true
value for verbose behaviour. # value for verbose behaviour.
:param kwargs: a dict containing keys ("shx", "shy", "shz", "chi", "phi") # :param kwargs: a dict containing keys ("shx", "shy", "shz", "chi", "phi")
:type kwargs: dict # :type kwargs: dict
:return: # :return:
:rtype: # :rtype:
""" # """
xyz = { # xyz = {
k.upper(): v for k, v in kwargs.items() if k.lower() in ("shx", "shy", "shz", "chi", "phi") # k.upper(): v for k, v in kwargs.items() if k.lower() in ("shx", "shy", "shz", "chi", "phi")
} # }
thing = "&".join([f"{k.upper()}={float(v):.5f}" for k, v in xyz.items()]) # thing = "&".join([f"{k.upper()}={float(v):.5f}" for k, v in xyz.items()])
cmd = f"{base}/targetSCS_rel?{thing}" # cmd = f"{base}/targetSCS_rel?{thing}"
if kwargs.get("verbose", False): # if kwargs.get("verbose", False):
print(cmd) # print(cmd)
put(cmd) # put(cmd)
def bcsrelput(**kwargs): # def bcsrelput(**kwargs):
""" # """
Issue relative increments to current beamline coordinate system. # Issue relative increments to current beamline coordinate system.
The key "verbose" may be passed in kwargs with any true # The key "verbose" may be passed in kwargs with any true
value for verbose behaviour. # value for verbose behaviour.
:param kwargs: a dict containing keys ("bx", "by", "bz") # :param kwargs: a dict containing keys ("bx", "by", "bz")
:type kwargs: dict # :type kwargs: dict
:return: # :return:
:rtype: # :rtype:
""" # """
xyz = {k.upper(): v for k, v in kwargs.items() if k.lower() in ("bx", "by", "bz")} # xyz = {k.upper(): v for k, v in kwargs.items() if k.lower() in ("bx", "by", "bz")}
thing = "&".join([f"{k.upper()}={float(v):.5f}" for k, v in xyz.items()]) # thing = "&".join([f"{k.upper()}={float(v):.5f}" for k, v in xyz.items()])
cmd = f"{base}/targetBCS_rel?{thing}" # cmd = f"{base}/targetBCS_rel?{thing}"
if kwargs.get("verbose", False): # if kwargs.get("verbose", False):
print(cmd) # print(cmd)
put(cmd) # put(cmd)
# url_redis = f"{beamline}-cons-705.psi.ch" # # url_redis = f"{beamline}-cons-705.psi.ch"
# print(f"connecting to redis DB #3 on host: {url_redis}") # # print(f"connecting to redis DB #3 on host: {url_redis}")
# redis_handle = redis.StrictRedis(host=url_redis, db=3) # # redis_handle = redis.StrictRedis(host=url_redis, db=3)
# pubsub = redis_handle.pubsub() # # pubsub = redis_handle.pubsub()
MODE_UNINITIALIZED = 0 # MODE_UNINITIALIZED = 0
MODE_INITIALIZING = 1 # MODE_INITIALIZING = 1
MODE_READY = 2 # MODE_READY = 2
MODE_ERROR = 99 # MODE_ERROR = 99
class SmarGon(object): # class SmarGon(object):
def __init__(self): # def __init__(self):
super(SmarGon, self).__init__() # super(SmarGon, self).__init__()
self.__dict__.update(target=None) # self.__dict__.update(target=None)
self.__dict__.update(bookmarks={}) # self.__dict__.update(bookmarks={})
self.__dict__.update(_latest_message={}) # self.__dict__.update(_latest_message={})
# pubsub.psubscribe(**{f"__keyspace@{SMARGON.value}__:*": self._cb_readbackSCS}) # # pubsub.psubscribe(**{f"__keyspace@{SMARGON.value}__:*": self._cb_readbackSCS})
# pubsub.run_in_thread(sleep_time=0.5, daemon=True) # # pubsub.run_in_thread(sleep_time=0.5, daemon=True)
def __repr__(self): # def __repr__(self):
BX, BY, BZ, OMEGA, CHI, PHI, a, b, c = self.readback_bcs().values() # BX, BY, BZ, OMEGA, CHI, PHI, a, b, c = self.readback_bcs().values()
return f"<{self.__class__.__name__} X={BX:.3f}, Y={BY:.3f}, Z={BZ:.3f}, CHI={CHI:.3f}, PHI={PHI:.3f}, OMEGA={OMEGA:.3f}>" # return f"<{self.__class__.__name__} X={BX:.3f}, Y={BY:.3f}, Z={BZ:.3f}, CHI={CHI:.3f}, PHI={PHI:.3f}, OMEGA={OMEGA:.3f}>"
def _cb_readbackSCS(self, msg): # def _cb_readbackSCS(self, msg):
if msg["data"] in ["hset"]: # if msg["data"] in ["hset"]:
self._latest_message = msg # self._latest_message = msg
def move_home(self, wait=False) -> None: # def move_home(self, wait=False) -> None:
"""move to beamline coordinate system X, Y, Z, Chi, Phi = 0 0 0 0 0""" # """move to beamline coordinate system X, Y, Z, Chi, Phi = 0 0 0 0 0"""
self.apply_bookmark_sh({"shx": 0.0, "shy": 0.0, "shz": 18.0, "chi": 0.0, "phi": 0.0}) # self.apply_bookmark_sh({"shx": 0.0, "shy": 0.0, "shz": 18.0, "chi": 0.0, "phi": 0.0})
if wait: # if wait:
self.wait_home() # self.wait_home()
def xyz(self, coords: Tuple[float, float, float], wait: bool = True) -> None: # def xyz(self, coords: Tuple[float, float, float], wait: bool = True) -> None:
""" # """
Move smargon in absolute beamline coordinates # Move smargon in absolute beamline coordinates
:param coords: a tuple of floats representing X, Y, Z coordinates # :param coords: a tuple of floats representing X, Y, Z coordinates
:type coords: # :type coords:
:param wait: # :param wait:
:type wait: # :type wait:
:return: # :return:
:rtype: # :rtype:
""" # """
x, y, z = coords # x, y, z = coords
# the two steps below are necessary otherwise the control system # # the two steps below are necessary otherwise the control system
# remembers *a* previous CHI # # remembers *a* previous CHI
bcs = self.bcs # bcs = self.bcs
bcs.update({"BX": x, "BY": y, "BZ": z}) # bcs.update({"BX": x, "BY": y, "BZ": z})
self.bcs = bcs # self.bcs = bcs
if wait: # if wait:
self.wait() # self.wait()
def wait_home(self, timeout: float = 20.0) -> None: # def wait_home(self, timeout: float = 20.0) -> None:
""" # """
wait for the smargon to reach its home position: # wait for the smargon to reach its home position:
SHX = 0.0 # SHX = 0.0
SHY = 0.0 # SHY = 0.0
SHZ = 18.0 # SHZ = 18.0
CHI = 0.0 # CHI = 0.0
PHI = 0.0 # PHI = 0.0
:param timeout: time to wait for positions to be reached raises TimeoutError if timeout reached # :param timeout: time to wait for positions to be reached raises TimeoutError if timeout reached
:type timeout: float # :type timeout: float
:return: # :return:
:rtype: # :rtype:
""" # """
tout = timeout + time() # tout = timeout + time()
in_place = [False, False] # in_place = [False, False]
rbv = -999.0 # rbv = -999.0
while not all(in_place) and time() < tout: # while not all(in_place) and time() < tout:
rbv = self.readback_scs() # rbv = self.readback_scs()
in_place = [] # in_place = []
for k, v in {"SHX": 0.0, "SHY": 0.0, "SHZ": 18.0, "CHI": 0.0, "PHI": 0.0}.items(): # for k, v in {"SHX": 0.0, "SHY": 0.0, "SHZ": 18.0, "CHI": 0.0, "PHI": 0.0}.items():
in_place.append(abs(rbv[k] - v) < 0.01) # in_place.append(abs(rbv[k] - v) < 0.01)
if time() > tout: # if time() > tout:
raise TimeoutError(f"timeout waiting for smargon to reach home position: {rbv}") # raise TimeoutError(f"timeout waiting for smargon to reach home position: {rbv}")
def push_bookmark(self): # def push_bookmark(self):
""" # """
save current absolute coordinates in FIFO stack # save current absolute coordinates in FIFO stack
:return: # :return:
:rtype: # :rtype:
""" # """
t = round(time()) # t = round(time())
self.bookmarks[t] = self.readback_scs() # self.bookmarks[t] = self.readback_scs()
def pop_bookmark(self): # def pop_bookmark(self):
return self.bookmarks.popitem()[1] # return self.bookmarks.popitem()[1]
def apply_bookmark_sh(self, scs): # def apply_bookmark_sh(self, scs):
scsput(**scs) # scsput(**scs)
def apply_last_bookmark_sh(self): # def apply_last_bookmark_sh(self):
scs = self.pop_bookmark() # scs = self.pop_bookmark()
scsput(**scs) # scsput(**scs)
def readback_mcs(self): # def readback_mcs(self):
"""current motor positions of the smargon sliders""" # """current motor positions of the smargon sliders"""
return gonget("readbackMCS") # return gonget("readbackMCS")
def readback_scs(self): # def readback_scs(self):
"""current SH coordinates of the smargon model""" # """current SH coordinates of the smargon model"""
return gonget("readbackSCS") # return gonget("readbackSCS")
def readback_bcs(self): # def readback_bcs(self):
"""current beamline coordinates of the smargon""" # """current beamline coordinates of the smargon"""
return gonget("readbackBCS") # return gonget("readbackBCS")
def target_scs(self): # def target_scs(self):
"""currently assigned targets for the smargon control system""" # """currently assigned targets for the smargon control system"""
return gonget("targetSCS") # return gonget("targetSCS")
def initialize(self): # def initialize(self):
"""initialize the smargon""" # """initialize the smargon"""
self.set_mode(MODE_UNINITIALIZED) # self.set_mode(MODE_UNINITIALIZED)
sleep(0.1) # sleep(0.1)
self.set_mode(MODE_INITIALIZING) # self.set_mode(MODE_INITIALIZING)
def set_mode(self, mode: int): # def set_mode(self, mode: int):
"""put smargon control system in a given mode # """put smargon control system in a given mode
MODE_UNINITIALIZED = 0 # MODE_UNINITIALIZED = 0
MODE_INITIALIZING = 1 # MODE_INITIALIZING = 1
MODE_READY = 2 # MODE_READY = 2
MODE_ERROR = 99 # MODE_ERROR = 99
""" # """
gonput(f"mode?mode={mode}") # gonput(f"mode?mode={mode}")
def enable_correction(self): # def enable_correction(self):
"""enable calibration based corrections""" # """enable calibration based corrections"""
gonput("corr_type?corr_type=1") # gonput("corr_type?corr_type=1")
def disable_correction(self): # def disable_correction(self):
"""disable calibration based corrections""" # """disable calibration based corrections"""
gonput("corr_type?corr_type=0") # gonput("corr_type?corr_type=0")
def chi(self, val=None, wait=False): # def chi(self, val=None, wait=False):
if val is None: # if val is None:
return self.readback_scs()["CHI"] # return self.readback_scs()["CHI"]
scsput(CHI=val) # scsput(CHI=val)
if wait: # if wait:
timeout = 10 + time() # timeout = 10 + time()
while time() < timeout: # while time() < timeout:
if abs(val - self.readback_scs()["CHI"]) < 0.1: # if abs(val - self.readback_scs()["CHI"]) < 0.1:
break # break
if time() > timeout: # if time() > timeout:
raise RuntimeError(f"SmarGon CHI did not reach requested target {val} in time") # raise RuntimeError(f"SmarGon CHI did not reach requested target {val} in time")
def phi(self, val=None, wait=False): # def phi(self, val=None, wait=False):
if val is None: # if val is None:
return self.readback_scs()["PHI"] # return self.readback_scs()["PHI"]
scsput(PHI=val) # scsput(PHI=val)
if wait: # if wait:
timeout = 70 + time() # timeout = 70 + time()
while time() < timeout: # while time() < timeout:
if abs(val - self.readback_scs()["PHI"]) < 0.1: # if abs(val - self.readback_scs()["PHI"]) < 0.1:
break # break
if time() > timeout: # if time() > timeout:
raise RuntimeError(f"SmarGon PHI did not reach requested target {val} in time") # raise RuntimeError(f"SmarGon PHI did not reach requested target {val} in time")
def wait(self, timeout=60.0): # def wait(self, timeout=60.0):
"""waits up to `timeout` seconds for smargon to reach target""" # """waits up to `timeout` seconds for smargon to reach target"""
target = { # target = {
k.upper(): v # k.upper(): v
for k, v in self.target_scs().items() # for k, v in self.target_scs().items()
if k.lower() in ("shx", "shy", "shz", "chi", "phi") # if k.lower() in ("shx", "shy", "shz", "chi", "phi")
} # }
timeout = timeout + time() # timeout = timeout + time()
while time() < timeout: # while time() < timeout:
s = { # s = {
k: (abs(v - target[k]) < 0.01) # k: (abs(v - target[k]) < 0.01)
for k, v in self.readback_scs().items() # for k, v in self.readback_scs().items()
if k.upper() in ("SHX", "SHY", "SHZ", "CHI", "PHI") # if k.upper() in ("SHX", "SHY", "SHZ", "CHI", "PHI")
} # }
if all(list(s.values())): # if all(list(s.values())):
break # break
if time() > timeout: # if time() > timeout:
raise TimeoutError("timed out waiting for smargon to reach target") # raise TimeoutError("timed out waiting for smargon to reach target")
def __setattr__(self, key, value): # def __setattr__(self, key, value):
key = key.lower() # key = key.lower()
if key == "mode": # if key == "mode":
self.set_mode(value) # self.set_mode(value)
elif key == "correction": # elif key == "correction":
assert value in ( # assert value in (
0, # 0,
1, # 1,
False, # False,
True, # True,
), "correction is either 1 or True (enabled) or 0 (disabled)" # ), "correction is either 1 or True (enabled) or 0 (disabled)"
gonput(f"corr_type?corr_type?{value}") # gonput(f"corr_type?corr_type?{value}")
elif key == "scs": # elif key == "scs":
scsput(**value) # scsput(**value)
elif key == "bcs": # elif key == "bcs":
bcsput(**value) # bcsput(**value)
elif key == "target": # elif key == "target":
if not isinstance(value, dict): # if not isinstance(value, dict):
raise Exception( # raise Exception(
f"expected a dict with target axis and values got something else: {value}" # f"expected a dict with target axis and values got something else: {value}"
) # )
for k in value.keys(): # for k in value.keys():
if k.lower() not in "shx shy shz chi phi ox oy oz".split(): # if k.lower() not in "shx shy shz chi phi ox oy oz".split():
raise Exception(f'unknown axis in target "{k}"') # raise Exception(f'unknown axis in target "{k}"')
scsput(**value) # scsput(**value)
elif key in "shx shy shz chi phi ox oy oz".split(): # elif key in "shx shy shz chi phi ox oy oz".split():
scsput(**{key: value}) # scsput(**{key: value})
elif key in "bx by bz".split(): # elif key in "bx by bz".split():
bcs = self.readback_bcs() # bcs = self.readback_bcs()
bcs[key] = value # bcs[key] = value
bcsput(**bcs) # bcsput(**bcs)
else: # else:
self.__dict__[key].update(value) # self.__dict__[key].update(value)
def __getattr__(self, key): # def __getattr__(self, key):
key = key.lower() # key = key.lower()
if key == "mode": # if key == "mode":
return self.readback_mcs()["mode"] # return self.readback_mcs()["mode"]
elif key == "correction": # elif key == "correction":
return gonget("corr_type") # return gonget("corr_type")
elif key == "bcs": # elif key == "bcs":
return self.readback_bcs() # return self.readback_bcs()
elif key == "mcs": # elif key == "mcs":
return self.readback_mcs() # return self.readback_mcs()
elif key == "scs": # elif key == "scs":
return self.readback_scs() # return self.readback_scs()
elif key in "shx shy shz chi phi ox oy oz".split(): # elif key in "shx shy shz chi phi ox oy oz".split():
return self.readback_scs()[key.upper()] # return self.readback_scs()[key.upper()]
elif key in "bx by bz".split(): # elif key in "bx by bz".split():
return self.readback_bcs()[key.upper()] # return self.readback_bcs()[key.upper()]
else: # else:
return self.__getattribute__(key) # return self.__getattribute__(key)
if __name__ == "__main__": # if __name__ == "__main__":
import argparse # import argparse
parser = argparse.ArgumentParser(description="SmarGon client") # parser = argparse.ArgumentParser(description="SmarGon client")
parser.add_argument("-i", "--initialize", help="initialize smargon", action="store_true") # parser.add_argument("-i", "--initialize", help="initialize smargon", action="store_true")
args = parser.parse_args() # args = parser.parse_args()
smargon = SmarGon() # smargon = SmarGon()
if args.initialize: # if args.initialize:
print("initializing smargon device") # print("initializing smargon device")
import Aerotech # import Aerotech
print("moving aerotech back by 50mm") # print("moving aerotech back by 50mm")
abr = Aerotech.Abr() # abr = Aerotech.Abr()
abr.incr_x(-50.0, wait=True, velo=100.0) # abr.incr_x(-50.0, wait=True, velo=100.0)
print("issuing init command to smargon") # print("issuing init command to smargon")
smargon.initialize() # smargon.initialize()
sleep(0.5) # sleep(0.5)
print("waiting for init routine to complete") # print("waiting for init routine to complete")
while MODE_READY != smargon.mode: # while MODE_READY != smargon.mode:
sleep(0.5) # sleep(0.5)
print("moving smargon to HOME position") # print("moving smargon to HOME position")
smargon.move_home() # smargon.move_home()
print("moving aerotech to its previous position") # print("moving aerotech to its previous position")
abr.incr_x(50.0, wait=True, velo=100.0) # abr.incr_x(50.0, wait=True, velo=100.0)
exit(0) # exit(0)
pxiii_bec/devices/StdDaqPreview.py
+5 -1
View File
@@ -50,7 +50,11 @@ class StdDaqPreviewMixin(CustomDetectorMixin):
# Might hang on recv_multipart # Might hang on recv_multipart
self._mon.join(timeout=1) self._mon.join(timeout=1)
# So also disconnect the socket # So also disconnect the socket
self.parent._socket.disconnect(self.parent.url.get()) try:
self.parent._socket.disconnect(self.parent.url.get())
except zmq.error.ZMQError:
# Might be already closed
pass
def on_stop(self): def on_stop(self):
"""Stop a running preview""" """Stop a running preview"""