440 lines
14 KiB
Python
440 lines
14 KiB
Python
import numpy as np
|
|
from types import SimpleNamespace
|
|
from time import sleep
|
|
|
|
from epics import PV
|
|
from devices import *
|
|
|
|
from slic.core.acquisition import SFAcquisition
|
|
from slic.core.adjustable import Adjustable, PVAdjustable
|
|
from slic.core.device import Device, SimpleDevice
|
|
from slic.devices.general.motor import Motor
|
|
from slic.devices.general.delay_stage import DelayStage
|
|
from slic.devices.general.smaract import SmarActAxis
|
|
from slic.devices.general.shutter import Shutter
|
|
from slic.devices.xoptics.dcm import CoupledDoubleCrystalMonoEnergyWithTimeCorrection
|
|
from slic.core.sensor.bsmonitor import BSMonitor
|
|
|
|
from undulator import Undulators
|
|
|
|
ENERGY_OFFSET = -73.96501094395626 #-66.4807232884632 #-38.796532199490684 #-23.690679364204698 #-67.41724724743653 #-43.50130133659968 #-8.549246528231379 #-12.61393032963997 # -6.627670970333838 #-58.996124615499866 #set the mono-undulators energy offset here!
|
|
|
|
|
|
|
|
|
|
|
|
#s_diode_light = BSSensor("Diode", "SARES11-LSCP10-FNS:CH1:VAL_GET")
|
|
#s_i0 = BSSensor("I0", "SAROP11-PBPS122:INTENSITY")
|
|
#s_diode_dark = BSSensor("Diode", "SARES11-LSCP10-FNS:CH1:VAL_GET")
|
|
#s_i0 = BSSensor("I0", "SAROP11-PBPS122:INTENSITY")
|
|
#s = Norm("PP", s_diode, s_i0)
|
|
|
|
#s_bs = BSNorm("PP", "SARES11-LSCP10-FNS:CH1:VAL_GET", "SAROP11-PBPS122:INTENSITY")
|
|
|
|
|
|
class PP(BSMonitor):
|
|
|
|
def __init__(self,
|
|
ID = "PP",
|
|
evtmap = "SAR-CVME-TIFALL4:EvtSet",
|
|
diode = "SARES11-LSCP10-FNS:CH1:VAL_GET",
|
|
i0 = "SAROP11-PBPS122:INTENSITY",
|
|
**kwargs
|
|
):
|
|
self.cn_evtmap = evtmap
|
|
self.cn_diode = diode
|
|
self.cn_i0 = i0
|
|
chs = (self.cn_evtmap, self.cn_diode, self.cn_i0)
|
|
super().__init__(ID, chs, **kwargs)
|
|
self.cache_light = []
|
|
self.cache_dark = []
|
|
|
|
|
|
def get_aggregate(self):
|
|
return np.mean(self.cache_light) - np.mean(self.cache_dark)
|
|
|
|
|
|
def _unpack(self, data):
|
|
evtmap = data[self.cn_evtmap]
|
|
diode = data[self.cn_diode]
|
|
i0 = data[self.cn_i0]
|
|
|
|
value = diode / i0
|
|
cond_light = cond_check(evtmap)
|
|
if cond_light:
|
|
light = self.cache_light[-1]
|
|
dark = value
|
|
else:
|
|
dark = self.cache_dark[-1]
|
|
light = value
|
|
|
|
return light - dark
|
|
|
|
|
|
def _collect(self, data):
|
|
if data is None:
|
|
return
|
|
|
|
evtmap = data[self.cn_evtmap]
|
|
diode = data[self.cn_diode]
|
|
i0 = data[self.cn_i0]
|
|
|
|
value = diode / i0
|
|
cond_light = cond_check(evtmap)
|
|
if cond_light:
|
|
self.cache_light.append(value)
|
|
else:
|
|
self.cache_dark.append(value)
|
|
|
|
|
|
def cond_check(self, evtmap):
|
|
fel = evtmap[13]
|
|
laser = evtmap[18]
|
|
darkshot = evtmap[21]
|
|
return np.logical_and.reduce((fel, laser, np.logical_not(darkshot)))
|
|
|
|
|
|
|
|
def _clear(self):
|
|
self.cache_light.clear()
|
|
self.cache_dark.clear()
|
|
|
|
|
|
#pp = PP()
|
|
|
|
laser_pitch = SmarActAxis("SARES11-XICM125:ROX1")
|
|
laser_trans = SmarActAxis("SARES11-XICM125:TRX1")
|
|
laser_yaw = SmarActAxis("SARES11-XICM125:ROY1")
|
|
microscope_pitch = SmarActAxis("SARES11-XMI125:ROY1")
|
|
microscope_yaw = SmarActAxis("SARES11-XMI125:ROZ1")
|
|
#laser_mod = PVAdjustable("SIN-TIMAST-TMA:Evt-23-Off-SP", process_time=1, name="Laser Mod")
|
|
|
|
|
|
CDCMEWTC = CoupledDoubleCrystalMonoEnergyWithTimeCorrection(limit_low=2450, limit_high=2520)
|
|
|
|
PSSS = Motor("SARFE10-PSSS059:MOTOR_Y3", name="PSSS XTAL y")
|
|
|
|
#opo_delay = PVAdjustable("SLAAR03-LTIM-PDLY:DELAY", name="OPO Delay")
|
|
XrayShutter = Shutter("SARFE10-OPSH059")
|
|
|
|
|
|
import requests
|
|
|
|
class UndulatorEnergy(PVAdjustable):
|
|
|
|
def __init__(self, process_time=1):
|
|
|
|
self.process_time = process_time
|
|
super().__init__("SARUN:USER-DELTA", "SARUN07-UIND030:FELPHOTENE", process_time=0.2) # process_time here is only to insert the delta into the panel
|
|
# assert self.units == "keV"
|
|
print('units={}'.format(self.units))
|
|
self.units = "eV"
|
|
|
|
def set_target_value(self, value):
|
|
#value /= 1000
|
|
current = self.get_current_value()
|
|
delta = (value - current)
|
|
print(f"delta: {value} - {current} = {delta}")
|
|
delta /= 1000
|
|
task = super().set_target_value(delta)
|
|
sleep(0.1)
|
|
# task.wait() # make sure the delta is set correctly
|
|
#TODO: the following should be a task?
|
|
url_go = "http://sf-daq-mgmt.psi.ch:8090/run/Undulators/K_AR_scale"
|
|
reply = requests.get(url_go)#.json()
|
|
print('Pressing "GO" produced:', reply)
|
|
sleep(self.process_time) # wait for undulator change
|
|
return task
|
|
|
|
def get_current_value(self):
|
|
value = super().get_current_value() - 0.02037
|
|
value *= 1000
|
|
return value
|
|
|
|
def stop(self):
|
|
url_abort = "http://sf-daq-mgmt.psi.ch:8090/abort"
|
|
reply = requests.get(url_abort)#.json()
|
|
print('Pressing "Abort" produced:', reply)
|
|
|
|
|
|
|
|
|
|
|
|
class UndulatorCoupledDoubleCrystalMonoEnergy(Adjustable):
|
|
|
|
def __init__(self, ID, name=None, process_time=1):
|
|
# self.und = UndulatorEnergy(process_time=process_time)
|
|
self.und = Undulators(energy_offset=ENERGY_OFFSET, n_und_ref=7)
|
|
|
|
pvname_setvalue = "SAROP11-ARAMIS:ENERGY_SP"
|
|
pvname_readback = "SAROP11-ARAMIS:ENERGY"
|
|
pvname_moving = "SAROP11-ODCM105:MOVING"
|
|
|
|
# This is the WRONG coupling, we need to make sure it is disabled!
|
|
pvname_ebeam_coupling = "SGE-OP2E-ARAMIS:MODE_SP"
|
|
|
|
pv_setvalue = PV(pvname_setvalue)
|
|
pv_readback = PV(pvname_readback)
|
|
pv_moving = PV(pvname_moving)
|
|
pv_ebeam_coupling = PV(pvname_ebeam_coupling)
|
|
|
|
units = pv_readback.units
|
|
super().__init__(ID, name=name, units=units)
|
|
|
|
self.pvnames = SimpleNamespace(
|
|
setvalue = pvname_setvalue,
|
|
readback = pvname_readback,
|
|
moving = pvname_moving,
|
|
ebeam_coupling = pvname_ebeam_coupling
|
|
)
|
|
|
|
self.pvs = SimpleNamespace(
|
|
setvalue = pv_setvalue,
|
|
readback = pv_readback,
|
|
moving = pv_moving,
|
|
ebeam_coupling = pv_ebeam_coupling
|
|
)
|
|
|
|
|
|
def get_current_value(self):
|
|
return self.pvs.readback.get()
|
|
|
|
def set_target_value(self, value, hold=False):
|
|
changer = lambda: self.move_and_wait(value)
|
|
return self._as_task(changer, hold=hold, stopper=self.stop)
|
|
|
|
def move_and_wait(self, value, wait_time=0.1):
|
|
# self.disable_ebeam_coupling()
|
|
self.pvs.setvalue.put(value)
|
|
self.und.set_target_value(value)
|
|
sleep(1) #TODO: is this correct?
|
|
while self.is_moving():
|
|
print("mono is moving")
|
|
sleep(wait_time)
|
|
|
|
def is_moving(self):
|
|
moving = self.pvs.moving.get()
|
|
return bool(moving)
|
|
|
|
# def enable_ebeam_coupling(self):
|
|
# self.pvs.ebeam_coupling.put(1)
|
|
|
|
# def disable_ebeam_coupling(self):
|
|
# self.pvs.ebeam_coupling.put(0)
|
|
|
|
# @property
|
|
# def ebeam_coupling(self):
|
|
# return self.pvs.ebeam_coupling.get(as_string=True)
|
|
|
|
|
|
|
|
|
|
|
|
class UndulatorCoupledDoubleCrystalMonoEnergyWithTimeCorrection(Adjustable):
|
|
|
|
def __init__(self, ID="UCDCMEWTC", name="Alvra DCM Undulator-coupled energy with time correction", limit_low=2800, limit_high=3050, process_time=1):
|
|
# self.und = UndulatorEnergy(process_time=process_time)
|
|
# self.und = Undulators(energy_offset=ENERGY_OFFSET)
|
|
self.und = Undulators(energy_offset=ENERGY_OFFSET, n_und_ref=7)
|
|
|
|
self.wait_time = 0.1
|
|
|
|
pvname_setvalue = "SAROP11-ARAMIS:ENERGY_SP"
|
|
pvname_readback = "SAROP11-ARAMIS:ENERGY"
|
|
pvname_moving = "SAROP11-ODCM105:MOVING"
|
|
|
|
# This is the WRONG coupling, we need to make sure it is disabled!
|
|
pvname_ebeam_coupling = "SGE-OP2E-ARAMIS:MODE_SP"
|
|
|
|
pv_setvalue = PV(pvname_setvalue)
|
|
pv_readback = PV(pvname_readback)
|
|
pv_moving = PV(pvname_moving)
|
|
pv_ebeam_coupling = PV(pvname_ebeam_coupling)
|
|
|
|
self.timing = Motor("SLAAR11-LMOT-M452:MOTOR_1")
|
|
# self.electron_energy_rb = PV("SARCL02-MBND100:P-READ")
|
|
# self.electron_energy_sv = PV("SGE-OP2E-ARAMIS:E_ENERGY_SP")
|
|
|
|
units = pv_readback.units
|
|
super().__init__(ID, name=name, units=units)
|
|
|
|
self.limit_low = limit_low
|
|
self.limit_high = limit_high
|
|
|
|
self.pvnames = SimpleNamespace(
|
|
setvalue = pvname_setvalue,
|
|
readback = pvname_readback,
|
|
moving = pvname_moving,
|
|
ebeam_coupling = pvname_ebeam_coupling
|
|
)
|
|
|
|
self.pvs = SimpleNamespace(
|
|
setvalue = pv_setvalue,
|
|
readback = pv_readback,
|
|
moving = pv_moving,
|
|
ebeam_coupling = pv_ebeam_coupling
|
|
)
|
|
|
|
|
|
def get_current_value(self):
|
|
return self.pvs.readback.get()
|
|
|
|
def set_target_value(self, value, hold=False):
|
|
ll = self.limit_low
|
|
if ll is not None:
|
|
if value < ll:
|
|
msg = f"requested value is outside the allowed range: {value} < {ll}"
|
|
print(msg)
|
|
raise KeyboardInterrupt(msg)
|
|
lh = self.limit_high
|
|
if lh is not None:
|
|
if value > lh:
|
|
msg = f"requested value is outside the allowed range: {value} > {lh}"
|
|
print(msg)
|
|
raise KeyboardInterrupt(msg)
|
|
|
|
wait_time = self.wait_time
|
|
|
|
self.disable_ebeam_coupling()
|
|
|
|
current_energy = self.get_current_value()
|
|
delta_energy = value - current_energy
|
|
#delta_energy = value
|
|
|
|
timing = self.timing
|
|
current_delay = timing.get_current_value()
|
|
delta_delay = convert_E_to_distance(delta_energy)
|
|
target_delay = current_delay + delta_delay
|
|
#target_delay = delta_delay
|
|
|
|
print(f"Energy = {current_energy} -> delta = {delta_energy} -> {value}")
|
|
print(f"Delay = {current_delay} -> delta = {delta_delay} -> {target_delay}")
|
|
|
|
timing.set_target_value(target_delay).wait()
|
|
|
|
self.move_and_wait(value)
|
|
sleep(1) # wait so that the set value has changed
|
|
|
|
print("start waiting for DCM")
|
|
while self.is_moving(): #TODO: moving PV seems broken
|
|
print("mono is moving")
|
|
sleep(wait_time)
|
|
# print("start waiting for electron beam")
|
|
# while abs(self.electron_energy_rb.get() - self.electron_energy_sv.get()) > 0.25:
|
|
# sleep(wait_time)
|
|
sleep(wait_time)
|
|
|
|
|
|
def move_and_wait(self, value, wait_time=0.1):
|
|
self.disable_ebeam_coupling()
|
|
self.pvs.setvalue.put(value)
|
|
self.und.set_target_value(value)
|
|
sleep(1) #TODO: is this correct?
|
|
while self.is_moving():
|
|
print("mono is moving")
|
|
sleep(wait_time)
|
|
|
|
|
|
def is_moving(self):
|
|
moving = self.pvs.moving.get()
|
|
return bool(moving)
|
|
|
|
# def enable_ebeam_coupling(self):
|
|
# self.pvs.ebeam_coupling.put(1)
|
|
|
|
def disable_ebeam_coupling(self):
|
|
self.pvs.ebeam_coupling.put(0)
|
|
|
|
# @property
|
|
# def ebeam_coupling(self):
|
|
# return self.pvs.ebeam_coupling.get(as_string=True)
|
|
|
|
|
|
|
|
# adjust this function!!
|
|
# be careful, CDCMEWTC has its own conversion!!!
|
|
def convert_E_to_distance(E): ### returns a value in mm!
|
|
#return 0.0061869 * E ### calibration May 2021, S k-edge, 2450 eV, corresponds to 41.2 fs/eV --> this is with InSb111!!
|
|
|
|
#return 0.00524244 * E ### calibration Nov 2021, Rh L-edge, 3000 eV, corresponds to 35 fs/eV at 20 mm gap
|
|
#return 0.0005445 * E ### calibration Mar 2022, Mn K-edge, 6500 eV, corresponds to 3.6 fs/eV at 20 mm gap
|
|
#return 0.00029134 * E ### calibration Nov 2023, Mn K-edge, 6500 eV, corresponds to 1.9 fs/eV at 10 mm gap
|
|
#return 0.0004692 * E ### calibration Mar 2022, Fe K-edge, 7100 eV, corresponds to 3.13 fs/eV at 20 mm gap
|
|
#return 0.0148402 * E ### calibration May 2022, S K-edge, 2470 eV, corresponds to 99 fs/eV at 20 mm gap
|
|
return 0.00021 * E ### calibration Oct 2023, Ni K-edge, 8350 eV, corresponds to 1.4 fs/eV at 10 mm gap
|
|
|
|
#return 0.0067003 * E ### calibration Jan 2023, Cl K-edge / Ru L3-edge, 2830 eV, corresponds to 45 fs/eV at 20 mm gap
|
|
#return 0.00495847 * E ### calibration Jan 2023, Rh L-edge, 3000 eV, corresponds to 33 fs/eV, very similar to Nov '21
|
|
#return 0.0147342 * E ### calibration Oct 2024, S K-edge, 2470 eV, corresponds to 98 fs/eV at 20 mm gap
|
|
|
|
#params = [-3.62574226e-04, 1.62862011e+00, -1.80030011e+03] ### calibration May 2024, P K-edge, 2150 eV, corresponds to ~500 fs/eV
|
|
#p = np.poly1d(params)
|
|
#return p(E)
|
|
|
|
|
|
|
|
#und = UndulatorEnergy()
|
|
mono_und = UndulatorCoupledDoubleCrystalMonoEnergy("MONO_UND", name="Alvra DCM Undulator-coupled energy")
|
|
UCDCMEWTC = UndulatorCoupledDoubleCrystalMonoEnergyWithTimeCorrection(limit_low=2450, limit_high=2550)
|
|
|
|
|
|
|
|
|
|
|
|
class TXS(Device):
|
|
|
|
def __init__(self, ID="SAROP11-PTXS128", name="TXS"):
|
|
super().__init__(ID, name=name)
|
|
|
|
self.x = Motor(ID + ":TRX1")
|
|
self.y = Motor(ID + ":TRY1")
|
|
self.z = Motor(ID + ":TRZ1")
|
|
|
|
self.crystals = SimpleDevice("TXS Crystals",
|
|
left = SimpleDevice("Left Crystals",
|
|
linear = SmarActAxis(ID + ":TRX11"),
|
|
rotary = SmarActAxis(ID + ":ROTY11"),
|
|
gonio1 = SmarActAxis(ID + ":ROTX11"),
|
|
gonio2 = SmarActAxis(ID + ":ROTX12"),
|
|
gonio3 = SmarActAxis(ID + ":ROTX13")
|
|
),
|
|
right = SimpleDevice("Right Crystals",
|
|
linear = SmarActAxis(ID + ":TRX21"),
|
|
rotary = SmarActAxis(ID + ":ROTY21"),
|
|
gonio1 = SmarActAxis(ID + ":ROTX21"),
|
|
gonio2 = SmarActAxis(ID + ":ROTX22"),
|
|
gonio3 = SmarActAxis(ID + ":ROTX23")
|
|
)
|
|
)
|
|
|
|
|
|
|
|
from colorama import Fore, Style
|
|
|
|
|
|
class Deprecator:
|
|
|
|
def __init__(self, old_name, new_name):
|
|
self._old_name = old_name
|
|
self._new_name = new_name
|
|
|
|
def __repr__(self):
|
|
return ""
|
|
|
|
def __getattr__(self, name):
|
|
print(
|
|
Fore.RED +
|
|
f"\"{self._old_name}\" has been renamed \"{self._new_name}\""
|
|
+ Style.RESET_ALL
|
|
)
|
|
return lambda *args, **kwargs: self._method(self._new_name, name, *args, **kwargs)
|
|
|
|
def _method(self, new_name, meth_name, *args, **kwargs):
|
|
args = [repr(v) for v in args]
|
|
args += [f"{k}={repr(v)}" for k, v in kwargs.items()]
|
|
args = ", ".join(args)
|
|
print(f"please use the following command instead:\n {new_name}.{meth_name}({args})")
|
|
|
|
|
|
|