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