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

Helical scan #20

Merged
mohacsi_i merged 15 commits from feature/helical-scan into main 2025-03-11 13:26:49 +01:00
Conversation 14 Commits 15 Files Changed 15 +771 -131
15 changed files with 771 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
pxiii_bec/bec_widgets/auto_updates.py
View File

@@ -1,7 +1,16 @@
from bec_widgets.cli.auto_updates import AutoUpdates, ScanInfo
from bec_widgets.cli.rpc.rpc_base import RPCResponseTimeoutError
class PlotUpdate(AutoUpdates):
create_default_dock = True
enabled = True
_scan_msg = None
def do_update(self, msg):
"""Save the original scan message for future use"""
self._scan_msg = msg
return super().do_update(msg)
# def simple_line_scan(self, info: ScanInfo) -> None:
# """
@@ -15,7 +24,42 @@ class PlotUpdate(AutoUpdates):
# plt = self.figure.plot(dev_x, dev_y)
# plt.set(title=f"PXIII: Scan {info.scan_number}", x_label=dev_x, y_label=dev_y)
def keyword_handler(self, info: ScanInfo) -> None:
"""Simple keyword handler
This simple keyword handler looks for the keyword 'datasource' in the scan arguments.
This allows the user to explictly specify the desired data source. Useful for alignment
scans.
"""
dev_x = info.scan_report_devices[0]
if "kwargs" in self._scan_msg.info:
dev_y = self._scan_msg.info["kwargs"].get("datasource", None)
else:
dev_y = None
if not dev_y:
return
plt1 = self.get_default_figure()
try:
# This will throw RPCResponseTimeoutError
plt1.clear_all()
except RPCResponseTimeoutError:
pass
try:
# TODO: What about 2D scans?
# This will throw RPCResponseTimeoutError
plt1.plot(x_name=dev_x, y_name=dev_y)
except RPCResponseTimeoutError:
pass
try:
# This will throw RPCResponseTimeoutError
plt1.set(title=f"PXIII: Scan {info.scan_number}", x_label=dev_x, y_label=dev_y)
except RPCResponseTimeoutError:
pass
# plt1.add_dap(dev_x, dev_y, dap="LinearModel")
def handler(self, info: ScanInfo) -> None:
"""Dock configuration handler"""
# EXAMPLES:
# if info.scan_name == "line_scan" and info.scan_report_devices:
# self.simple_line_scan(info)
@@ -24,3 +68,4 @@ class PlotUpdate(AutoUpdates):
# self.run_grid_scan_update(info)
# return
super().handler(info)
self.keyword_handler(info)

115
pxiii_bec/device_configs/x06da_device_config.yaml
View File

