diff --git a/slic/devices/timing/lasertiming.py b/slic/devices/timing/lasertiming.py index afbf9278..99ac35a1 100644 --- a/slic/devices/timing/lasertiming.py +++ b/slic/devices/timing/lasertiming.py @@ -1,15 +1,14 @@ import os -from time import sleep +from time import sleep, asctime from types import SimpleNamespace import numpy as np from epics import PV -from slic.core.task import Task from slic.core.adjustable import Adjustable +from slic.devices.general.motor import check_pos_type +from slic.utils import typename -_posTypes = ["user", "dial", "raw"] - _OSCILLATOR_PERIOD = 1 / 71.368704e6 _POCKELS_CELL_RESOLUTION = 7e-9 @@ -43,7 +42,7 @@ sdg_set = sdg + "-SP" class ETiming(Adjustable): - def __init__(self, Id, name="Globi laser electronic timing", units="ps"): + def __init__(self, Id, name="Globi Laser Electronic Timing", units="ps"): super().__init__(name=name, units=units) self.Id = Id @@ -58,8 +57,8 @@ class ETiming(Adjustable): return self.pvs.readback.get() * 1e6 # convert from us to ps def set_target_value(self, value, hold=False): - changer = lambda: self.put_and_wait(value) - return self._as_task(changer, hold=hold) + change = lambda: self.put_and_wait(value) + return self._as_task(change, hold=hold) def put_and_wait(self, value, checktime=0.01): self.pvs.setvalue.put(value) @@ -71,14 +70,14 @@ class ETiming(Adjustable): waiting = self.pvs.waiting.get() return bool(waiting) -# def wait_for_valid_value(self): -# tval = np.nan -# while not np.isfinite(tval): -# tval = self.get_current_value() -# return tval + def wait_for_valid_value(self): + tval = np.nan + while not np.isfinite(tval): + tval = self.get_current_value() + return tval -# def set_current_value(self, value): -# self.pvs.setvalue.put(value) + def set_current_value(self, value): + self.pvs.setvalue.put(value) def __repr__(self): res = super().__repr__() @@ -89,260 +88,297 @@ class ETiming(Adjustable): -class LXT: +class LXT(Adjustable): - def __init__(self, accuracy_poly=[100e-15, 1e-7]): - self.sdg1 = PockelsTrigger(sdg_get, sdg_set, sdg_off) - self.slicer_gate = PockelsTrigger(sg_get, sg_set, sg_off) - self.phase_shifter = PhaseShifter(PS) - self.Id = PS - self.name = "lxt" - self.accuracy_poly = accuracy_poly + def __init__(self, Id, tolerance_poly_coeff=(100e-15, 1e-7), name="Laser X-ray Timing", units="s"): + super().__init__(name=name, units=units) + self.Id = Id + self.tolerance_poly_coeff = tolerance_poly_coeff - def move_sdg(self, value): - self.sdg1.move(value) + self.devices = SimpleNamespace( + phase_shifter = PhaseShifter(PS), + sdg1 = PockelsTrigger(sdg_get, sdg_set, sdg_off), #TODO: naming + slicer_gate = PockelsTrigger(sg_get, sg_set, sg_off) + ) - def move(self, value, accuracy=None): - self.sdg1.move(-value) - self.slicer_gate.move(-value) - if not accuracy: - accuracy = np.abs(value) * self.accuracy_poly[1] + self.accuracy_poly[0] - self.phase_shifter.move(value, accuracy=accuracy) - def set(self, value): - self.phase_shifter.set(value) - self.slicer_gate.set(-value) - self.sdg1.set(-value) - - def get(self): - # pulses are at SOME_IDX*OSCILLATOR_PERIOD-PHASESHITER - # the -PHASESHITER is due to the inverted sign - phase_shifter = self.phase_shifter.get() - sdg1_delay = self.sdg1.get() - - idx_pulse = (sdg1_delay + phase_shifter) / _OSCILLATOR_PERIOD - - delay = int(idx_pulse) * _OSCILLATOR_PERIOD - phase_shifter + def get_current_value(self): + # pulses are at: INDEX * OSCILLATOR_PERIOD - PHASE_SHIFTER + # the -PHASE_SHIFTER is due to the inverted sign + phase_shifter = self.devices.phase_shifter.get() + sdg1 = self.devices.sdg1.get() + index = (phase_shifter + sdg1) // _OSCILLATOR_PERIOD + delay = index * _OSCILLATOR_PERIOD - phase_shifter return -delay def set_target_value(self, value, hold=False): - changer = lambda: self.move(value) - return Task(changer, hold=hold) + change = lambda: self.move(value) + return self._as_task(change, hold=hold) - def get_current_value(self): - return self.get() + def is_moving(self): + raise NotImplementedError + + def move(self, value, tolerance=None): + self.devices.sdg1.move(-value) + self.devices.slicer_gate.move(-value) + if tolerance is None: + slope, const = self.tolerance_poly_coeff + tolerance = np.abs(value) * slope + const + self.devices.phase_shifter.move(value, tolerance=tolerance) + + def move_sdg(self, value): + self.devices.sdg1.move(value) def set_current_value(self, value): - self.set(value) + self.devices.phase_shifter.set(value) + self.devices.sdg1.set(-value) + self.devices.slicer_gate.set(-value) def __repr__(self): - delay = nice_time_to_str(self.get()) - return "delay = %s" % (delay) + res = super().__repr__() + delay = self.get_current_value() + delay = nice_time_to_str(delay) + res += f" ({delay})" + return res -class PhaseShifterAramis: +class PhaseShifterAramis(Adjustable): def __init__(self, Id, name=None): + super().__init__(name=name) self.Id = Id - self._pshifter = PhaseShifter(Id) - self.name = name + self._phase_shifter = PhaseShifter(Id) - def set_target_value(self, value, hold=False, check=True): - mover = lambda: self._pshifter.move(value) - return Task(mover, hold=hold) + def set_target_value(self, value, hold=False): + change = lambda: self._phase_shifter.move(value) + return self._as_task(change, hold=hold) - def stop(self): - pass + def get_current_value(self, pos_type="user"): + check_pos_type(pos_type, {"user", "dial"}) + if pos_type == "user": + return self._phase_shifter.get() + if pos_type == "dial": + return self._phase_shifter.get_dial() - def get_current_value(self, posType="user", readback=True): - _keywordChecker([("posType", posType, _posTypes)]) - if posType == "user": - return self._pshifter.get() - if posType == "dial": - return self._pshifter.get_dial() + def is_moving(self): + raise NotImplementedError - def set_current_value(self, value, posType="user"): - _keywordChecker([("posType", posType, _posTypes)]) - if posType == "user": - return self._motor.set(value) + def set_current_value(self, value, pos_type="user"): + check_pos_type(pos_type, {"user"}) + if pos_type == "user": + return self._phase_shifter.set(value) -class PhaseShifter(PV): +class PhaseShifter: - def __init__(self, pv_basename=PS, dial_max=14.0056e-9, precision=100e-15): - pvname = pv_basename + ":CURR_DELTA_T" - PV.__init__(self, pvname) - self._filename = os.path.join(_basefolder, pvname) - self._pv_setvalue = PV(pv_basename + ":NEW_DELTA_T") - self._pv_execute = PV(pv_basename + ":SET_NEW_PHASE.PROC") - self._storage = Storage(pvname) + def __init__(self, pv_basename=PS, dial_max=14.0056e-9, tolerance=100e-15): + self.pv_basename = pv_basename self.dial_max = dial_max - self.retry = precision + self.tolerance = tolerance + + pvname = pv_basename + ":CURR_DELTA_T" #TODO: should this be the basename only? actually stores the offset! + self.storage = Storage(pvname) + self.filename = self.storage.filename + + self.pvs = SimpleNamespace( + setvalue = PV(pv_basename + ":NEW_DELTA_T"), + readback = PV(pv_basename + ":CURR_DELTA_T"), + execute = PV(pv_basename + ":SET_NEW_PHASE.PROC") + ) + + + def get(self): + return self.get_dial() - self.offset # offset in s + + def get_dial(self): + return self.pvs.readback.get() * 1e-12 # convert from ps to s @property def offset(self): - return self._storage.value - - def get_dial(self): - return super().get() * 1e-12 - - def get(self): - return self.get_dial() - self.offset - - def store(self, value=None): - if value == None: - value = self.get_dial() - self._storage.store(value) - - def move(self, value, accuracy=None): - if accuracy is None: - accuracy = self.retry - dial = value + self.offset - dial = np.mod(dial, _OSCILLATOR_PERIOD) - if dial > self.dial_max: - dial = self.dial_max - dial_ps = dial * 1e12 - self._pv_setvalue.put(dial_ps) - time.sleep(0.1) - self._pv_execute.put(1) - #print(accuracy) - while np.abs(self.get_dial() - dial) > accuracy: - #print(np.abs(self.get_dial()-dial)) - time.sleep(0.2) + return self.storage.value def set(self, value): - newoffset = self.get_dial() - value - newoffset = np.mod(newoffset, _OSCILLATOR_PERIOD) - self.store(newoffset) + new_offset = self.get_dial() - value + new_offset %= _OSCILLATOR_PERIOD + self.store(new_offset) + + def store(self, value=None): + if value is None: + value = self.get_dial() + self.storage.store(value) + + + def move(self, value, tolerance=None): # tolerance in s + dial = value + self.offset + dial %= _OSCILLATOR_PERIOD + dial = min(dial, self.dial_max) + dial_ps = dial * 1e12 + self.pvs.setvalue.put(dial_ps) + sleep(0.1) + self.pvs.execute.put(1) + + if tolerance is None: + tolerance = self.tolerance + + while abs(self.get_dial() - dial) > tolerance: + #print(abs(self.get_dial() - dial), tolerance) + sleep(0.2) + def __repr__(self): - dial = time_to_str(self.get_dial(), n=15) - user = time_to_str(self.get(), n=15) - return "Phase Shifter: user,dial = %s , %s" % (user, dial) + tname = typename(self) + name = self.pv_basename + n = 15 + user = time_to_str(self.get(), n=n) + dial = time_to_str(self.get_dial(), n=n) + units = "sec" + return f"{tname} \"{name}\" at {user} {units} (dial at {dial} {units})" -class PockelsTrigger(PV): +class PockelsTrigger: - def __init__(self, pv_get, pv_set, pv_offset_get): #TODO make offset optional - pvname = pv_get - PV.__init__(self, pvname) - self._pv_offset_get = PV(pv_offset_get) - self._pv_setvalue = PV(pv_set) - self._filename = os.path.join(_basefolder, pvname) - self._storage = Storage(pvname) + def __init__(self, pv_get, pv_set, pv_offset): #TODO make offset optional? + self.pvname = pvname = pv_get + self.storage = Storage(pvname) + self.filename = self.storage.filename - @property - def offset(self): - return self._storage.value + self.pvs = SimpleNamespace( + setvalue = PV(pv_set), + readback = PV(pv_get), + offset = PV(pv_offset) + ) - def get_dial(self): - return np.round(super().get() * 1e-6, 9) + self._pv_offset_get.get() * 1e-9 - 7.41e-9 #TODO what is the constant? def get(self): return self.get_dial() - self.offset + def get_dial(self): + readback = self.pvs.readback.get() * 1e-6 + readback = round(readback, 9) #TODO: why? + offset = self.pvs.offset.get() * 1e-9 - 7.41e-9 #TODO what is the constant? + return readback + offset + + @property + def offset(self): + return self.storage.value + + def set(self, value): + new_offset = self.get_dial() - value + self.store(new_offset) + def store(self, value=None): - if value == None: + if value is None: value = self.get_dial() - self._storage.store(value) + self.storage.store(value) def move(self, value): dial = value + self.offset - self._pv_setvalue.put(dial * 1e6) - - def set(self, value): - newoffset = self.get_dial() - value - self.store(newoffset) + self.pvs.setvalue.put(dial * 1e6) def __repr__(self): - dial = time_to_str(self.get_dial(), n=12) - user = time_to_str(self.get(), n=12) - return "Pockel Trigger PV: %s user , dial = %s, %s" % (self.pvname, user, dial) + tname = typename(self) + name = self.pvname + n = 12 + user = time_to_str(self.get(), n=n) + dial = time_to_str(self.get_dial(), n=n) + units = "sec" + return f"{tname} \"{name}\" at {user} {units} (dial at {dial} {units})" class Storage: - """ this class is needed to store the offset in files and read in s """ + """ + Read/write a value from/to file + """ def __init__(self, pvname): - self._filename = os.path.join(_basefolder, pvname) self.pvname = pvname - self.last_read_time = -1 + self.filename = os.path.join(_basefolder, pvname) + self.last_value = None + self.last_read_time = None - @property - def last_modified_time(self): - if os.path.isfile(self._filename): - return os.stat(self._filename).st_mtime - else: - return -1 @property def value(self): - lmod = self.last_modified_time - if os.path.isfile(self._filename): - # need to read again ? - if self.last_read_time == -1 or lmod > self.last_read_time: - #print("actually reading") - value = float(np.loadtxt(self._filename)) - self.last_read_time = lmod - self.last_read = value - else: - value = self.last_read - else: - print("could not read", self._filename) - value = 0 - return value + if not self.exists: + fname = self.filename + print(f"Warning: Storage file \"{fname}\" does not exist") #TODO: raise an error? + return None + + tlread = self.last_read_time + tlmod = self.last_modified_time + if tlread is None or tlmod > tlread: + self.last_value = self.load() + self.last_read_time = tlmod + + return self.last_value + + + @property + def exists(self): + return os.path.isfile(self.filename) + + @property + def last_modified_time(self): + if self.exists: + return os.stat(self.filename).st_mtime def store(self, value): - with open(self._filename, "w") as f: - f.write("# %s\n" % time.asctime()) - f.write("%.15f" % value) + timestamp = asctime() + with open(self.filename, "w") as f: #TODO: use numpy here as well? + f.write(f"# {timestamp}\n") + f.write(f"{value}\n") + + def load(self): + value = np.loadtxt(self.filename) #TODO: parse manually here as well? + return float(value) -def _keywordChecker(kw_key_list_tups): - for tkw, tkey, tlist in kw_key_list_tups: - assert tkey in tlist, "Keyword %s should be one of %s" % (tkw, tlist) - - - -def time_to_str(value, n=12): +def time_to_str(time, n=12): fmt = "%%+.%df" % n - value = fmt % value - #print(value) - idx_point = value.find(".") - ret_str = value[:idx_point] + " ." - ngroups = (len(value) - idx_point) // 3 - for n in range(ngroups): - ret_str += " %s" % value[idx_point + 1 + 3 * n : idx_point + 1 + 3 * (n + 1)] - #print(idx_point+1+3*n,idx_point+1*3*(n-1),ret_str) + time = fmt % time + #print(time) + idx_point = time.find(".") + ret_str = time[:idx_point] + " ." + ngroups = (len(time) - idx_point) // 3 + for ng in range(ngroups): + ret_str += " %s" % time[idx_point + 1 + 3 * ng : idx_point + 1 + 3 * (ng + 1)] + #print(idx_point + 1 + 3 * ng, idx_point + 1 * 3 * (ng - 1), ret_str) return ret_str -def nice_time_to_str(delay, fmt="%+.0f"): - a_delay = abs(delay) - if a_delay >= 1: - ret = fmt % delay + "s" - elif 1e-3 <= a_delay < 1: - ret = fmt % (delay * 1e3) + "ms" - elif 1e-6 <= a_delay < 1e-3: - ret = fmt % (delay * 1e6) + "us" - elif 1e-9 <= a_delay < 1e-6: - ret = fmt % (delay * 1e9) + "ns" - elif 1e-12 <= a_delay < 1e-9: - ret = fmt % (delay * 1e12) + "ps" - elif 1e-15 <= a_delay < 1e-12: - ret = fmt % (delay * 1e12) + "fs" - elif 1e-18 <= a_delay < 1e-15: - ret = fmt % (delay * 1e12) + "as" - elif a_delay < 1e-18: - ret = "0s" +def nice_time_to_str(time, fmt="%+.0f"): + abs_time = abs(time) + if abs_time >= 1: + factor = 1 + units = "s" + elif 1e-3 <= abs_time < 1: + factor = 1e3 + units = "ms" + elif 1e-6 <= abs_time < 1e-3: + factor = 1e6 + units = "us" + elif 1e-9 <= abs_time < 1e-6: + factor = 1e9 + units = "ns" + elif 1e-12 <= abs_time < 1e-9: + factor = 1e12 + units = "ps" + elif 1e-15 <= abs_time < 1e-12: + factor = 1e15 + units = "fs" + elif 1e-18 <= abs_time < 1e-15: + factor = 1e18 + units = "as" + elif abs_time < 1e-18: + return "0 s" else: - ret = str(delay) + "s" - return ret + return str(time) + return fmt % (time * factor) + " " + units