bec/debye_bec
0
0
Fork 0
Code Issues 9 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

Refactored advanced scan.

Browse Source 
Added calculator function to convert energy/angle.
This commit is contained in:
gac-x01da
2024-09-13 11:44:28 +02:00
parent 31dd582648
commit 0062da5a6b
5 changed files with 171 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+23 -13
View File
@@ -6,6 +6,8 @@ Debye-specific plugins and configs for BEC
### Open visual studio code
```
ssh x01da-bec-001
cd /data/test/x01da-test-bec/bec_deployment
code
```
### Git
@@ -15,29 +17,37 @@ git push origin feat/add_advanced_scan_modes
git status
```
### Start or restart BEC Server
### BEC Server
```
ssh x01da-bec-001
cd /data/test/x01da-test-bec/bec_deployment
. /data/test/x01da-test-bec/bec_deployment/bec_venv/bin/activate
```
Then
```
bec-server start
```
or
```
bec-server restart
bec-server stop
bec-server attach
```
after change in code:
- ctrl-c + ctrl-c to stop scan server and device server module
- restart server modules
To restart individual server modules:
- ctrl-c + ctrl-c to stop for example scan server or device server module
- restart server module(s)
### Start BEC Client
### BEC Client
```
ssh x01da-test-cons
ssh x01da-bec-001
cd /data/test/x01da-test-bec/bec_deployment
. /data/test/x01da-test-bec/bec_deployment/bec_venv/bin/activate
bec
```
#### Useful commands in bec
Update Session with specific config:
```
bec.config.update_session_with_file("debye_bec/debye_bec/device_configs/x01da_test_config.yaml")
```
Define folder and sample name for written files:
```
bec.system_config.file_directory="test"
bec.system_config.file_suffix ="sampleA"
```
debye_bec/devices/mo1_bragg.py
+72 -75
View File
@@ -12,8 +12,8 @@ import enum
import threading
import time
import traceback
import numpy as np
from scipy.interpolate import BSpline
from typeguard import typechecked
from debye_bec.devices.utils.mo1_bragg_utils import compute_spline
from dataclasses import dataclass
from typing import Literal
@@ -119,7 +119,7 @@ class Mo1BraggStatus(Device):
class Mo1BraggEncoder(Device):
"""Mo1 Bragg PVs to communicate with the encoder"""
enc_reinit = Cpt(EpicsSignal, suffix="enc_reinit.PROC", kind="config")
enc_reinit = Cpt(EpicsSignal, suffix="enc_reinit", kind="config")
enc_reinit_done = Cpt(EpicsSignalRO, suffix="enc_reinit_done_RBV", kind="config")
@@ -131,7 +131,7 @@ class Mo1BraggCrystal(Device):
xtal_enum = Cpt(EpicsSignalWithRBV, suffix="xtal_ENUM", kind="config")
d_spacing_si111 = Cpt(EpicsSignalWithRBV, suffix="d_spacing_si111", kind="config")
d_spacing_si311 = Cpt(EpicsSignalWithRBV, suffix="d_spacing_si311", kind="config")
set_offset = Cpt(EpicsSignal, suffix="set_offset.PROC", kind="config", put_complete=True)
set_offset = Cpt(EpicsSignal, suffix="set_offset", kind="config", put_complete=True)
current_xtal = Cpt(
EpicsSignalRO, suffix="current_xtal_ENUM_RBV", kind="normal", auto_monitor=True
)
@@ -171,6 +171,11 @@ class Mo1BraggScanSettings(Device):
a_scan_vel = Cpt(EpicsSignalWithRBV, suffix="a_scan_vel", kind="config", auto_monitor=True)
a_scan_time = Cpt(EpicsSignalWithRBV, suffix="a_scan_time", kind="config", auto_monitor=True)
class Mo1BraggCalculator(Device):
calc_reset = Cpt(EpicsSignal, suffix="calc_reset", kind="config", put_complete=True)
calc_done = Cpt(EpicsSignalRO, suffix="calc_done_RBV", kind="config")
calc_energy = Cpt(EpicsSignalWithRBV, suffix="calc_energy", kind="config")
calc_angle = Cpt(EpicsSignalWithRBV, suffix="calc_angle", kind="config")
class Mo1BraggScanControl(Device):
"""Mo1 Bragg PVs to control the scan after setting the parameters."""
@@ -179,15 +184,15 @@ class Mo1BraggScanControl(Device):
scan_duration = Cpt(
EpicsSignalWithRBV, suffix="scan_duration", kind="config", auto_monitor=True
)
scan_load = Cpt(EpicsSignal, suffix="scan_load.PROC", kind="config", put_complete=True)
scan_load = Cpt(EpicsSignal, suffix="scan_load", kind="config", put_complete=True)
scan_msg = Cpt(EpicsSignalRO, suffix="scan_msg_ENUM_RBV", kind="config", auto_monitor=True)
scan_start_infinite = Cpt(
EpicsSignal, suffix="scan_start_infinite.PROC", kind="config", put_complete=True
EpicsSignal, suffix="scan_start_infinite", kind="config", put_complete=True
)
scan_start_timer = Cpt(
EpicsSignal, suffix="scan_start_timer.PROC", kind="config", put_complete=True
EpicsSignal, suffix="scan_start_timer", kind="config", put_complete=True
)
scan_stop = Cpt(EpicsSignal, suffix="scan_stop.PROC", kind="config", put_complete=True)
scan_stop = Cpt(EpicsSignal, suffix="scan_stop", kind="config", put_complete=True)
scan_status = Cpt(
EpicsSignalRO, suffix="scan_status_ENUM_RBV", kind="config", auto_monitor=True
)
@@ -196,7 +201,7 @@ class Mo1BraggScanControl(Device):
)
scan_done = Cpt(EpicsSignalRO, suffix="scan_done_RBV", kind="config", auto_monitor=True)
scan_val_reset = Cpt(
EpicsSignal, suffix="scan_val_reset.PROC", kind="config", put_complete=True
EpicsSignal, suffix="scan_val_reset", kind="config", put_complete=True
)
scan_progress = Cpt(EpicsSignalRO, suffix="scan_progress_RBV", kind="config", auto_monitor=True)
scan_spectra_done = Cpt(
@@ -241,6 +246,7 @@ class Mo1Bragg(Device, PositionerBase):
crystal = Cpt(Mo1BraggCrystal, "")
encoder = Cpt(Mo1BraggEncoder, "")
scan_settings = Cpt(Mo1BraggScanSettings, "")
calculator = Cpt(Mo1BraggCalculator, "")
scan_control = Cpt(Mo1BraggScanControl, "")
status = Cpt(Mo1BraggStatus, "")
@@ -277,8 +283,8 @@ class Mo1Bragg(Device, PositionerBase):
)
# Execute motion
move_abs = Cpt(EpicsSignal, suffix="move_abs.PROC", kind="config", put_complete=True)
move_stop = Cpt(EpicsSignal, suffix="move_stop.PROC", kind="config", put_complete=True)
move_abs = Cpt(EpicsSignal, suffix="move_abs", kind="config", put_complete=True)
move_stop = Cpt(EpicsSignal, suffix="move_stop", kind="config", put_complete=True)
SUB_READBACK = "readback"
_default_sub = SUB_READBACK
@@ -528,85 +534,76 @@ class Mo1Bragg(Device, PositionerBase):
self.scan_settings.s_scan_angle_hi.put(high)
self.scan_settings.s_scan_scantime.put(scan_time)
@typechecked
def convert_angle_energy(self, mode:Literal["AngleToEnergy", "EnergyToAngle"], inp:float) -> float:
"""Calculate energy to angle or vice versa
Args:
mode (Literal["AngleToEnergy", "EnergyToAngle"]): Mode of calculation
input (float): Either angle or energy
Returns:
output (float): Converted angle or energy
"""
# TODO the calc_reset should reset from the IOC itself, check EPICS implementation with Alvin/Xiaoqiang
self.calculator.calc_reset.put(0)
self.calculator.calc_reset.put(1)
if not self.wait_for_signals(
signal_conditions=[(self.calculator.calc_done.get, 0)],
timeout=self.timeout_for_pvwait,
check_stopped=True,
):
raise TimeoutError(
f"Timeout after {self.timeout_for_pvwait} while waiting for calc done,"
)
if mode == "AngleToEnergy":
self.calculator.calc_angle.put(inp)
elif mode == "EnergyToAngle":
self.calculator.calc_energy.put(inp)
if not self.wait_for_signals(
signal_conditions=[(self.calculator.calc_done.get, 1)],
timeout=self.timeout_for_pvwait,
check_stopped=True,
):
raise TimeoutError(
f"Timeout after {self.timeout_for_pvwait} while waiting for calc done,"
)
time.sleep(0.25)
if mode == "AngleToEnergy":
return self.calculator.calc_energy.get()
elif mode == "EnergyToAngle":
return self.calculator.calc_angle.get()
def set_advanced_xas_settings(self, low: float, high:float, scan_time: float, p_kink: float, e_kink: float) -> None:
"""Set Advanced XAS parameters for upcoming scan.
Args:
low (float): Low angle value of the scan in deg
high (float): High angle value of the scan in deg
scan_time (float): Time for a half oscillation in s
p_kink (float): Position of kink in %
e_kink (float): Energy of kink in eV
Args:
"""
## TODO Change to energy only, calculation done on ACS controller
## ACS requires energy as input, outputs angle
## + reset bit signal, + calc done bit signal
## In the meantime, misuse s_scan_angle/energy PVs
## TODO Add fallback solution for automatic testing, otherwise test will fail
## because no monochromator will calculate the angle
## Unsure how to implement this
move_type = self.move_type.get()
if move_type == MoveType.ENERGY:
self.scan_settings.s_scan_energy_lo.put(e_kink)
time.sleep(1)
e_kink_deg = self.scan_settings.s_scan_angle_hi.get()
self.scan_settings.s_scan_energy_lo.put(low)
self.scan_settings.s_scan_energy_hi.put(high)
time.sleep(1)
low_deg = self.scan_settings.s_scan_angle_lo.get()
high_deg = self.scan_settings.s_scan_angle_hi.get()
e_kink_deg = self.convert_angle_energy(mode="EnergyToAngle", inp=e_kink)
# Angle and Energy are inverse proportional!
high_deg = self.convert_angle_energy(mode="EnergyToAngle", inp=low)
low_deg = self.convert_angle_energy(mode="EnergyToAngle", inp=high)
else:
raise Mo1BraggError("MoveType Angle not implemented for advanced scans, use Energy")
sf = 0.025 # safety factor to limit acceleration -> NEVER SET TO ZERO !
n = 41 # number of samples to generate -> Always choose uneven number, otherwise peak value will not be included
degree = 3 # degree of spline, 3 works good
tc = 0.0062 # time to be compensated each spline in s
pc = 0.02 # angle to add at both limits, must be same values as used on ACS controller for simple scans
pos, vel, dt = compute_spline(low_deg=low_deg, high_deg=high_deg, p_kink =p_kink, e_kink_deg = e_kink_deg, scan_time=scan_time)
# increase motion range slightly so that xas trigger signals will occur at defined energy limits
low_deg = low_deg - pc
high_deg = high_deg + pc
if p_kink < 0 or p_kink > 100:
raise Exception("Kink position not within range of [0..100%]")
if e_kink_deg < low_deg or e_kink_deg > high_deg:
raise Exception("Kink energy not within selected energy range of scan")
tc1 = sf / scan_time * tc
t_kink = (scan_time - tc - 2*(sf - tc1)) * p_kink/100 + (sf - tc1)
t_input = [0, sf - tc1, t_kink , scan_time - tc - sf + tc1, scan_time - tc ]
p_input = [0, 0 , e_kink_deg - low_deg , high_deg - low_deg , high_deg - low_deg]
cv = np.stack((t_input, p_input)).T # spline coefficients
max_param = len(cv) - degree
kv = np.clip(np.arange(len(cv)+degree+1)-degree,0,max_param) # knots
spl = BSpline(kv, cv, degree) # get spline function
p = spl(np.linspace(0,max_param,n))
v = spl(np.linspace(0,max_param,n), 1)
a = spl(np.linspace(0,max_param,n), 2)
j = spl(np.linspace(0,max_param,n), 3)
tim, pos = p.T
pos = pos + low_deg
vel = v[:,1]/v[:,0]
acc = []
for item in a:
acc.append(0) if item[1] == 0 else acc.append(item[1]/item[0])
jerk = []
for item in j:
jerk.append(0) if item[1] == 0 else jerk.append(item[1]/item[0])
dt = np.zeros(len(tim))
for i in np.arange(len(tim)):
if i == 0:
dt[i] = 0
else:
dt[i] = 1000*(tim[i]-tim[i-1])
self.scan_settings.a_scan_pos.put(pos)
self.scan_settings.a_scan_vel.put(vel)
self.scan_settings.a_scan_time.put(dt)
self.scan_settings.a_scan_pos.set(pos)
self.scan_settings.a_scan_vel.set(vel)
self.scan_settings.a_scan_time.set(dt)
def set_xrd_settings(
self,
debye_bec/devices/utils/__init__.py
View File
debye_bec/devices/utils/mo1_bragg_utils.py
+73
View File
@@ -0,0 +1,73 @@
import numpy as np
from scipy.interpolate import BSpline
################ Define Constants ############
SAFETY_FACTOR = 0.025 # safety factor to limit acceleration -> NEVER SET TO ZERO !
N_SAMPLES = 41 # number of samples to generate -> Always choose uneven number, otherwise peak value will not be included
DEGREE_SPLINE = 3 # DEGREE_SPLINE of spline, 3 works good
TIME_COMPENSATE_SPLINE = 0.0062 # time to be compensated each spline in s
POSITION_COMPONSATION = 0.02 # angle to add at both limits, must be same values as used on ACS controller for simple scans
class Mo1UtilsSplineError(Exception):
""" Exception for spline computation"""
def compute_spline(low_deg:float, high_deg:float, p_kink:float, e_kink_deg:float, scan_time:float) -> tuple[float, float, float]:
""" Spline computation for the advanced scan mode
Args:
low_deg (float): Low angle value of the scan in deg
high_deg (float): High angle value of the scan in deg
scan_time (float): Time for a half oscillation in s
p_kink (float): Position of kink in %
e_kink_deg (float): Position of kink in degree
Returns:
tuple[float,float,float] : Position, Velocity and delta T arrays for the spline
"""
# increase motion range slightly so that xas trigger signals will occur at defined energy limits
low_deg = low_deg - POSITION_COMPONSATION
high_deg = high_deg + POSITION_COMPONSATION
if p_kink < 0 or p_kink > 100:
raise Mo1UtilsSplineError(f"Kink position not within range of [0..100%] for p_kink: {p_kink}")
if e_kink_deg < low_deg or e_kink_deg > high_deg:
raise Mo1UtilsSplineError(f"Kink energy not within selected energy range of scan, for e_kink_deg {e_kink_deg}, low_deg {low_deg} and high_deg {high_deg}.")
tc1 = SAFETY_FACTOR / scan_time * TIME_COMPENSATE_SPLINE
t_kink = (scan_time - TIME_COMPENSATE_SPLINE - 2*(SAFETY_FACTOR - tc1)) * p_kink/100 + (SAFETY_FACTOR - tc1)
t_input = [0, SAFETY_FACTOR - tc1, t_kink , scan_time - TIME_COMPENSATE_SPLINE - SAFETY_FACTOR + tc1, scan_time - TIME_COMPENSATE_SPLINE ]
p_input = [0, 0 , e_kink_deg - low_deg , high_deg - low_deg , high_deg - low_deg]
cv = np.stack((t_input, p_input)).T # spline coefficients
max_param = len(cv) - DEGREE_SPLINE
kv = np.clip(np.arange(len(cv)+DEGREE_SPLINE+1)-DEGREE_SPLINE,0,max_param) # knots
spl = BSpline(kv, cv, DEGREE_SPLINE) # get spline function
p = spl(np.linspace(0,max_param,N_SAMPLES))
v = spl(np.linspace(0,max_param,N_SAMPLES), 1)
a = spl(np.linspace(0,max_param,N_SAMPLES), 2)
j = spl(np.linspace(0,max_param,N_SAMPLES), 3)
tim, pos = p.T
pos = pos + low_deg
vel = v[:,1]/v[:,0]
acc = []
for item in a:
acc.append(0) if item[1] == 0 else acc.append(item[1]/item[0])
jerk = []
for item in j:
jerk.append(0) if item[1] == 0 else jerk.append(item[1]/item[0])
dt = np.zeros(len(tim))
for i in np.arange(len(tim)):
if i == 0:
dt[i] = 0
else:
dt[i] = 1000*(tim[i]-tim[i-1])
return pos, vel, dt
debye_bec/scans/mono_bragg_scans.py
+3 -3
View File
@@ -89,7 +89,7 @@ class XASSimpleScan(AsyncFlyScanBase):
while True:
# Readout monitored devices
yield from self.stubs.read_and_wait(group="primary", wait_group="readout_primary")
yield from self.stubs.read_and_wait(group="primary", wait_group="readout_primary", point_id=self.point_id)
# Check if complete call on Mo1 Bragg has been finished
status = self.stubs.get_req_status(
device=self.motor, RID=self.metadata["RID"], DIID=target_diid
@@ -202,7 +202,7 @@ class XASAdvancedScan(XASSimpleScan):
motor (DeviceBase, optional): Motor device to be used for the scan. Defaults to "mo1_bragg".
Examples:
>>> scans.xas_advanced_scan(start=9, stop=11, scan_time=0.5, scan_duration=10, p_kink=0.15, e_kink=1)
>>> scans.xas_advanced_scan(start=10000, stop=12000, scan_time=0.5, scan_duration=10, p_kink=50, e_kink=10500)
"""
super().__init__(
start=start,
@@ -269,7 +269,7 @@ class XASAdvancedScanWithXRD(XASAdvancedScan):
motor (DeviceBase, optional): Motor device to be used for the scan. Defaults to "mo1_bragg".
Examples:
>>> scans.xas_advanced_scan_with_xrd(start=9, stop=11, scan_time=0.5, scan_duration=10, p_kink=0.15, e_kink=1, xrd_enable_low=True, num_trigger_low=5, cycle_low=2, exp_time_low=100, xrd_enable_high=False, num_trigger_high=3, cycle_high=1, exp_time_high=1000)
>>> scans.xas_advanced_scan_with_xrd(start=10000, stop=12000, scan_time=0.5, scan_duration=10, p_kink=50, e_kink=10500, xrd_enable_low=True, num_trigger_low=5, cycle_low=2, exp_time_low=100, xrd_enable_high=False, num_trigger_high=3, cycle_high=1, exp_time_high=1000)
"""
super().__init__(
start=start,