@@ -1,3 +1,40 @@
sldi_cenx:
description: FE slit-diaphragm horizontal center
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-FE-SLDI:CENX'}
onFailure: buffer
enabled: true
readoutPriority: monitored
readOnly: false
softwareTrigger: false
sldi_sizex:
description: FE slit-diaphragm horizontal size
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-FE-SLDI:GAPX'}
onFailure: buffer
enabled: true
readoutPriority: monitored
readOnly: false
softwareTrigger: false
sldi_ceny:
description: FE slit-diaphragm vertical center
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-FE-SLDI:CENY'}
onFailure: buffer
enabled: true
readoutPriority: monitored
readOnly: false
softwareTrigger: false
sldi_sizey:
description: FE slit-diaphragm vertical size
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-FE-SLDI:GAPY'}
onFailure: buffer
enabled: true
readoutPriority: monitored
readOnly: false
softwareTrigger: false
slh_trxr:
description: OP slit inner blade motion
deviceClass: ophyd.EpicsMotor
@@ -25,7 +62,7 @@ fi1_try:
readoutPriority: monitored
readOnly: false
softwareTrigger: false
dccm_pitch1:
dccm_theta1:
description: Monochromator pitch 1
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-OP-DCCM:PITCH1'}
@@ -34,15 +71,6 @@ dccm_pitch1:
readoutPriority: monitored
readOnly: false
softwareTrigger: false
dccm_energy1:
description: Monochromator energy 1
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-OP-DCCM:ENERGY1'}
onFailure: buffer
enabled: true
readoutPriority: monitored
readOnly: false
softwareTrigger: false
dccm_diode:
description: Diode between mono crystals
deviceClass: ophyd.EpicsSignalRO
@@ -52,7 +80,7 @@ dccm_diode:
readoutPriority: monitored
readOnly: true
softwareTrigger: false
dccm_pitch2:
dccm_theta2:
description: Monochromator pitch 2
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-OP-DCCM:PITCH2'}
@@ -61,15 +89,6 @@ dccm_pitch2:
readoutPriority: monitored
readOnly: false
softwareTrigger: false
dccm_energy2:
description: Monochromator energy 2
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-OP-DCCM:ENERGY2'}
onFailure: buffer
enabled: true
readoutPriority: monitored
readOnly: false
softwareTrigger: false
dccm_xbpm:
description: XBPM total intensity after monochromator
deviceClass: ophyd.EpicsSignalRO
@@ -80,18 +99,18 @@ dccm_xbpm:
readOnly: true
softwareTrigger: false
dccm_energy:
description: Monochromator energy using ECMC virtual motors
description: Monochromator energy
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-OP-DCCM:_ENERGY'}
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
dccm_eoffset:
description: Monochromator energy offset
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-OP-DCCM:_OFFSET'}
deviceConfig: {prefix: 'X06DA-OP-DCCM:EOFFSET'}
onFailure: buffer
enabled: true
readoutPriority: monitored
@@ -441,17 +460,17 @@ samstream:
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
# 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:
@@ -492,11 +511,11 @@ bstop_z:
softwareTrigger: false
bstop_pneum:
description: Beamstop pneumatic
deviceClass: ophyd.EpicsSignal
deviceConfig: {read_pv: 'X06DA-ES-BS:GET-POS', write_pv: 'X06DA-ES-BS:SET-POS', put_complete: true}
deviceClass: pxiii_bec.devices.PneumaticValve
deviceConfig: {read_pv: 'X06DA-ES-BS:GET-POS', write_pv: 'X06DA-ES-BS:SET-POS', kind: 'config', auto_monitor: true, put_complete: true}
onFailure: buffer
enabled: true
readoutPriority: monitored
readoutPriority: baseline
readOnly: false
softwareTrigger: false
bstop_diode:
@@ -511,7 +530,7 @@ bstop_diode:
frontlight:
description: Microscope frontlight
deviceClass: ophyd.EpicsSignal
deviceConfig: {read_pv: 'X06DA-ES-FL:SET-BRGHT', put_complete: true}
deviceConfig: {read_pv: 'X06DA-ES-FL:SET-BRGHT', kind: 'config', put_complete: true}
onFailure: buffer
enabled: true
readoutPriority: monitored
@@ -519,11 +538,11 @@ frontlight:
softwareTrigger: false
backlight:
description: Backlight reflector
deviceClass: ophyd.EpicsSignal
deviceConfig: {read_pv: 'X06DA-ES-BL:GET-POS', write_pv: 'X06DA-ES-BL:SET-POS', put_complete: true}
deviceClass: pxiii_bec.devices.PneumaticValve
deviceConfig: {read_pv: 'X06DA-ES-BL:GET-POS', write_pv: 'X06DA-ES-BL:SET-POS', kind: 'config', auto_monitor: true, put_complete: true}
onFailure: buffer
enabled: true
readoutPriority: monitored
readoutPriority: baseline
readOnly: false
softwareTrigger: false
det_y:
@@ -595,7 +614,7 @@ abr:
softwareTrigger: false
shx:
description: SmarGon X axis
deviceClass: pxiii_bec.devices.SmarGonAxis
deviceClass: pxiii_bec.devices.SmarGonAxisB
deviceConfig: {prefix: 'SCS', low_limit: -2, high_limit: 2, sg_url: 'http://x06da-smargopolo.psi.ch:3000'}
onFailure: buffer
enabled: true
@@ -604,7 +623,7 @@ shx:
softwareTrigger: false
shy:
description: SmarGon Y axis
deviceClass: pxiii_bec.devices.SmarGonAxis
deviceClass: pxiii_bec.devices.SmarGonAxisB
deviceConfig: {prefix: 'SCS', low_limit: -2, high_limit: 2, sg_url: 'http://x06da-smargopolo.psi.ch:3000'}
onFailure: buffer
enabled: true
@@ -613,7 +632,7 @@ shy:
softwareTrigger: false
shz:
description: SmarGon Z axis
deviceClass: pxiii_bec.devices.SmarGonAxis
deviceClass: pxiii_bec.devices.SmarGonAxisB
deviceConfig: {prefix: 'SCS', low_limit: 10, high_limit: 22, sg_url: 'http://x06da-smargopolo.psi.ch:3000'}
onFailure: buffer
enabled: true
@@ -622,7 +641,7 @@ shz:
softwareTrigger: false
chi:
description: SmarGon CHI axis
deviceClass: pxiii_bec.devices.SmarGonAxis
deviceClass: pxiii_bec.devices.SmarGonAxisB
deviceConfig: {prefix: 'SCS', low_limit: 0, high_limit: 40, sg_url: 'http://x06da-smargopolo.psi.ch:3000'}
onFailure: buffer
enabled: true
@@ -631,7 +650,7 @@ chi:
softwareTrigger: false
phi:
description: SmarGon PHI axis
deviceClass: pxiii_bec.devices.SmarGonAxis
deviceClass: pxiii_bec.devices.SmarGonAxisB
deviceConfig: {prefix: 'SCS', sg_url: 'http://x06da-smargopolo.psi.ch:3000'}
onFailure: buffer
enabled: true

39
pxiii_bec/devices/A3200.py
View File

