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

First helical scan passed

Browse Source
This commit is contained in:
gac-x06da
2025-01-30 10:43:38 +01:00
parent a455a490c6
commit 59bd4aeb9a
6 changed files with 217 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pxiii_bec/device_configs/x06da_device_config.yaml
+9 -1
View File
@@ -452,7 +452,15 @@ samimg:
readoutPriority: async
readOnly: false
softwareTrigger: false
samimg_ad:
description: Sample camera image via AD plugin
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
+7 -9
View File
@@ -113,7 +113,7 @@ class AerotechAbrMixin(CustomPrepare):
scanargs = self.parent.scaninfo.scan_msg.info["kwargs"]
scanname = self.parent.scaninfo.scan_msg.info["scan_name"]
if scanname in ("standardscan"):
if scanname in ("standardscan", "helicalscan"):
d["scan_command"] = AbrCmd.MEASURE_STANDARD
d["var_1"] = scanargs["start"]
d["var_2"] = scanargs["range"]
@@ -186,7 +186,7 @@ class AerotechAbrStage(BECDeviceBase):
"""
custom_prepare_cls = AerotechAbrMixin
USER_ACCESS = ["reset", "kickoff", "complete", "set_axis_mode"]
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)
@@ -236,10 +236,7 @@ class AerotechAbrStage(BECDeviceBase):
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.
@@ -328,9 +325,10 @@ class AerotechAbrStage(BECDeviceBase):
# 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"""
@@ -353,7 +351,7 @@ class AerotechAbrStage(BECDeviceBase):
# 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
pxiii_bec/devices/SmarGon.py
+42 -23
View File
@@ -1,5 +1,12 @@
"""
``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
from threading import Thread, Lock
import requests
from ophyd import Component, Kind, Signal, PVPositioner
from ophyd.status import SubscriptionStatus
@@ -14,6 +21,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)
@@ -31,17 +43,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
#pylint: disable=protected-access
# pylint: disable=protected-access
r = self.parent._go_n_put(f"{self.write_addr}?{self.addr.upper()}={value}")
#pylint: disable=attribute-defined-outside-init
# pylint: disable=attribute-defined-outside-init
old_value = self._readback
self._timestamp = timestamp
self._readback = r[self.addr.upper()]
@@ -68,11 +79,10 @@ class SmarGonSignal(Signal):
raise ValueError(f"Target {value} outside of limits {self.limits}")
def get(self, **kwargs):
#pylint: disable=protected-access
# pylint: disable=protected-access
r = self.parent._go_n_get(self.write_addr)
# print(r)
#pylint: disable=attribute-defined-outside-init
# pylint: disable=attribute-defined-outside-init
self._value = r[self.addr.upper()] if isinstance(r, dict) else r
return super().get(**kwargs)
@@ -80,7 +90,7 @@ class SmarGonSignal(Signal):
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", auto_monitor=False, **kwargs):
@@ -94,7 +104,7 @@ class SmarGonSignalRO(Signal):
self._mon.start()
def get(self, **kwargs):
#pylint: disable=protected-access
# pylint: disable=protected-access
r = self.parent._go_n_get(self.read_addr)
if isinstance(r, dict):
@@ -104,7 +114,7 @@ class SmarGonSignalRO(Signal):
return self._readback
def poll(self):
""" Fooo"""
"""Fooo"""
time.sleep(2)
while True:
time.sleep(0.25)
@@ -117,7 +127,8 @@ class SmarGonSignalRO(Signal):
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"]
@@ -132,10 +143,9 @@ class SmarGonAxis(PVPositioner):
setpoint = Component(SmarGonSignal, 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
# pylint: disable=too-many-arguments
def __init__(
self,
prefix="SCS",
@@ -173,11 +183,11 @@ class SmarGonAxis(PVPositioner):
print(r)
def move(self, position, wait=True, timeout=None, moved_cb=None):
""" Move command that's masked by BEC"""
"""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"""
"""Original move command without the BEC wrappers"""
status = self.setpoint.set(position, settle_time=0.1)
if not wait:
return status
@@ -185,12 +195,11 @@ class SmarGonAxis(PVPositioner):
status.wait()
def on_target(*, value, **_):
distance = abs(value-position)
distance = abs(value - position)
print(distance)
return bool(distance<self._tol)
status = SubscriptionStatus(
self.readback, on_target, timeout=timeout, settle_time=0.1
)
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):
@@ -199,7 +208,12 @@ class SmarGonAxis(PVPositioner):
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}"
@@ -209,7 +223,12 @@ class SmarGonAxis(PVPositioner):
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/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)