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

samcam stream preview #19

Merged
mohacsi_i merged 12 commits from feature/samcam-preview into main 2025-01-30 12:26:48 +01:00
Conversation 10 Commits 12 Files Changed 14
+1077 -653
14 changed files with 1077 additions and 653 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pxiii_bec/device_configs/x06da_device_config.yaml
+58 -41
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
@@ -394,8 +394,8 @@ xbox_diode:
readoutPriority: monitored
readOnly: true
softwareTrigger: false
samdist:
description: Sample distance
gonpos:
description: Sample sensor distance
deviceClass: ophyd.EpicsSignalRO
deviceConfig: {read_pv: 'X06DA-ES-DF1:CBOX-USER1', auto_monitor: true}
onFailure: buffer
@@ -403,7 +403,7 @@ samdist:
readoutPriority: monitored
readOnly: true
softwareTrigger: false
samrange:
gonvalid:
description: Sample in valid distance
deviceClass: ophyd.EpicsSignalRO
deviceConfig: {read_pv: 'X06DA-ES-DF1:CBOX-CMP1', auto_monitor: true}
@@ -422,14 +422,36 @@ samzoom:
readOnly: false
softwareTrigger: false
samcam:
description: Sample camera device
deviceClass: ophyd_devices.devices.areadetector.cam.GenICam
deviceConfig: {prefix: 'X06DA-SAMCAM:cam1:'}
description: Sample camera aggregate device
deviceClass: pxiii_bec.devices.SamCamDetector
deviceConfig: {prefix: 'X06DA-SAMCAM:'}
onFailure: buffer
enabled: true
readoutPriority: monitored
readOnly: false
softwareTrigger: false
samstream:
description: Sample camera ZMQ stream
deviceClass: pxiii_bec.devices.StdDaqPreviewDetector
deviceConfig:
url: 'tcp://129.129.110.12:9089'
deviceTags:
- detector
enabled: true
readoutPriority: async
readOnly: false
softwareTrigger: false
samimg:
description: Sample camera image from EPICS
deviceClass: pxiii_bec.devices.NDArrayPreview
deviceConfig:
prefix: 'X06DA-SAMCAM:image1:'
deviceTags:
- detector
enabled: true
readoutPriority: async
readOnly: false
softwareTrigger: false
bstop_pneum:
@@ -544,6 +566,15 @@ gmy:
readoutPriority: monitored
readOnly: false
softwareTrigger: false
gmz:
description: ABR axial stage
deviceClass: pxiii_bec.devices.A3200Axis
deviceConfig: {prefix: 'X06DA-ES-DF1:GMZ', base_pv: 'X06DA-ES'}
onFailure: buffer
enabled: true
readoutPriority: monitored
readOnly: false
softwareTrigger: false
omega:
description: ABR rotation stage
deviceClass: pxiii_bec.devices.A3200Axis
@@ -565,7 +596,7 @@ abr:
shx:
description: SmarGon X axis
deviceClass: pxiii_bec.devices.SmarGonAxis
deviceConfig: {prefix: 'SCS', sg_url: 'http://x06da-smargopolo.psi.ch:3000'}
deviceConfig: {prefix: 'SCS', low_limit: -2, high_limit: 2, sg_url: 'http://x06da-smargopolo.psi.ch:3000'}
onFailure: buffer
enabled: true
readoutPriority: monitored
@@ -574,7 +605,7 @@ shx:
shy:
description: SmarGon Y axis
deviceClass: pxiii_bec.devices.SmarGonAxis
deviceConfig: {prefix: 'SCS', sg_url: 'http://x06da-smargopolo.psi.ch:3000'}
deviceConfig: {prefix: 'SCS', low_limit: -2, high_limit: 2, sg_url: 'http://x06da-smargopolo.psi.ch:3000'}
onFailure: buffer
enabled: true
readoutPriority: monitored
@@ -592,7 +623,7 @@ shz:
chi:
description: SmarGon CHI axis
deviceClass: pxiii_bec.devices.SmarGonAxis
deviceConfig: {prefix: 'SCS', sg_url: 'http://x06da-smargopolo.psi.ch:3000'}
deviceConfig: {prefix: 'SCS', low_limit: 0, high_limit: 40, sg_url: 'http://x06da-smargopolo.psi.ch:3000'}
onFailure: buffer
enabled: true
readoutPriority: monitored
@@ -607,17 +638,3 @@ phi:
readoutPriority: monitored
readOnly: false
softwareTrigger: false
samimg:
description: Sample camera image
deviceClass: ophyd_devices.devices.areadetector.plugins.ImagePlugin_V35
deviceConfig: {prefix: 'X06DA-SAMCAM:image1:'}
onFailure: buffer
enabled: false
readoutPriority: monitored
readOnly: true
softwareTrigger: false
pxiii_bec/devices/A3200.py
+32 -125
View File
@@ -28,36 +28,7 @@ Aerotech.wait_status(status)
Aerotech.move(angle, wait=False, speed=None)
Aerotech.set_shutter(state)
Attribute Access in Abr class
=============================
The Abr class overwrites the getattr and setattr methods to provide a pythonic
way of controlling the rootation stage of the Abr.
The following properties are implemented:
velocity
Sets the velocity of rotation: ``-ES-DF1:ROTX-SETV``
omega
Move the omega angle without any wait: ``-ES-DF1:ROTX-VAL``
exp_time
Sets the PV for the measurement's exposure time: ``-ES-OSC:ETIME``
start_angle
Sets the PV for the measurement's starting angle: ``-ES-OSC:START-POS``
oscillation_angle
Sets the PV for the measurement's oscillation angle: ``-ES-OSC:RANGE``
shutter
Controls the shutter: ``-ES-PH1:SET`` and ``-ES-PH1:GET``
measurement_state
Returns the PV for the measurement state: ``-ES-OSC:DONE``
axis_mode
Returns the axis mode: ``-ES-DF1:AXES-MODE``
Examples
@@ -82,10 +53,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 +61,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):
@@ -120,29 +84,22 @@ class AerotechAbrMixin(CustomDeviceMixin):
scanargs = self.parent.scaninfo.scan_msg.info["kwargs"]
scanname = self.parent.scaninfo.scan_msg.info["scan_name"]
if scanname in ("standardscan"):
scan_start = scanargs["start"]
scan_range = scanargs["range"]
scan_move_time = scanargs["move_time"]
scan_ready_rate = scanargs.get("ready_rate", 500)
if scanname in ("standardscan", "helicalscan"):
d["scan_command"] = AbrCmd.MEASURE_STANDARD
d["var_1"] = scan_start
d["var_2"] = scan_range
d["var_3"] = scan_move_time
d["var_4"] = scan_ready_rate
d["var_1"] = scanargs["start"]
d["var_2"] = scanargs["range"]
d["var_3"] = scanargs["move_time"]
d["var_4"] = scanargs.get("ready_rate", 500)
d["var_5"] = 0
d["var_6"] = 0
d["var_7"] = 0
d["var_8"] = 0
d["var_9"] = 0
if scanname in ("verticallinescan", "vlinescan"):
scan_exp_time = scanargs["exp_time"]
scan_range_y = scanargs["range"]
scan_steps_y = scanargs["steps"]
d["scan_command"] = AbrCmd.VERTICAL_LINE_SCAN
d["var_1"] = scan_range_y / scan_steps_y
d["var_2"] = scan_steps_y
d["var_3"] = scan_exp_time
d["var_1"] = scanargs["range"] / scanargs["steps"]
d["var_2"] = scanargs["steps"]
d["var_3"] = scanargs["exp_time"]
d["var_4"] = 0
d["var_5"] = 0
d["var_6"] = 0
@@ -150,37 +107,23 @@ class AerotechAbrMixin(CustomDeviceMixin):
d["var_8"] = 0
d["var_9"] = 0
if scanname in ("screeningscan"):
scan_start = scanargs["start"]
scan_range = scanargs["range"]
scan_stepnum_o = scanargs["steps"]
scan_exp_time = scanargs["exp_time"]
scan_oscrange = scanargs["oscrange"]
scan_delta = scanargs.get("delta", 0.5)
scan_stepsize_o = scan_range / scan_stepnum_o
d["scan_command"] = AbrCmd.SCREENING
d["var_1"] = scan_start
d["var_2"] = scan_oscrange
d["var_3"] = scan_exp_time
d["var_4"] = scan_stepsize_o
d["var_5"] = scan_stepnum_o
d["var_6"] = scan_delta
d["var_1"] = scanargs["start"]
d["var_2"] = scanargs["oscrange"]
d["var_3"] = scanargs["exp_time"]
d["var_4"] = scanargs["range"] / scanargs["steps"]
d["var_5"] = scanargs["steps"]
d["var_6"] = scanargs.get("delta", 0.5)
d["var_7"] = 0
d["var_8"] = 0
d["var_9"] = 0
if scanname in ("rasterscan", "rastersimplescan"):
scan_exp_time = scanargs["exp_time"]
scan_range_x = scanargs["range_x"]
scan_range_y = scanargs["range_y"]
scan_stepnum_x = scanargs["steps_x"]
scan_stepnum_y = scanargs["steps_y"]
scan_stepsize_x = scan_range_x / scan_stepnum_x
scan_stepsize_y = scan_range_y / scan_stepnum_y
d["scan_command"] = AbrCmd.RASTER_SCAN_SIMPLE
d["var_1"] = scan_exp_time
d["var_2"] = scan_stepsize_x
d["var_3"] = scan_stepsize_y
d["var_4"] = scan_stepnum_x
d["var_5"] = scan_stepnum_y
d["var_1"] = scanargs["exp_time"]
d["var_2"] = scanargs["range_x"] / scanargs["steps_x"]
d["var_3"] = scanargs["range_y"] / scanargs["steps_y"]
d["var_4"] = scanargs["steps_x"]
d["var_5"] = scanargs["steps_y"]
d["var_6"] = 0
d["var_7"] = 0
d["var_8"] = 0
@@ -202,7 +145,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 +157,7 @@ class AerotechAbrStage(PsiDeviceBase):
"""
custom_prepare_cls = AerotechAbrMixin
USER_ACCESS = ["reset", "kickoff", "complete"]
USER_ACCESS = ["reset", "kickoff", "bluekickoff", "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)
@@ -264,10 +207,7 @@ class AerotechAbrStage(PsiDeviceBase):
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)
scan_done = Component(EpicsSignal, "-GRD:SCAN-DONE", kind=Kind.config)
def set_axis_mode(self, mode: str, settle_time=0.1) -> None:
"""Set axis mode to direct/measurement mode.
@@ -345,7 +285,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"""
@@ -357,9 +296,10 @@ class AerotechAbrStage(PsiDeviceBase):
# Subscribe and wait for update
status = SubscriptionStatus(
self.raster_scan_done, is_busy, timeout=timeout, settle_time=0.1
self.scan_done, is_busy, timeout=timeout, settle_time=0.1
)
return status
status.wait()
# return status
def blueunstage(self, settle_time=0.1):
"""Stops current script and releases the axes"""
@@ -382,7 +322,7 @@ class AerotechAbrStage(PsiDeviceBase):
# Subscribe and wait for update
# status = SubscriptionStatus(self.task1, is_idle, timeout=timeout, settle_time=0.5)
status = SubscriptionStatus(
self.raster_scan_done, is_idle, timeout=timeout, settle_time=0.5
self.scan_done, is_idle, timeout=timeout, settle_time=0.5
)
return status
@@ -405,6 +345,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,32 +358,9 @@ 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):
"""Read axis mode"""
return self.axisAxesMode.get()
# @property
@@ -485,17 +403,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
+5 -9
View File
@@ -8,10 +8,10 @@ import types
from ophyd import Component, PVPositioner, Signal, EpicsSignal, EpicsSignalRO, Kind, PositionerBase
from ophyd.status import Status, MoveStatus
from .A3200enums import AbrMode
from bec_lib import bec_logger
from .A3200enums import AbrMode
logger = bec_logger.logger
@@ -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
@@ -94,6 +89,7 @@ class A3200Axis(PVPositioner):
vmax = Component(Signal, kind=Kind.config)
offset = Component(EpicsSignal, "-OFF", put_complete=True, kind=Kind.config)
#pylint: disable=too-many-arguments
def __init__(
self,
prefix="",
@@ -115,7 +111,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):
def maybe_add_prefix(self, _, kw, suffix):
# Patched not to enforce parent prefix when no parent
if kw in self.add_prefix:
return suffix
@@ -246,5 +242,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
+83
View File
@@ -0,0 +1,83 @@
# -*- coding: utf-8 -*-
"""
``NDArrayPreview`` --- Standalone Preview for ImagePlugin
*********************************************************
This module provides a standalone object to receive images to ophyd from the
AreaDetector's ImagePlugin.
Created on Wed Jan 29 2025
@author: mohacsi_i
"""
import numpy as np
from ophyd import Device, Component, EpicsSignal, Kind, Staged
from ophyd.areadetector import NDDerivedSignal
from bec_lib import bec_logger
logger = bec_logger.logger
class NDArrayPreview(Device):
"""Wrapper class around AreaDetector's NDStdArray plugins
This is a standalone class to display images from AreaDetector's
ImagePlugin without using a parent device. It also offers BEC exposed
methods to transfer image and change image array Kind-ness.
NOTE: As an explicit request, it can toggle data recording
"""
# Subscriptions for plotting image
USER_ACCESS = ["image", "savemode"]
SUB_MONITOR = "device_monitor_2d"
_default_sub = SUB_MONITOR
# Status attributes
array_size_x = Component(EpicsSignal, "ArraySize0_RBV", kind=Kind.config)
array_size_y = Component(EpicsSignal, "ArraySize1_RBV", kind=Kind.config)
array_size_z = Component(EpicsSignal, "ArraySize2_RBV", kind=Kind.config)
ndimensions = Component(EpicsSignal, "NDimensions_RBV", kind=Kind.config)
array_data = Component(EpicsSignal, "ArrayData", kind=Kind.omitted)
shaped_image = Component(
NDDerivedSignal,
derived_from="array_data",
shape=("array_size_z", "array_size_y", "array_size_x"),
num_dimensions="ndimensions",
kind=Kind.omitted,
)
def read(self):
"""Stream out data on every read()"""
if self._staged == Staged.yes:
image = self.shaped_image.get()
self._run_subs(sub_type=self.SUB_MONITOR, value=image)
return super().read()
def savemode(self, save=False):
"""Toggle save mode for the shaped image"""
# pylint: disable=protected-access
if save:
self.shaped_image._kind = Kind.normal
else:
self.shaped_image._kind = Kind.omitted
def image(self):
"""Fallback method in case image streaming fills up the BEC"""
array_size = (self.array_size_z.get(), self.array_size_y.get(), self.array_size_x.get())
if array_size == (0, 0, 0):
raise RuntimeError("Invalid image; ensure array_callbacks are on")
if array_size[-1] == 0:
array_size = array_size[:-1]
image = self.array_data.get()
return np.array(image).reshape(array_size)
# Automatically connect to SAMCAM at PXIII if directly invoked
if __name__ == "__main__":
img = NDArrayPreview("X06DA-SAMCAM:image1:", name="samimg")
img.wait_for_connection()
pxiii_bec/devices/SamCamDetector.py
+49
View File
@@ -0,0 +1,49 @@
# -*- coding: utf-8 -*-
"""
``SamCam`` --- Sample Camera control software
*********************************************
This module provides an object to control the sample camera at the PX III
beamline. The camera should run continously and stream data via ZMQ for
the GUI and alignment scripts.
Created on Thu Jan 30 2025
@author: mohacsi_i
"""
from ophyd import ADComponent
from ophyd_devices.devices.areadetector.cam import GenICam
from ophyd_devices.devices.areadetector.plugins import ImagePlugin_V35
from ophyd_devices.interfaces.base_classes.psi_detector_base import (
PSIDetectorBase,
CustomDetectorMixin,
)
from bec_lib import bec_logger
logger = bec_logger.logger
class SamCamSetup(CustomDetectorMixin):
def on_stage(self):
"""Just make sure it's running continously"""
self.parent.cam.acquire.put(1, wait=True)
def on_unstage(self):
"""Should run continously"""
def on_stop(self):
"""Should run continously"""
class SamCamDetector(PSIDetectorBase):
"""Sample camera device
The SAMCAM continously streams images to the GUI and sample alignment
scripts via ZMQ.
"""
custom_prepare_cls = SamCamSetup
cam = ADComponent(GenICam, "cam1:")
image = ADComponent(ImagePlugin_V35, "image1:")
pxiii_bec/devices/SmarGon.py
+91 -28
View File
@@ -1,6 +1,16 @@
"""
``SmarGon`` --- SmarGon control software
******************************************
The module provides an object to control the SmarGon goniometer axes at PX III.
The SmarGon axes are interfaced as positioners.
"""
import time
from threading import Thread, Lock
import requests
from ophyd import Component, Device, Kind, Signal, SignalRO
from ophyd import Component, Kind, Signal, PVPositioner
from ophyd.status import SubscriptionStatus
try:
from bec_lib import bec_logger
@@ -12,6 +22,11 @@ except ModuleNotFoundError:
logger = logging.getLogger("SmarGon")
# SmarGon contoller can't really handle multiple connections
# Use this mutex to ensure one access at a time
mutex = Lock()
class SmarGonSignal(Signal):
"""SmarGonSignal (R/W)
@@ -29,15 +44,16 @@ class SmarGonSignal(Signal):
def put(self, value, *, timestamp=None, **kwargs):
"""Overriden put to add communication with smargopolo"""
# Validate new value
# Validate new value and get timestamp
self.check_value(value)
if timestamp is None:
timestamp = time.time()
# Perform the actual write to SmargoPolo
r = self.parent._go_n_put(f"{self.write_addr}?{self.addr.upper()}={value}", **kwargs)
# pylint: disable=protected-access
r = self.parent._go_n_put(f"{self.write_addr}?{self.addr.upper()}={value}")
# pylint: disable=attribute-defined-outside-init
old_value = self._readback
self._timestamp = timestamp
self._readback = r[self.addr.upper()]
@@ -63,56 +79,74 @@ class SmarGonSignal(Signal):
if value > hil:
raise ValueError(f"Target {value} outside of limits {self.limits}")
def get(self, *args, **kwargs):
def get(self, **kwargs):
# pylint: disable=protected-access
r = self.parent._go_n_get(self.write_addr)
# print(r)
if isinstance(r, dict):
self._value = r[self.addr.upper()]
else:
self._value = r
return super().get(*args, **kwargs)
# pylint: disable=attribute-defined-outside-init
self._value = r[self.addr.upper()] if isinstance(r, dict) else r
return super().get(**kwargs)
class SmarGonSignalRO(Signal):
"""Small helper class for read-only parameters PVs from SmarGon.
TODO: Add monitoring
Reads and optionally monitors a variable on the SmarGon.
"""
def __init__(self, *args, read_addr="readbackSCS", **kwargs):
def __init__(self, *args, read_addr="readbackSCS", auto_monitor=False, **kwargs):
super().__init__(*args, **kwargs)
self._metadata["write_access"] = False
self.read_addr = read_addr
self.addr = self.parent.name
def get(self, *args, **kwargs):
if auto_monitor:
self._mon = Thread(target=self.poll, daemon=True)
self._mon.start()
def get(self, **kwargs):
# pylint: disable=protected-access
r = self.parent._go_n_get(self.read_addr)
# print(r)
if isinstance(r, dict):
self.put(r[self.addr.upper()], force=True)
else:
self.put(r, force=True)
return super().get(*args, **kwargs)
return self._readback
def poll(self):
"""Fooo"""
time.sleep(2)
while True:
time.sleep(0.25)
try:
self.get()
except requests.ConnectTimeout as ex:
logger.error(f"[{self.name}] {ex}")
class SmarGonAxis(Device):
class SmarGonAxis(PVPositioner):
"""SmarGon client deice
This class controls the SmarGon goniometer via the REST interface.
This class controls the SmarGon goniometer via the REST interface. All
SmarGon axes share a common mutex to manage actual HW access.
"""
USER_ACCESS = ["omove"]
# Status attributes
sg_url = Component(Signal, kind=Kind.config, metadata={"write_access": False})
corr = Component(SmarGonSignalRO, read_addr="corr_type", kind=Kind.config)
mode = Component(SmarGonSignalRO, read_addr="mode", kind=Kind.config)
# Axis parameters
readback = Component(SmarGonSignalRO, kind=Kind.hinted)
readback = Component(SmarGonSignalRO, kind=Kind.hinted, auto_monitor=True)
setpoint = Component(SmarGonSignal, kind=Kind.normal)
done = Component(SignalRO, value=1, kind=Kind.normal)
done = Component(Signal, value=1, kind=Kind.normal)
# moving = Component(SmarGonMovingSignalRO, kind=Kind.config)
moving = 1
_tol = 0.001
# pylint: disable=too-many-arguments
def __init__(
self,
prefix="SCS",
@@ -122,7 +156,6 @@ class SmarGonAxis(Device):
read_attrs=None,
configuration_attrs=None,
parent=None,
device_manager=None,
sg_url: str = "http://x06da-smargopolo.psi.ch:3000",
low_limit=None,
high_limit=None,
@@ -132,7 +165,7 @@ class SmarGonAxis(Device):
self.__class__.__dict__["setpoint"].kwargs["write_addr"] = f"target{prefix}"
self.__class__.__dict__["setpoint"].kwargs["low_limit"] = low_limit
self.__class__.__dict__["setpoint"].kwargs["high_limit"] = high_limit
self.__class__.__dict__["sg_url"].kwargs["value"] = sg_url
super().__init__(
prefix=prefix,
name=name,
@@ -140,11 +173,8 @@ class SmarGonAxis(Device):
read_attrs=read_attrs,
configuration_attrs=configuration_attrs,
parent=parent,
# device_manager=device_manager,
**kwargs,
)
self.sg_url._metadata["write_access"] = False
self.sg_url.set(sg_url, force=True).wait()
def initialize(self):
"""Helper function for initial readings"""
@@ -153,10 +183,38 @@ class SmarGonAxis(Device):
r = self._go_n_get("corr_type")
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
status.wait()
def on_target(*, value, **_):
distance = abs(value - position)
print(distance)
return bool(distance < self._tol)
status = SubscriptionStatus(self.readback, on_target, timeout=timeout, settle_time=0.1)
return status
def _pos_changed(self, timestamp=None, value=None, **kwargs):
pass
def _go_n_get(self, address, **kwargs):
"""Helper function to connect to smargopolo"""
cmd = f"{self.sg_url.get()}/{address}"
r = requests.get(cmd, timeout=1, **kwargs)
try:
with mutex:
r = requests.get(cmd, timeout=1, **kwargs)
except TimeoutError:
with mutex:
r = requests.get(cmd, timeout=1, **kwargs)
if not r.ok:
raise RuntimeError(
f"[{self.name}] Error getting {address}; reply was {r.status_code} => {r.reason}"
@@ -166,7 +224,12 @@ class SmarGonAxis(Device):
def _go_n_put(self, address, **kwargs):
"""Helper function to connect to smargopolo"""
cmd = f"{self.sg_url.get()}/{address}"
r = requests.put(cmd, timeout=1, **kwargs)
try:
with mutex:
r = requests.put(cmd, timeout=1, **kwargs)
except TimeoutError:
with mutex:
r = requests.put(cmd, timeout=1, **kwargs)
if not r.ok:
raise RuntimeError(
f"[{self.name}] Error putting {address}; reply was {r.status_code} => {r.reason}"
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
+206
View File
@@ -0,0 +1,206 @@
# -*- coding: utf-8 -*-
"""
Standard DAQ preview image stream module
Created on Thu Jun 27 17:28:43 2024
@author: mohacsi_i
"""
import json
from time import sleep, time
from threading import Thread
import zmq
import numpy as np
from ophyd import Device, Signal, Component, Kind
from ophyd_devices.interfaces.base_classes.psi_detector_base import (
CustomDetectorMixin,
PSIDetectorBase,
)
from bec_lib import bec_logger
logger = bec_logger.logger
ZMQ_TOPIC_FILTER = b""
class StdDaqPreviewMixin(CustomDetectorMixin):
"""Setup class for the standard DAQ preview stream
Parent class: CustomDetectorMixin
"""
_mon = None
def on_stage(self):
"""Start listening for preview data stream"""
if self._mon is not None:
self.parent.unstage()
sleep(0.5)
logger.info(f"[{self.parent.name}] Attaching monitor to {self.parent.url.get()}")
self.parent.connect()
self._stop_polling = False
self._mon = Thread(target=self.poll, daemon=True)
self._mon.start()
def on_unstage(self):
"""Stop a running preview"""
if self._mon is not None:
self._stop_polling = True
# Might hang on recv_multipart
self._mon.join(timeout=1)
# So also disconnect the socket
try:
#pylint: disable=protected-access
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"""
self.on_unstage()
def poll(self):
"""Collect streamed updates"""
try:
t_last = time()
while True:
try:
# Exit loop and finish monitoring
if self._stop_polling:
logger.info(f"[{self.parent.name}]\tDetaching monitor")
break
# pylint: disable=no-member
#pylint: disable=protected-access
r = self.parent._socket.recv_multipart(flags=zmq.NOBLOCK)
# Length and throtling checks
if len(r) != 2:
logger.warning(
f"[{self.parent.name}] Received malformed array of length {len(r)}"
)
t_curr = time()
t_elapsed = t_curr - t_last
if t_elapsed < self.parent.throttle.get():
sleep(0.1)
continue
# Unpack the Array V1 reply to metadata and array data
meta, data = r
# Update image and update subscribers
header = json.loads(meta)
image = np.frombuffer(data, dtype=header["type"])
if image.size != np.prod(header["shape"]):
err = f"Unexpected array size of {image.size} for header: {header}"
raise ValueError(err)
image = image.reshape(header["shape"])
# Update image and update subscribers
self.parent.array_counter.put(header["frame"], force=True)
self.parent.ndimensions.put(len(header["shape"]), force=True)
self.parent.array_size.put(header["shape"], force=True)
# self.parent.array_data.put(data, force=True)
self.parent.shaped_image.put(image, force=True)
#pylint: disable=protected-access
self.parent._last_image = image
self.parent._run_subs(sub_type=self.parent.SUB_MONITOR, value=image)
t_last = t_curr
logger.info(
f"[{self.parent.name}] Updated frame {header['frame']}\t"
f"Shape: {header['shape']}\tMean: {np.mean(image):.3f}"
)
except ValueError:
# Happens when ZMQ partially delivers the multipart message
pass
except zmq.error.Again:
# Happens when receive queue is empty
sleep(0.1)
except Exception as ex:
logger.info(f"[{self.parent.name}]\t{str(ex)}")
raise
finally:
self._mon = None
logger.info(f"[{self.parent.name}]\tDetaching monitor")
class StdDaqPreviewDetector(PSIDetectorBase):
"""Detector wrapper class around the StdDaq preview image stream.
This was meant to provide live image stream directly from the StdDAQ but
also works with other ARRAY v1 streamers, like the AreaDetector ZMQ plugin.
Note that the preview stream must be already throtled in order to cope with
the incoming data and the python class might throttle it further.
NOTE: As an explicit request, it does not record the image data.
You can add a preview widget to the dock by:
cam_widget = gui.add_dock('cam_dock1').add_widget('BECFigure').image('daq_stream1')
"""
# Subscriptions for plotting image
USER_ACCESS = ["image", "savemode"]
SUB_MONITOR = "device_monitor_2d"
_default_sub = SUB_MONITOR
custom_prepare_cls = StdDaqPreviewMixin
# Configuration attributes
url = Component(Signal, kind=Kind.config, metadata={"write_access": False})
throttle = Component(Signal, value=0.25, kind=Kind.config)
# Streamed data status
array_counter = Component(Signal, kind=Kind.hinted, metadata={"write_access": False})
ndimensions = Component(Signal, kind=Kind.normal, metadata={"write_access": False})
array_size = Component(Signal, kind=Kind.normal, metadata={"write_access": False})
# array_data = Component(Signal, kind=Kind.omitted, metadata={"write_access": False})
shaped_image = Component(Signal, kind=Kind.omitted, metadata={"write_access": False})
_last_image = None
def __init__(
self, *args, url: str = "tcp://129.129.95.38:20000", parent: Device = None, **kwargs
) -> None:
super().__init__(*args, parent=parent, **kwargs)
self.url.set(url, force=True).wait()
# Connect to the DAQ
self.connect()
def connect(self):
"""Connect to te StDAQs PUB-SUB streaming interface
StdDAQ may reject connection for a few seconds when it restarts,
so if it fails, wait a bit and try to connect again.
"""
# pylint: disable=no-member
# Socket to talk to server
context = zmq.Context()
self._socket = context.socket(zmq.SUB)
self._socket.setsockopt(zmq.SUBSCRIBE, ZMQ_TOPIC_FILTER)
try:
self._socket.connect(self.url.get())
except ConnectionRefusedError:
sleep(1)
self._socket.connect(self.url.get())
def savemode(self, save=False):
""" Toggle save mode for the shaped image"""
#pylint: disable=protected-access
if save:
self.shaped_image._kind = Kind.normal
else:
self.shaped_image._kind = Kind.omitted
def image(self):
"""
Gets the last image as an attribute in case image must be abandoned
due to some caching on the BEC.
"""
return self._last_image
# Automatically connect to MicroSAXS testbench if directly invoked
if __name__ == "__main__":
daq = StdDaqPreviewDetector(url="tcp://129.129.95.111:20000", name="preview")
daq.wait_for_connection()
pxiii_bec/devices/__init__.py
+3
View File
@@ -7,3 +7,6 @@ Ophyd devices for the PX III beamline, including the MX specific Aerotech A3200
from .A3200 import AerotechAbrStage
from .A3200utils import A3200Axis
from .SmarGon import SmarGonAxis
from .StdDaqPreview import StdDaqPreviewDetector
from .NDArrayPreview import NDArrayPreview
from .SamCamDetector import SamCamDetector
pxiii_bec/scans/__init__.py
+1
View File
@@ -3,4 +3,5 @@ from .mx_measurements import (
MeasureVerticalLine,
MeasureRasterSimple,
MeasureScreening,
MeasureHelical,
)
pxiii_bec/scans/mx_measurements.py
+106
View File
@@ -4,6 +4,8 @@ Scan primitives for standard BEC scans at the PX beamlines at SLS.
Theese scans define the event model and can be called from higher levels.
"""
import time
import numpy as np
from bec_lib import bec_logger
from bec_server.scan_server.scans import AsyncFlyScanBase
@@ -247,3 +249,107 @@ class MeasureScreening(AerotechFlyscanBase):
scan_name = "screeningscan"
required_kwargs = ["start", "range", "steps", "exp_time", "oscrange"]
class MeasureHelical(AerotechFlyscanBase):
"""Helical scan using the OMEGA motor
Measure an absolute continous line scan from `start` to `start` + `range`
during `move_time` on the Omega axis with PSO output.
The scan itself is executed by the scan service running on the Aerotech
controller. Ophyd just configures, launches it and waits for completion.
Example
-------
>>> scans.standard_wedge(start=42, range=10, move_time=20)
Parameters
----------
start : float
Scan start position of the axis.
range : float
Scan range of the axis.
move_time : float
Total travel time for the movement [s].
ready_rate : float, optional
No clue what is this... (default=500)
sg_start : (float, float, float, float, float)
Complete SmarGon coordinate in tuple form.
sg_end : (float, float, float, float, float)
Complete SmarGon coordinate in tuple form.
sg_steps : int
Number of steps with SmarGon.
"""
scan_name = "helicalscan"
required_kwargs = ["start", "range", "move_time", "sg_start", "sg_end", "sg_steps"]
def pre_scan(self):
"""Mostly just checking if ABR stage is ok..."""
# Smargon has no velocity control
self.smargon_start = np.array(self.caller_kwargs.get("sg_start"))
self.smargon_end = np.array(self.caller_kwargs.get("sg_end"))
self.smargon_steps = self.caller_kwargs.get("sg_steps")
self.smargon_range = self.smargon_end - self.smargon_start
self.smargon_step_size = self.smargon_range / self.smargon_steps
self.smargon_step_time = self.caller_kwargs.get("move_time") / self.smargon_steps
logger.info(f"Start:\t{self.smargon_start}")
logger.info(f"End:\t{self.smargon_end}")
logger.info(f"Steps:\t{self.smargon_steps}")
logger.info(f"Range:\t{self.smargon_range}")
logger.info(f"StepSize:\t{self.smargon_step_size}")
logger.info(f"StepTime:\t{self.smargon_step_time}")
# TODO: Move roughly to start position???
st0 = yield from self.stubs.send_rpc("shx", "omove", self.smargon_start[0])
st1 = yield from self.stubs.send_rpc("shy", "omove", self.smargon_start[1])
st2 = yield from self.stubs.send_rpc("shz", "omove", self.smargon_start[2])
st3 = yield from self.stubs.send_rpc("chi", "omove", self.smargon_start[3])
st4 = yield from self.stubs.send_rpc("phi", "omove", self.smargon_start[4])
st0.wait()
st1.wait()
st2.wait()
st3.wait()
st4.wait()
# Call super
yield from super().pre_scan()
def scan_core(self):
"""The actual scan logic comes here."""
# Kick off the run
yield from self.stubs.send_rpc_and_wait("abr", "kickoff")
logger.info("Measurement launched on the ABR stage...")
logger.info("Performing SmarGon stepping...")
for ss in range(self.smargon_steps):
sg_pos = self.smargon_start + ss * self.smargon_step_size
# Move to position but don't care
st0 = yield from self.stubs.send_rpc("shx", "omove", sg_pos[0])
st1 = yield from self.stubs.send_rpc("shy", "omove", sg_pos[1])
st2 = yield from self.stubs.send_rpc("shz", "omove", sg_pos[2])
st3 = yield from self.stubs.send_rpc("chi", "omove", sg_pos[3])
st4 = yield from self.stubs.send_rpc("phi", "omove", sg_pos[4])
t_start = time.time()
st0.wait()
st1.wait()
st2.wait()
st3.wait()
st4.wait()
t_end = time.time()
t_elapsed = t_end-t_start
time.sleep(max(self.smargon_step_time-t_elapsed, 0))
# Wait for scan task to finish
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()
pxiii_bec/scripts/scanwrappers.py
+52
View File
@@ -0,0 +1,52 @@
from bec_widgets.cli.client_utils import BECGuiClient
def ascan(
motor,
scan_start,
scan_end,
steps,
exp_time,
datasource,
**kwargs
):
"""Demo step scan with plotting
This is a small BEC user-space demo step scan. It tries to be a
standard BEC scan, while still setting up the environment.
Example:
--------
ascan(dev.dccm_energy, 12,13, steps=21, exp_time=0.1, datasource=dev.dccm_xbpm)
"""
# if not bl_check_beam():
# raise RuntimeError("Beamline is not in ready state")
# # GUI setup
# # Get or create gui
# gui = BECGuiClient()
# gui.start()
# window = None
# for _, val in gui.windows.items():
# if val.title == "Current scan":
# window = val.widget
# window.clear_all()
# if window is None:
# window = gui.new("Current scan")
# dock = window.add_dock(f"ScanDisplay {motor}")
# plt1 = dock.add_widget('BECWaveformWidget')
# plt1.plot(x_name=motor, y_name=datasource)
# plt1.set_x_label(motor)
# plt1.set_y_label(datasource)
print("Handing over to 'scans.line_scan'")
if 'relative' in kwargs:
del kwargs['relative']
scans.line_scan(motor, scan_start, scan_end, steps=steps, exp_time=exp_time, relative=False, **kwargs)
pyproject.toml
+1
View File
@@ -22,6 +22,7 @@ dependencies = [
"pyepics",
"pandas~=2.0",
"matplotlib",
"zmq",
]
[project.optional-dependencies]