"""
@author: Matthias Quintern
"""
from devctrl.devices import LockInAmp, Amplifier, Monochromator, Shutter
from devctrl.utility.time import duration_to_string
from devctrl.utility.threading import FakeThread, ThreadWithExceptionSupport
from ..data.prsdata import PrsData
from ..data.functions import eV_to_nm, bandwidth_eV_to_nm, nm_to_eV
from .settings import PrsSettings, LockInSettings, MeasurementSettings
from mqutil.dict_util import get_subdict
import time
import datetime
from multiprocessing import Queue
import numpy as np
import threading
import logging
log = logging.getLogger(__name__)
import pyvisa
[docs]
def get_time():
return datetime.datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
[docs]
class LockInHandler:
"""
Wrapper class for a lock-in amplifier, that abstracts some model-specific things like setting up a buffer,
where the number of quantities, max. number of values and the available sample rates depend on the model.
"""
def __init__(self, mode: str, lockin: LockInAmp, lockin_settings: LockInSettings):
self.mode:str = mode
self.lockin:LockInAmp = lockin
self.lockin_settings: LockInSettings = lockin_settings
# the lock-in aux port where the DC signal is connected to
log.debug(f"[mode={mode}]: Set aux_DC={self.lockin_settings.aux_DC} for lockin {lockin}")
# which lock-in type is used
# use strings to avoid having to import the modules, which might not be available
self.SR830 = "SR830" in str(lockin)
self.Model7260 = "7260" in str(lockin)
self.n_try = 2 # how often to try to recover from a communication error
self.sample_rate_Hz = -1
self.n_bins = -1 # number of data points to capture in the buffer
# use h_ prefix (for handler) to avoid any name conflicts with LockInAmp class
[docs]
def set_measurement_time(self, measurement_time_s, target_sample_rate_Hz=None):
"""
Convert the target sample rate to one that is compatible with the device.
Also calculates the number of values to be recorded for a given measurement time
:param target_sample_rate_Hz: targeted sample rate, or 'max' to use the maximum for the given measurement time
"""
n_vals = 2 if self.SR830 else 3 # R, Aux for SR830, R, theta, aux else
max_len = self.lockin.buffer_get_max_size(n_vals)
if not target_sample_rate_Hz:
target_sample_rate_Hz = self.lockin_settings.sample_rate_Hz
if target_sample_rate_Hz == "max":
target_sample_rate_Hz = max_len/measurement_time_s # perfect sample rate to fully fill the buffer in the measurement time
self.sample_rate_Hz = self.lockin.get_next_valid_sample_rate_Hz(target_sample_rate_Hz)
self.n_bins = int(measurement_time_s * self.sample_rate_Hz)
log.info(f"[mode={self.mode}]: Lock-In={self.lockin.get_device_name()}, target_sample_rate={target_sample_rate_Hz} Hz, sample_rate={self.sample_rate_Hz} Hz, for t = {measurement_time_s} => n_bins={self.n_bins}")
return self.sample_rate_Hz, self.n_bins
def __getattr__(self, name):
"""
This allows any LockInAmp method to be called on an LockInHandler object.
A wrapper is added, so that if an IO error occurs during that method, a recovery attempt is performed.
"""
try:
attr = getattr(self.lockin, name)
except AttributeError:
raise AttributeError(f"LockInHandler or LockInAmp do not have attribute '{name}'")
if callable(attr):
def method_wrapper(*args, **kwargs):
com_success = self.n_try
err = None
while com_success > 0:
try:
return attr(*args, **kwargs)
except pyvisa.VisaIOError as er:
err = er
# TODO: retry if status bit is set
self.lockin.try_recover_from_communication_error(er)
com_success -= 1
raise RuntimeError(f"[mode={self.mode}]: pyvisa.VisaIOError: {err}")
return method_wrapper
else:
return attr
[docs]
class MeasurementHandler:
def __init__(self,
measurement_settings: MeasurementSettings,
wavelengths: list[str|int|float],
data_queue: Queue,
command_queue: Queue,
monochromators: Monochromator|list[Monochromator],
shutter: Shutter,
add_measurement_info_to_metadata: bool=True,
compress_data: bool=True,
):
"""
Class that performs the actual measurement.
:param monochromator:
:param shutter:
:param measurement_params:
:param signal_to_noise_f: A function that takes the PrsData and the current wavelength and determines whether the
signal to noise ratio is already satisfying. If the function returns True, the measurement is stopped early.
:param command_queue: A queue used to receive a "stop" or other signals. Useful when `run()` is run in another thread
:param data_queue: A queue where the data of a certain wavelength is put in after it has been measured.
Useful for live monitoring when `run()` is run in another thread
:param add_measurement_info_to_metadata:
:param compress_data: Whether to use gzip compression on full data file
"""
self.measurement_settings = measurement_settings
self.data_queue:Queue = data_queue
self.command_queue:Queue = command_queue
self.modes: dict[str, tuple[LockInHandler, PrsData, Amplifier|None]] = {}
if isinstance(monochromators, Monochromator):
self.monochromators = [monochromators]
else:
self.monochromators = monochromators
self.shutter = shutter
self.add_measurement_info_to_metadata = add_measurement_info_to_metadata
self.compress_data = compress_data
self.auto_gain = measurement_settings.auto_gain
self.auto_sensitivity = measurement_settings.auto_sensitivity
self.wait_time_s = -1
self.measurement_time_s = -1
self.timeout_s = -1
self.timeout_interval = -1
self.wavelengths = wavelengths
self.target_I_range_V: tuple[float, float] = (0.6, 9.5)
#
# PREPARATION
#
[docs]
def add_measurement_mode(self, mode: str, lockin: LockInAmp, lockin_settings: LockInSettings, data: PrsData, pre_amp: Amplifier|None=None):
"""
Add a Lock-In that records data using the measurement.
:param mode: key of the measurement mode, eg 'ref' or 'tra'
:param lockin:
:param data: PrsData object with no data initialized (it may already contain metadata). Each mode needs its own PrsData object
:param aux_DC: The name of auxiliary input channel the DC signal is connected to on the lock-in
"""
handler = LockInHandler(mode, lockin, lockin_settings)
self.modes[mode] = (handler, data, pre_amp)
[docs]
def prepare(self):
"""
Setup some settings, filling in defaults for missing parameters.
"""
# unfortunately, some stuff is lock-in specific
# use string checks to avoid having to import the modules
self.wait_time_s = self.measurement_settings.wait_time_s
self.measurement_time_s = self.measurement_settings.measurement_time_s
for (handler, data, _) in self.modes.values():
sample_rate_Hz, n_bins = handler.set_measurement_time(self.measurement_time_s)
# add sample rate that is actually going to be used to metadata
print(data.metadata)
if self.add_measurement_info_to_metadata:
get_subdict(data.metadata, "lock-in-amp")["sample_rate_Hz"] = sample_rate_Hz
self.timeout_s = 3 * self.measurement_time_s
self.timeout_interval = 0.25
[docs]
def auto_determine_gain(self,
min_DC: float=0.8,
max_DC: float=9.95,
N_test: int=6,
wait_time_s: float=0.2,
):
"""
Auto-determine the optimum gain value for each pre-amplifier.
Sweeps a few points in the spectrum and adjust the gain so that the DC value does not exceed `max_DC` V, but if possible, stays above `min_DC` mV.
This function must be called after `prepare`, all relevant parameters (especially the monochromator bandwidth) must have been set on the devices already.
:param min_DC: The minimum voltage for the DC signal.
:param max_DC: The maximum voltage for the DC signal. If any value crosses this threshold, the gain is reduced, even if this causes other wavelengths to o below `min_DC`.
:param wait_time_s: Time to wait after setting the wavelength. Small values should be enough since only the DC value is measured.
:param N_test: Target number of wavelengths for which to check the DC value
:raises RuntimeError: If `prepare()` was not called first
"""
if self.measurement_time_s < 0:
raise RuntimeError(f"You must call `prepare()` before `auto_determine_gain()`")
# filter offsets
wls = [wl for wl in self.wavelengths if isinstance(wl, float)]
N_wls = min(N_test, len(wls))
test_wavelengths = np.linspace(wls[0], wls[-1], N_wls)
# sort to start by shortest wavelength
# least DC signal to be expected at low wavelengths because of weak lamp emission and strong sample absorption
test_wavelengths = np.sort(test_wavelengths)
log.debug(f"Testing {N_wls} wavelengths: {test_wavelengths}")
DC_values_modes: dict[str, list[float]] = {key: [] for key in self.modes.keys()}
self.shutter.open()
i = 0
while i < len(test_wavelengths):
wl = test_wavelengths[i]
log.info(f"Measuring at lambda={wl} nm")
for m in self.monochromators:
m.set_wavelength_nm(wl)
time.sleep(wait_time_s)
# Adjust the gain for each mode
for mode_idx, (mode_key, (lockin, data, pre_amp)) in enumerate(self.modes.items()):
if pre_amp is None: continue
DC_values = DC_values_modes[mode_key]
val = lockin.lockin.read_value(lockin.lockin_settings.aux_DC)
# INCREASE if value below threshold and all other values still below max after change
while val < min_DC and all([v * 10 <= max_DC for v in DC_values]):
try:
gain = pre_amp.increase_gain()
log.debug(f"Increased gain for mode='{mode_key}' to 10^{gain} V/A (value was {val:.3f} V)")
for j in range(len(DC_values)):
DC_values[j] *= 10
time.sleep(wait_time_s)
val = lockin.lockin.read_value(lockin.lockin_settings.aux_DC)
except ValueError as e:
log.error(f"Can not decrease gain anymore ({type(e)}: {e})")
break
# DECREASE if above max -> dont check if other values become too small
while val >= max_DC:
try:
gain = pre_amp.decrease_gain()
log.debug(f"Decreased gain for mode='{mode_key}' to 10^{gain} V/A (value was {val:.3f} V)")
for j in range(len(DC_values)):
DC_values[j] *= 0.1
time.sleep(wait_time_s)
val = lockin.lockin.read_value(lockin.lockin_settings.aux_DC)
except ValueError as e:
log.error(f"Can not decrease gain anymore ({type(e)}: {e})")
break
DC_values.append(val)
i += 1
self.shutter.close()
# Readout final gain
for mode_idx, (mode_key, (lockin, data, pre_amp)) in enumerate(self.modes.items()):
try:
gain = pre_amp.get_gain()
except Exception as e:
log.error(e)
gain = None
log.debug(f"Final gain for mode='{mode_key}' is 10^{gain} V/A")
# TODO: this is ugly, the value belongs to the measurement settings per mode, which are not available here
# self.lockin. = gain
return wls, DC_values_modes
#
# RUN
#
[docs]
def run(self):
"""
Run the measurement. You must call `prepare()` first.
`run()` may be run a separate thread.
"""
if self.measurement_time_s < 0:
raise RuntimeError(f"You must call `prepare()` before `run()`")
t_start = get_time()
t_start_t = time.time()
for k, (lockin, data, _) in self.modes.items():
if self.add_measurement_info_to_metadata:
data.metadata["measurement_time_start"] = t_start
data.metadata["reference_freq_Hz_start"] = lockin.get_frequency_Hz()
# write metadata to disk
data.write_metadata()
data.metadata["messages"] = []
try:
self._run()
except KeyboardInterrupt:
log.info("Keyboard interrupt, stopping measurement")
except Exception as e:
log.critical(f"Unexpected error, stopping measurement. Error: {e}")
if self.command_queue is not None:
self.command_queue.put(("exception", e))
raise RuntimeError(f"Unexpected error, stopping measurement") from e
for k, (lockin, data, _) in self.modes.items():
if self.add_measurement_info_to_metadata:
t_stop = get_time()
t_stop_t = time.time()
data.metadata["measurement_time_stop"] = t_stop
data.metadata["measurement_time_duration"] = duration_to_string(t_stop_t - t_start_t)
data.metadata["reference_freq_Hz_stop"] = lockin.get_frequency_Hz()
# Write again after having updated the stop time
data.write_metadata()
data.write_full_file(compress=self.compress_data)
log.info("Spectrum measurement done")
#
# INTERNAL RUN-UTILITIES
#
def _auto_sensitivity(self, wl):
"""
Check for a lock-in output overload and increase the sensitivity value until the overload is gone
"""
for k, (lockin, data, _) in self.modes.items():
mode = lockin.mode
# reset status register
lockin.start_check_overloads()
# Check if the sensitivity is out of range by testing for an overload of the output
overload = "output" in lockin.check_overloads().lower()
n_tries = 10
while overload and n_tries > 0:
n_tries -= 1
try:
new_sensi = lockin.increase_full_scale_sensitivity_volt()
msg = f"{wl} nm: lock-in reports output overload, increasing sensitivity to {new_sensi} V"
log.warning(f"[mode={mode}]:" + msg)
data.metadata["messages"].append(msg)
self.data_queue.put(("warning", msg))
time.sleep(5)
lockin.start_check_overloads()
time.sleep(5)
overload = "output" in lockin.check_overloads().lower()
except Exception as e:
log.error(f"[mode={mode}]: Failed to increase sensitivity: {e}")
def _is_gain_modification_needed_all(self) -> bool:
"""
Determine if any measurement mode requires a modification of the amplifier gain setting.
:return: True if modification is needed
"""
is_needed = False
for k, (lockin, data, pre_amp) in self.modes.items():
if pre_amp is None: continue
if self._is_gain_modification_needed(lockin):
is_needed = True
return is_needed
def _is_gain_modification_needed(self, lockin: LockInHandler):
I: float = lockin.lockin.read_value(lockin.lockin_settings.aux_DC)
if I < self.target_I_range_V[0]:
return True
elif I > self.target_I_range_V[1]:
return True
return False
def _auto_gain_all(self, wl_nm):
"""
Increase or decrease the pre-amp gain if the DC value goes out of the target range.
:note: Gain is only increased or decreased by max. 1 step during this function! Do not use it to set the initial gain.
"""
for k, (lockin, data, pre_amp) in self.modes.items():
if pre_amp is None: continue
while self._auto_gain(wl_nm, lockin, data, pre_amp):
time.sleep(0.1)
def _auto_gain(self, wl_nm, lockin: LockInHandler, data: PrsData, pre_amp:Amplifier|None) -> bool:
if pre_amp is None: return False
mode = lockin.mode
I: float = lockin.lockin.read_value(lockin.lockin_settings.aux_DC)
log.debug(f"[mode={mode}]: _auto_gain. I={I} V")
gain = None
old_gain = None
if I < self.target_I_range_V[0]:
try:
old_gain = pre_amp.get_gain()
gain = pre_amp.increase_gain()
except Exception as e:
log.error(f"[mode={mode}]: Failed to increase gain: {type(e)}: {e}")
return False
elif I > self.target_I_range_V[1]:
try:
old_gain = pre_amp.get_gain()
gain = pre_amp.decrease_gain()
except Exception as e:
log.error(f"[mode={mode}]: Failed to decrease gain: {type(e)}: {e}")
return False
if gain is not None and old_gain is not None:
if gain > old_gain:
msg = f"{wl_nm} nm: I={I:.3f} V outside of target range, decreased gain from 10^{old_gain} to 10^{gain}"
else:
msg = f"{wl_nm} nm: I={I:.3f} V outside of target range, increased gain from 10^{old_gain} to 10^{gain}"
log.info(f"[mode={mode}]: " + msg)
data.metadata["messages"].append(msg)
self.data_queue.put(("warning", msg))
return True
return False
def _set_wavelength_and_wait(self, wl_nm):
"""
If the wavelength is a number, set the monochromator to that number and make sure the shutter is open.
If it is not a number, close the shutter.
After that, wait for the wait_time_s until the signal should be stable
"""
try:
if type(wl_nm) == int or type(wl_nm) == float:
self.data_queue.put(("set_wavelength", wl_nm))
if not self.shutter.is_open(): self.shutter.open()
for mcm in self.monochromators:
mcm.set_wavelength_nm(wl_nm)
else:
self.data_queue.put(("close_shutter", wl_nm))
if self.shutter.is_open(): self.shutter.close()
# wait the wait time
self.data_queue.put(("wait_stable", self.wait_time_s))
time.sleep(self.wait_time_s)
except Exception as ex:
log.critical(f"Exception occurred while setting wavelength and waiting: {ex}")
def _get_data(self, wl:int|float|str, mode_key: str, lockin: LockInHandler, data: PrsData, measurement_duration: int):
"""
Transfer the data from the lock-in, add it to the PrsData object, calculate the spectrum data and send it via the data queue
"""
overload = lockin.check_overloads() # check must occur before setting new wavelength, as that may lead to overload
self.data_queue.put(("get_data", mode_key))
lockin.buffer_stop_fill()
lock_in_data = lockin.buffer_get_data(start=0, n_points=lockin.n_bins) # indices in the order of the "what" param during buffer_setup
data[wl]["lock-in-R_raw"] = lock_in_data[0,:]
data[wl]["lock-in-aux_raw"] = lock_in_data[1,:]
if lockin.SR830:
# convert to np array
data[wl]["lock-in-theta_raw"] = np.array(data[wl]["lock-in-theta_raw"])
else:
data[wl]["lock-in-theta_raw"] = lock_in_data[2,:]
if overload:
msg = f"[{mode_key}]: Overload of {overload} at {wl} nm"
log.warning(msg)
data.metadata["messages"].append(msg)
# this measurement time is likely greatly exaggerated, because it also includes the time to transmit and process the data
data[wl]["measurement_duration_s"] = measurement_duration
# save the spectrum data and send the results for wl through the queue
# using default spectrum columns
spec_data, cols = data.get_spectrum_data(wavelengths=[wl])
data.write_partial_file(wl)
# if wl is a string, write the result of R and theta to metadata make it compatible with versions < 0.4
if type(wl) == str:
# for offsets, lock-in-R = dI and lock-in-aux = I
R = float(spec_data[0, cols.index("lock-in-R")])
theta = float(spec_data[0, cols.index("theta")])
I = float(spec_data[0, cols.index("lock-in-aux")])
data.metadata[wl] = {
data.get_column_label_with_unit("dI"): R,
data.get_column_label_with_unit("I"): I,
data.get_column_label_with_unit("theta"): theta,
}
data.write_metadata()
self.data_queue.put(("offset", mode_key, R, theta))
elif self.data_queue is not None:
self.data_queue.put(("data", mode_key, spec_data))
def _run(self):
"""
Run a spectrum measurement.
self.wavelengths if a list of wavelengths and offsets to measure.
Parallelization in the form of multithreading is used to save time:
setting the wavelength+waiting and the data transfer+processing are run in separate threads, when all are done the next measurement phase begins.
One measurement includes:
- Setting the wavelength
- Wait until the signal should be stable
- Initializing empty data in PrsData objects
- Auto-increase sensitivity if necessary
- Measure using the lock-in buffer for <measurement time>, for models where only 2values can be stored in the buffer, the phase is measured separately in regular intervals (leading to much less data points)
- Transfer the data from the lock-in
- Calculate means and errors, send them to the parent thread via the data queue
- Check the command for commands
"""
i_wl = 0
wl = self.wavelengths[i_wl]
sub_thread_q = Queue()
thread_set_wl = ThreadWithExceptionSupport(queue=sub_thread_q, target=self._set_wavelength_and_wait, args=[wl])
thread_set_wl.start()
lockin_threads: dict[str, threading.Thread] = {}
thread_join_timeout = max(90, 2*self.wait_time_s)
err = None
try:
while i_wl < len(self.wavelengths):
wl = self.wavelengths[i_wl]
log.info(f"Measuring at lambda={wl} nm")
# join get-data threads
for k, thread in lockin_threads.items():
thread.join(timeout=thread_join_timeout) # getting data should never take longer than 90 seconds
if thread.is_alive():
raise RuntimeError(f"Timed out while waiting lock-in thread for '{k}' to finish")
# check errors
if sub_thread_q.qsize() > 0:
err, trace = sub_thread_q.get()
msg = f"A thread in _run had an error: {type(err)}: {err}\nTraceback: {trace}"
raise RuntimeError(msg) from err
# setup new measurement
for k, (lockin, data, _) in self.modes.items():
if lockin.SR830:
lockin.buffer_setup(what=["R", lockin.lockin_settings.aux_DC], length=lockin.n_bins, sample_rate_Hz=lockin.sample_rate_Hz)
else:
lockin.buffer_setup(what=["R", lockin.lockin_settings.aux_DC, "theta"], length=lockin.n_bins, sample_rate_Hz=lockin.sample_rate_Hz)
# while setting wavelength, prepare measurement
data[wl] = {}
data[wl]["lock-in-R_raw"] = np.empty(lockin.n_bins, dtype=float)
data[wl]["lock-in-aux_raw"] = np.empty(lockin.n_bins, dtype=float)
data[wl]["lock-in-theta_raw"] = []
def measure_phase():
for k, (lockin, data, _) in self.modes.items():
if lockin.SR830:
data[wl]["lock-in-theta_raw"].append(lockin.read_value("theta"))
else:
pass
# wait until signal is stable
thread_set_wl.join(timeout=thread_join_timeout)
if thread_set_wl.is_alive():
raise RuntimeError(f"Timed out while waiting for the thread setting the wavelength to finish")
if sub_thread_q.qsize() > 0:
err, trace = sub_thread_q.get()
msg = f"A thread in _run had an error: {err}"
raise RuntimeError(msg) from err
self.data_queue.put(("measuring", ))
# auto sensitivity could be sped up by parallelization
if self.auto_sensitivity: self._auto_sensitivity(wl)
for k, (_, data, _) in self.modes.items():
data[wl]["timestamp_start"] = time.time()
measure_phase()
for k, (lockin, _, _) in self.modes.items():
lockin.buffer_start_fill()
t_start = time.time()
# check if its done
t = self.timeout_s
while t > 0:
t -= self.timeout_interval
time.sleep(self.timeout_interval)
measure_phase()
all_done = True
# ideally, both should finish at the same time, but since the measurements be be started a few hundred ms apart
# they might not finish at exactly the same time
for k, (lockin, _, _) in self.modes.items():
n_points: int = lockin.buffer_get_n_points()
if n_points < lockin.n_bins: # not done
all_done = False
if all_done:
break
if t < 0: raise RuntimeError(f"Timed out waiting for measurement on wavelength {wl} nm to finish")
# measurement time up, get the data and set next wavelength
t_stop = time.time()
# acquire data, MUST be started before setting wavelength since the overload check occurs there,
# and setting a wavelength might cause an overload
for k, (lockin, data, _) in self.modes.items():
lockin_threads[k] = ThreadWithExceptionSupport(queue=sub_thread_q, target=self._get_data, args=[wl, k, lockin, data, t_stop-t_start])
lockin_threads[k].start()
if i_wl < len(self.wavelengths) - 1: # save time by already setting the wavelength now
next_wl = self.wavelengths[i_wl+1]
thread_set_wl = ThreadWithExceptionSupport(queue=sub_thread_q, target=self._set_wavelength_and_wait, args=[next_wl])
thread_set_wl.start()
# if a pipe was given, check for messages
if self.command_queue is not None and self.command_queue.qsize() > 0:
recv = self.command_queue.get(block=False)
if recv == "stop":
log.info(f"Received 'stop', stopping measurement")
break
elif type(recv) == tuple and recv[0] == "metadata":
log.info(f"Received 'metadata', updating metadata")
for k, (_, data, _) in self.modes.items():
data.metadata |= recv[1]
data.write_metadata()
else:
log.error(f"Received invalid message: '{recv}'")
i_wl += 1
except Exception as e:
err = e
finally:
if thread_set_wl.is_alive():
log.debug(f"Joining thread: 'set_wl'")
thread_set_wl.join(timeout=thread_join_timeout)
for k, thread in lockin_threads.items():
if thread.is_alive():
log.debug(f"Joining thread: 'get-data-{k}'")
thread.join()
if err:
raise err