@@ -12,11 +12,11 @@ Standard bluesky interface:
AerotechAbrStage.kickoff()
AerotechAbrStage.stop()
Additional bluesky functionality:
Aerotech.is_homed()
Aerotech.do_homing(wait=True)
Aerotech.get_ready(ostart=None, orange=None, etime=None, wait=True)
@@ -66,7 +66,6 @@ from bec_lib import bec_logger
logger = bec_logger.logger
# pylint: disable=logging-fstring-interpolation
class AerotechAbrMixin(CustomPrepare):
"""Configuration class for the Aerotech A3200 controller for the ABR stage"""
@@ -84,7 +83,13 @@ class AerotechAbrMixin(CustomPrepare):
scanargs = self.parent.scaninfo.scan_msg.info["kwargs"]
scanname = self.parent.scaninfo.scan_msg.info["scan_name"]
if scanname in ("standardscan", "helicalscan"):
if scanname in (
"standardscan",
"helicalscan",
"helicalscan1",
"helicalscan2",
"helicalscan3",
):
d["scan_command"] = AbrCmd.MEASURE_STANDARD
d["var_1"] = scanargs["start"]
d["var_2"] = scanargs["range"]
@@ -93,8 +98,8 @@ class AerotechAbrMixin(CustomPrepare):
d["var_5"] = 0
d["var_6"] = 0
d["var_7"] = 0
d["var_8"] = 0
d["var_9"] = 0
# d["var_8"] = 0
# d["var_9"] = 0
if scanname in ("verticallinescan", "vlinescan"):
d["scan_command"] = AbrCmd.VERTICAL_LINE_SCAN
d["var_1"] = scanargs["range"] / scanargs["steps"]
@@ -104,8 +109,8 @@ class AerotechAbrMixin(CustomPrepare):
d["var_5"] = 0
d["var_6"] = 0
d["var_7"] = 0
d["var_8"] = 0
d["var_9"] = 0
# d["var_8"] = 0
# d["var_9"] = 0
if scanname in ("screeningscan"):
d["scan_command"] = AbrCmd.SCREENING
d["var_1"] = scanargs["start"]
@@ -115,8 +120,8 @@ class AerotechAbrMixin(CustomPrepare):
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
# d["var_8"] = 0
# d["var_9"] = 0
if scanname in ("rasterscan", "rastersimplescan"):
d["scan_command"] = AbrCmd.RASTER_SCAN_SIMPLE
d["var_1"] = scanargs["exp_time"]
@@ -126,8 +131,8 @@ class AerotechAbrMixin(CustomPrepare):
d["var_5"] = scanargs["steps_y"]
d["var_6"] = 0
d["var_7"] = 0
d["var_8"] = 0
d["var_9"] = 0
# d["var_8"] = 0
# d["var_9"] = 0
# Reconfigure if got a valid scan config
if len(d) > 0:
@@ -157,7 +162,7 @@ class AerotechAbrStage(BECDeviceBase):
"""
custom_prepare_cls = AerotechAbrMixin
USER_ACCESS = ["reset", "kickoff", "bluekickoff", "complete", "set_axis_mode"]
USER_ACCESS = ["reset", "kickoff", "bluekickoff", "complete", "set_axis_mode", "arm", "disarm"]
taskStop = Component(EpicsSignal, "-AERO:TSK-STOP", put_complete=True, kind=Kind.omitted)
status = Component(EpicsSignal, "-AERO:STAT", put_complete=True, kind=Kind.omitted)
@@ -295,9 +300,7 @@ class AerotechAbrStage(BECDeviceBase):
return bool(value == 0)
# Subscribe and wait for update
status = SubscriptionStatus(
self.scan_done, is_busy, timeout=timeout, settle_time=0.1
)
status = SubscriptionStatus(self.scan_done, is_busy, timeout=timeout, settle_time=0.1)
status.wait()
# return status
@@ -321,9 +324,7 @@ class AerotechAbrStage(BECDeviceBase):
# Subscribe and wait for update
# status = SubscriptionStatus(self.task1, is_idle, timeout=timeout, settle_time=0.5)
status = SubscriptionStatus(
self.scan_done, is_idle, timeout=timeout, settle_time=0.5
)
status = SubscriptionStatus(self.scan_done, is_idle, timeout=timeout, settle_time=0.5)
return status
def reset(self, settle_time=0.1, wait_after_reload=1) -> None:

17
pxiii_bec/devices/A3200utils.py
View File

@@ -5,6 +5,7 @@ Created on Tue Jun 11 11:28:38 2024
@author: mohacsi_i
"""
import types
from collections import OrderedDict
from ophyd import Component, PVPositioner, Signal, EpicsSignal, EpicsSignalRO, Kind, PositionerBase
from ophyd.status import Status, MoveStatus
@@ -89,7 +90,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
# pylint: disable=too-many-arguments
def __init__(
self,
prefix="",
@@ -195,6 +196,18 @@ class A3200Axis(PVPositioner):
raise
return status
def describe(self):
"""Workaround to schema expected by the BEC"""
d = super().describe()
d[str(self.name)] = d[f"{self.name}_readback"]
return d
def read(self) -> OrderedDict[str, dict]:
"""Workaround to schema expected by the BEC"""
d = super().read()
d[str(self.name)] = d[f"{self.name}_readback"]
return d
def move(self, position, wait=True, timeout=None, moved_cb=None, **kwargs) -> MoveStatus:
"""Exposes the ophyd move command through BEC abstraction"""
return self.omove(position, wait=wait, timeout=timeout, moved_cb=moved_cb, **kwargs)
@@ -242,5 +255,5 @@ class A3200Axis(PVPositioner):
# Automatically start an axis if directly invoked
if __name__ == "__main__":
omega = A3200Axis(prefix="X06DA-ES-DF1:OMEGA", base_pv='X06DA-ES', name="omega")
omega = A3200Axis(prefix="X06DA-ES-DF1:OMEGA", base_pv="X06DA-ES", name="omega")
omega.wait_for_connection()

18
pxiii_bec/devices/NDArrayPreview.py
View File

@@ -11,7 +11,7 @@ Created on Wed Jan 29 2025
@author: mohacsi_i
"""
import numpy as np
from ophyd import Device, Component, EpicsSignal, Kind, Staged
from ophyd import Device, Component, EpicsSignal, EpicsSignalWithRBV, Kind, Staged
from ophyd.areadetector import NDDerivedSignal
@@ -20,6 +20,17 @@ from bec_lib import bec_logger
logger = bec_logger.logger
class SilentNDDerivedSignal(NDDerivedSignal):
"""Silent version of NDDerivedSignal, it does not spam the terminal on
every defective frame (shit happens, ok?)."""
def _array_shape_callback(self, **kwargs):
try:
super()._array_shape_callback(**kwargs)
except RuntimeError:
pass
class NDArrayPreview(Device):
"""Wrapper class around AreaDetector's NDStdArray plugins
@@ -36,13 +47,16 @@ class NDArrayPreview(Device):
_default_sub = SUB_MONITOR
# Status attributes
min_callback_time = Component(
EpicsSignalWithRBV, "MinCallbackTime", kind=Kind.config, put_complete=True
)
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,
SilentNDDerivedSignal,
derived_from="array_data",
shape=("array_size_z", "array_size_y", "array_size_x"),
num_dimensions="ndimensions",

47
pxiii_bec/devices/PneumaticValve.py Normal file
View File

@@ -0,0 +1,47 @@
from ophyd import EpicsSignal
from ophyd.status import SubscriptionStatus
class PneumaticValve(EpicsSignal):
"""Wrapper around EpicsSignal to wait until reaching target. Use the
status returned by set() to wait until movement is finished. Do NOT
use put if you want to wait, that's a low-level PV write op.
NOTE: The SET and GET states do not match exactly
"""
def set(self, value, *, timeout=5, settle_time=0.1):
"""Overloaded setter that waits for target state
NOTE: The SubscriptionStatus callback does not run in put()
"""
# Lazy hardcoded state lookup
target = 1 if value in (1, "Measure") else 2
# Define wait until an end state is reached
def on_target(*, value, **_):
return bool(value == target)
# Subscribe a monitor in advance and wait for update
status = SubscriptionStatus(self, on_target, timeout=timeout, settle_time=0.1)
# Set value to start movement
super().set(value, settle_time=settle_time).wait()
# Return the monitor
return status
def check_value(self, value):
"""Input validation"""
if value not in (0, 1, "Measure", "Park"):
raise ValueError(f"Unsupported pneumatic valve target {value}")
return super().check_value(value)
if __name__ == "__main__":
pneum = PneumaticValve(
read_pv="X06DA-ES-BS:GET-POS",
write_pv="X06DA-ES-BS:SET-POS",
auto_monitor=True,
put_complete=True,
name="bspump",
)
pneum.wait_for_connection()

9
pxiii_bec/devices/SamCamDetector.py
View File

@@ -13,7 +13,8 @@ Created on Thu Jan 30 2025
"""
from ophyd import ADComponent
from ophyd_devices.devices.areadetector.cam import GenICam
from ophyd_devices.devices.areadetector.plugins import ImagePlugin_V35
# from ophyd_devices.devices.areadetector.plugins import ImagePlugin_V35
from ophyd_devices.interfaces.base_classes.psi_detector_base import (
PSIDetectorBase,
CustomDetectorMixin,
@@ -25,6 +26,8 @@ logger = bec_logger.logger
class SamCamSetup(CustomDetectorMixin):
"""Simple camera mixin class, the SAMCAM is usually streaming"""
def on_stage(self):
"""Just make sure it's running continously"""
self.parent.cam.acquire.put(1, wait=True)
@@ -40,10 +43,10 @@ class SamCamDetector(PSIDetectorBase):
"""Sample camera device
The SAMCAM continously streams images to the GUI and sample alignment
scripts via ZMQ.
scripts via ZMQ.
"""
custom_prepare_cls = SamCamSetup
cam = ADComponent(GenICam, "cam1:")
image = ADComponent(ImagePlugin_V35, "image1:")
# image = ADComponent(ImagePlugin_V35, "image1:")

54
pxiii_bec/devices/SmarGon.py → pxiii_bec/devices/SmarGonA.py
View File

@@ -9,6 +9,8 @@ The SmarGon axes are interfaced as positioners.
import time
from threading import Thread, Lock
import requests
from requests.adapters import HTTPAdapter, Retry
from collections import OrderedDict
from ophyd import Component, Kind, Signal, PVPositioner
from ophyd.status import SubscriptionStatus
@@ -92,6 +94,7 @@ class SmarGonSignalRO(Signal):
"""Small helper class for read-only parameters PVs from SmarGon.
Reads and optionally monitors a variable on the SmarGon.
"""
def __init__(self, *args, read_addr="readbackSCS", auto_monitor=False, **kwargs):
@@ -166,6 +169,12 @@ class SmarGonAxis(PVPositioner):
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
# Fine-tune HTTP connection behavior
# NOTE: SmarGon has a few failed requests every one in a while
self._s = requests.Session()
retries = Retry(total=5, backoff_factor=0.05, status_forcelist=[500, 502, 503, 504])
self._s.mount("http://", HTTPAdapter(max_retries=retries))
super().__init__(
prefix=prefix,
name=name,
@@ -187,22 +196,33 @@ class SmarGonAxis(PVPositioner):
"""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):
def omove(self, position, wait=True, timeout=2.0, moved_cb=None):
"""Original move command without the BEC wrappers"""
status = self.setpoint.set(position, settle_time=0.1)
status.wait()
if not wait:
return status
status.wait()
def on_target(*, value, **_):
distance = abs(value - position)
print(distance)
distance = abs(value - self.setpoint._value)
print(f"[self.name] Distance: {distance}")
return bool(distance < self._tol)
status = SubscriptionStatus(self.readback, on_target, timeout=timeout, settle_time=0.1)
return status
def describe(self):
"""Workaround to schema expected by the BEC"""
d = super().describe()
d[str(self.name)] = d[f"{self.name}_readback"]
return d
def read(self) -> OrderedDict[str, dict]:
"""Workaround to schema expected by the BEC"""
d = super().read()
d[str(self.name)] = d[f"{self.name}_readback"]
return d
def _pos_changed(self, timestamp=None, value=None, **kwargs):
pass
@@ -211,10 +231,16 @@ class SmarGonAxis(PVPositioner):
cmd = f"{self.sg_url.get()}/{address}"
try:
with mutex:
r = requests.get(cmd, timeout=1, **kwargs)
r = self._s.get(cmd, timeout=1, **kwargs)
except TimeoutError:
with mutex:
r = requests.get(cmd, timeout=1, **kwargs)
try:
time.sleep(0.05)
with mutex:
r = self._s.get(cmd, timeout=0.5, **kwargs)
except TimeoutError:
time.sleep(0.05)
with mutex:
r = self._s.get(cmd, timeout=0.5, **kwargs)
if not r.ok:
raise RuntimeError(
f"[{self.name}] Error getting {address}; reply was {r.status_code} => {r.reason}"
@@ -226,10 +252,16 @@ class SmarGonAxis(PVPositioner):
cmd = f"{self.sg_url.get()}/{address}"
try:
with mutex:
r = requests.put(cmd, timeout=1, **kwargs)
r = self._s.put(cmd, timeout=1, **kwargs)
except TimeoutError:
with mutex:
r = requests.put(cmd, timeout=1, **kwargs)
try:
time.sleep(0.05)
with mutex:
r = self._s.put(cmd, timeout=0.5, **kwargs)
except TimeoutError:
time.sleep(0.05)
with mutex:
r = self._s.put(cmd, timeout=0.5, **kwargs)
if not r.ok:
raise RuntimeError(
f"[{self.name}] Error putting {address}; reply was {r.status_code} => {r.reason}"

249
pxiii_bec/devices/SmarGonB.py Normal file
View File

@@ -0,0 +1,249 @@
"""
``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
import threading
from collections import OrderedDict
import requests
from requests.adapters import HTTPAdapter, Retry
from ophyd import Component, Kind, Signal, PVPositioner
from ophyd.status import SubscriptionStatus
try:
from bec_lib import bec_logger
logger = bec_logger.logger
except ModuleNotFoundError:
import logging
logger = logging.getLogger("SmarGon")
# SmarGon contoller can't really handle multiple connections
# Use this mutex to ensure one access at a time
mutex = threading.Lock()
class LimitedSmarGonSignal(Signal):
"""SmarGonSignal (R/W)
Small helper class to read/write parameters from SmarGon. As there is no
motion status readback from smargopolo, this should be substituted with
setting with 'settle_time'.
"""
def __init__(self, *args, write_addr="targetSCS", low_limit=None, high_limit=None, **kwargs):
self._limits = (low_limit, high_limit)
super().__init__(*args, **kwargs)
self.write_addr = write_addr
@property
def limits(self):
return self._limits
def check_value(self, value, **kwargs):
"""Check if value falls within limits"""
lol = self.limits[0]
if lol is not None:
if value < lol:
raise ValueError(f"Target {value} outside of limits {self.limits}")
hil = self.limits[1]
if hil is not None:
if value > hil:
raise ValueError(f"Target {value} outside of limits {self.limits}")
def put(self, value, *, timestamp=None, force=False, metadata=None, **kwargs):
"""Overriden put to add communication with smargopolo"""
# Validate new value and get timestamp
if not force:
self.check_value(value)
if timestamp is None:
timestamp = time.time()
# Perform the actual write to SmargoPolo
# pylint: disable=protected-access
r = self.parent._go_n_put(f"{self.write_addr}?{self.parent.name.upper()}={value}")
# pylint: disable=attribute-defined-outside-init
old_value = self._readback
self._timestamp = timestamp
self._readback = r[self.parent.name.upper()]
self._value = r[self.parent.name.upper()]
# Notify subscribers
self._run_subs(
sub_type=self.SUB_VALUE, old_value=old_value, value=value, timestamp=self._timestamp
)
class SmarGonAxis(PVPositioner):
"""SmarGon client deice
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", "oldmove"]
# Status attributes
sg_url = Component(Signal, kind=Kind.config, metadata={"write_access": False})
# Axis parameters
readback = Component(Signal, kind=Kind.hinted, metadata={"write_access": False})
setpoint = Component(LimitedSmarGonSignal, kind=Kind.normal)
done = Component(Signal, value=1, kind=Kind.normal, metadata={"write_access": False})
_tol = 0.001
# pylint: disable=too-many-arguments
def __init__(
self,
prefix="SCS",
*,
name,
kind=None,
read_attrs=None,
configuration_attrs=None,
parent=None,
sg_url: str = "http://x06da-smargopolo.psi.ch:3000",
low_limit=None,
high_limit=None,
**kwargs,
) -> None:
# 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
# Fine-tune HTTP connection behavior
# NOTE: SmarGon has a few failed requests every one in a while
self._s = requests.Session()
retries = Retry(total=5, backoff_factor=0.05, status_forcelist=[500, 502, 503, 504])
self._s.mount("http://", HTTPAdapter(max_retries=retries))
super().__init__(
prefix=prefix,
name=name,
kind=kind,
read_attrs=read_attrs,
configuration_attrs=configuration_attrs,
parent=parent,
**kwargs,
)
def on_target():
"""Monitors the setpoint and readback and calculates the on_target flag"""
time.sleep(2)
while True:
# Read back target and setpoint values
# pylint: disable=protected-access
r = self._go_n_get("readbackSCS")
rb = r[self.name.upper()]
self.readback.set(rb, force=True).wait()
r = self._go_n_get("targetSCS")
sp = r[self.name.upper()]
self.setpoint._value = sp
# print(f"Readback: {rb}\tSetpoint: {sp}")
# Check if they're within tolerance
distance = abs(rb - sp)
done = 1 if distance < self._tol else 0
self.done.put(done, force=True)
time.sleep(0.2)
self._mon = threading.Thread(target=on_target, daemon=True)
self._mon.start()
def omove(self, position, wait=True, timeout=None, moved_cb=None):
"""Move command that's masked by BEC"""
self.done.put(0, force=True)
return self.move(position, wait, timeout, moved_cb)
def oldmove(self, position, wait=True, timeout=2.0, moved_cb=None):
"""Original move command without the BEC wrappers"""
status = self.setpoint.set(position, settle_time=0.1).wait()
if not wait:
return status
def on_target(*, value, **_):
distance = abs(value - self.setpoint._value)
print(f"[self.name] Distance: {distance}")
return bool(distance < self._tol)
status = SubscriptionStatus(self.readback, on_target, timeout=timeout, settle_time=0.1)
return status
def describe(self):
"""Workaround to schema expected by the BEC"""
d = super().describe()
d[str(self.name)] = d[f"{self.name}_readback"]
return d
def read(self) -> OrderedDict[str, dict]:
"""Workaround to schema expected by the BEC"""
d = super().read()
d[str(self.name)] = d[f"{self.name}_readback"]
return d
def _pos_changed(self, timestamp=None, value=None, **kwargs):
"""Remove EPICS dependency"""
pass
def _go_n_get(self, address, **kwargs):
"""Helper function to connect to smargopolo"""
cmd = f"{self.sg_url.get()}/{address}"
try:
with mutex:
r = self._s.get(cmd, timeout=1, **kwargs)
except TimeoutError:
try:
time.sleep(0.05)
with mutex:
r = self._s.get(cmd, timeout=0.5, **kwargs)
except TimeoutError:
time.sleep(0.05)
with mutex:
r = self._s.get(cmd, timeout=0.5, **kwargs)
if not r.ok:
raise RuntimeError(
f"[{self.name}] Error getting {address}; reply was {r.status_code} => {r.reason}"
)
return r.json()
def _go_n_put(self, address, **kwargs):
"""Helper function to connect to smargopolo"""
cmd = f"{self.sg_url.get()}/{address}"
try:
with mutex:
r = self._s.put(cmd, timeout=1, **kwargs)
except TimeoutError:
try:
time.sleep(0.05)
with mutex:
r = self._s.put(cmd, timeout=0.5, **kwargs)
except TimeoutError:
time.sleep(0.05)
with mutex:
r = self._s.put(cmd, timeout=0.5, **kwargs)
if not r.ok:
raise RuntimeError(
f"[{self.name}] Error putting {address}; reply was {r.status_code} => {r.reason}"
)
return r.json()
if __name__ == "__main__":
shx = SmarGonAxis(prefix="SCS", name="shx", sg_url="http://x06da-smargopolo.psi.ch:3000")
shy = SmarGonAxis(prefix="SCS", name="shy", sg_url="http://x06da-smargopolo.psi.ch:3000")
shz = SmarGonAxis(
prefix="SCS",
name="shz",
low_limit=10,
high_limit=22,
sg_url="http://x06da-smargopolo.psi.ch:3000",
)
shx.wait_for_connection()
shy.wait_for_connection()
shz.wait_for_connection()

12
pxiii_bec/devices/StdDaqPreview.py
View File

@@ -51,7 +51,7 @@ class StdDaqPreviewMixin(CustomDetectorMixin):
self._mon.join(timeout=1)
# So also disconnect the socket
try:
#pylint: disable=protected-access
# pylint: disable=protected-access
self.parent._socket.disconnect(self.parent.url.get())
except zmq.error.ZMQError:
# Might be already closed
@@ -73,7 +73,7 @@ class StdDaqPreviewMixin(CustomDetectorMixin):
break
# pylint: disable=no-member
#pylint: disable=protected-access
# pylint: disable=protected-access
r = self.parent._socket.recv_multipart(flags=zmq.NOBLOCK)
# Length and throtling checks
@@ -105,7 +105,7 @@ class StdDaqPreviewMixin(CustomDetectorMixin):
# self.parent.array_data.put(data, force=True)
self.parent.shaped_image.put(image, force=True)
#pylint: disable=protected-access
# 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
@@ -133,7 +133,7 @@ class StdDaqPreviewDetector(PSIDetectorBase):
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.
the incoming data and the python class might throttle it further.
NOTE: As an explicit request, it does not record the image data.
@@ -185,8 +185,8 @@ class StdDaqPreviewDetector(PSIDetectorBase):
self._socket.connect(self.url.get())
def savemode(self, save=False):
""" Toggle save mode for the shaped image"""
#pylint: disable=protected-access
"""Toggle save mode for the shaped image"""
# pylint: disable=protected-access
if save:
self.shaped_image._kind = Kind.normal
else:

4
pxiii_bec/devices/__init__.py
View File

@@ -6,7 +6,9 @@ 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 .SmarGonA import SmarGonAxis as SmarGonAxisA
from .SmarGonB import SmarGonAxis as SmarGonAxisB
from .StdDaqPreview import StdDaqPreviewDetector
from .NDArrayPreview import NDArrayPreview
from .SamCamDetector import SamCamDetector
from .PneumaticValve import PneumaticValve

1
pxiii_bec/scans/__init__.py
View File

@@ -4,4 +4,5 @@ from .mx_measurements import (
MeasureRasterSimple,
MeasureScreening,
MeasureHelical,
MeasureHelical2,
)

142
pxiii_bec/scans/mx_measurements.py
View File

@@ -1,4 +1,4 @@
""" MX measurements module
"""MX measurements module
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.
@@ -102,10 +102,6 @@ class AerotechFlyscanBase(AsyncFlyScanBase):
# Call super
yield from super().pre_scan()
# def stage(self):
# """ ToDo: Sot sure if we should call super() here as it'd stage the whole beamline"""
# return super().stage()
def scan_core(self):
"""The actual scan logic comes here."""
# Kick off the run
@@ -275,18 +271,16 @@ class MeasureHelical(AerotechFlyscanBase):
ready_rate : float, optional
No clue what is this... (default=500)
sg_start : (float, float, float, float, float)
Complete SmarGon coordinate in tuple form.
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.
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..."""
@@ -305,7 +299,7 @@ class MeasureHelical(AerotechFlyscanBase):
logger.info(f"StepSize:\t{self.smargon_step_size}")
logger.info(f"StepTime:\t{self.smargon_step_time}")
# TODO: Move roughly to start position???
# 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])
@@ -342,8 +336,132 @@ class MeasureHelical(AerotechFlyscanBase):
st3.wait()
st4.wait()
t_end = time.time()
t_elapsed = t_end-t_start
time.sleep(max(self.smargon_step_time-t_elapsed, 0))
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()
class MeasureHelical2(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 = "helicalscan2"
required_kwargs = ["start", "range", "move_time", "sg_start", "sg_end", "sg_steps"]
point_id = 0
# def __init__(self, *args, parameter: dict = None, **kwargs):
# """Just set num_pos=0 to avoid hanging and override defaults if explicitly set from
# parameters.
# """
# self.num_pos = kwargs["sg_steps"]
# super().__init__(*args, parameter=parameter, **kwargs)
def prepare_positions(self):
# 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}")
self.num_pos = self.smargon_steps
self.positions = np.linspace(self.smargon_start, self.smargon_end, self.smargon_steps)
self.start_pos = self.positions[0, :]
# Call super
yield from super().prepare_positions()
# def update_scan_motors(self):
# """ Update step scan motors"""
# self.scan_motors = ['shx', 'shy', 'shz', 'chi', 'phi']
def pre_scan(self):
"""Mostly just checking if ABR stage is ok..."""
# Move roughly to start position
st0 = yield from self.stubs.send_rpc("shx", "omove", self.start_pos[0])
st1 = yield from self.stubs.send_rpc("shy", "omove", self.start_pos[1])
st2 = yield from self.stubs.send_rpc("shz", "omove", self.start_pos[2])
st3 = yield from self.stubs.send_rpc("chi", "omove", self.start_pos[3])
st4 = yield from self.stubs.send_rpc("phi", "omove", self.start_pos[4])
st0.wait()
st1.wait()
st2.wait()
st3.wait()
st4.wait()
# print(f"\n\n{self.readout_priority}\n\n")
# 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 _, sg_pos in enumerate(self.positions):
# 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))
yield from self.stubs.read(group="monitored", point_id=self.point_id)
self.point_id += 1
# Wait for scan task to finish
if self.abr_complete:

68
pxiii_bec/scripts/beamlinescripts.py Normal file
View File

@@ -0,0 +1,68 @@
import bec
import bec_lib.devicemanager.DeviceContainer as dev
def rock(steps, exp_time, scan_start=None, scan_end=None, datasource=None, visual=True, **kwargs):
"""Demo step scan with plotting
This is a simple user-space demo step scan with the BEC. It 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)
"""
# Dummy method to check beamline status
if not bl_check_beam():
raise RuntimeError("Beamline is not in ready state")
motor = dev.dccm_theta2
if scan_start is None:
scan_start = -0.05 / dev.dccm_energy.user_readback.get()
if scan_end is None:
scan_end = 0.05 / dev.dccm_energy.user_readback.get()
if visual:
# Get or create scan specific window
window = None
for _, val in bec.gui.windows.items():
if val.title == "CurrentScan":
window = val.widget
window.clear_all()
if window is None:
window = bec.gui.new("CurrentScan")
# Draw a simploe plot in the window
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)
plt1.add_dap(motor, datasource, dap="LinearModel")
window.show()
print("Handing over to 'scans.line_scan'")
s = scans.line_scan(
motor,
scan_start,
scan_end,
steps=steps,
exp_time=exp_time,
datasource=datasource,
relative=True,
**kwargs,
)
if visual:
# If fitting via GUI
firt_par = plt1.get_dap_params()
else:
# Without GUI
firt_par = bec.dap.LinearModel.fit(
s, motor.name, motor.name, datasource.name, datasource.name
)
# TODO: Validate fitted position
# TODO: Move to fitted maximum
return s, firt_par

82
pxiii_bec/scripts/scanwrappers.py
View File

@@ -1,7 +1,10 @@
from bec_widgets.cli.client_utils import BECGuiClient
import bec
import bec_lib.devicemanager.DeviceContainer as dev
def bl_check_beam():
"""Check beamline status before scan"""
return True
def ascan(
@@ -10,43 +13,68 @@ def ascan(
scan_end,
steps,
exp_time,
datasource,
**kwargs
datasource=None,
visual=True,
relative=False,
**kwargs,
):
"""Demo step scan with plotting
This is a small BEC user-space demo step scan. It tries to be a
This is a simple user-space demo step scan with the BEC. It 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")
# Dummy method to check beamline status
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)
if visual:
# Get or create scan specific window
window = None
for _, val in bec.gui.windows.items():
if val.title == "CurrentScan":
window = val.widget
window.clear_all()
if window is None:
window = bec.gui.new("CurrentScan")
# Draw a simploe plot in the window
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)
plt1.add_dap(motor, datasource, dap="LinearModel")
window.show()
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)
s = scans.line_scan(
motor,
scan_start,
scan_end,
steps=steps,
exp_time=exp_time,
datasource=datasource,
relative=relative,
**kwargs,
)
if visual:
# Fitting via GUI
firt_par = plt1.get_dap_params()
else:
# Fitting without GUI
firt_par = bec.dap.LinearModel.fit(
s, motor.name, motor.name, datasource.name, datasource.name
)
# # Some basic fit
# dkey = datasource.full_name
# NOTE: s.scan.data == bec.history[-1]
# datapoints = bec.history[-1].devices[dkey].read()[dkey]['value']
# positions
return s, firt_par