Source code for prsctrl_gui.task.run.measurement
"""Widgets for running measurement tasks."""
from PyQt6.QtWidgets import QHBoxLayout, QLabel, QProgressBar
from devctrl.task.base import Task
from devctrl.gui.task import TaskRunWidget
from devctrl.devices.device_manager import DeviceManager
from devctrl.devices import Device, Monochromator, Amplifier, DummyDevice, Laser, LedController, LockInAmp, TemperatureSensor
from devctrl.utility.file_io import get_date_time_for_filename, get_next_filename, sanitize_filename
from devctrl.utility.time import duration_to_string
from devctrl.gui.exception import show_error, show_warning, show_exception
from prsctrl.version import version as prsctrl_version
from prsctrl import data as prsdata
from prsctrl.data import PrsData, REFLECTION, TRANSMISSION, ABSORPTION
from prsctrl import device_keys as devkeys
from prsctrl.measurement.measurement import MeasurementHandler
from prsctrl.measurement.time_estimate import get_time_estimate_per_wavelength
from prsctrl.measurement.pump_power_calibration import PowerCalibration
from dataclasses import dataclass
import os
import time
import logging
log = logging.getLogger(__name__)
from typing import Optional
import multiprocessing
from threading import Thread
from ...app_state import device_manager
from ..tasks import MeasurementTask
from ...utility.config import AppConfig
from ...version import version as prsctrl_gui_version
from ...app_state import live_data, sample_db, power_calibration_db
@dataclass
class _State:
proc_measure: Thread
measurement_handler: MeasurementHandler
command_queue: multiprocessing.Queue
data_queue: multiprocessing.Queue
[docs]
class MeasurementTaskRunWidget(TaskRunWidget):
"""Widget for running the measurement.
This widget manages the measurement thread.
"""
[docs]
def __init__(self, task: MeasurementTask, parent=None):
super().__init__(task, parent=parent)
self.task: MeasurementTask = task
self.setLayout(QHBoxLayout())
self.w_label = QLabel("Status: N/A")
self.layout().addWidget(self.w_label)
self.w_progress = QProgressBar()
self.layout().addWidget(self.w_progress)
# self._proc_measure: Optional[Thread] = None
# self._measurement_handler: Optional[MeasurementHandler] = None
#
# self._command_queue: Optional[multiprocessing.Queue] = None
# self._data_queue: Optional[multiprocessing.Queue] = None
self._state: Optional[_State] = None
self._datas: dict[str, PrsData] = {}
# self._current_wl: Optional[float] = None
self.time_start = 0.
[docs]
def __del__(self):
if self._state is None:
return
self._state.command_queue.put("stop")
self._state.proc_measure.join(timeout=10)
if self._state.proc_measure.is_alive():
log.critical("Timed out waiting for measurement thread to finish")
[docs]
def start_task(self) -> bool:
"""Start the measurement task.
1. Get device from global device manager
2. Apply measurement settings on devices
3. Initialize data directory
4. Start measurement thread
"""
from dataclasses import asdict
measurement_settings: MeasurementTask = self.task
modes = sorted(measurement_settings.modes.keys())
self.time_start = time.time()
# 1. GET DEVICES
self.set_status("Status: Acquire devices.")
devmg: DeviceManager = device_manager.get()
mode_devices: dict[str, dict[str, Optional[Device]]] = {}
for mode in modes:
mode_devices[mode] = {}
mode_devices[mode]["amplifier"] = devmg.get_device_or(f"amplifier_{mode}")
mode_devices[mode]["lock-in"] = devmg.get_device_or(f"lock-in_{mode}")
shutter = devmg.get_device_or(devkeys.SHUTTER_PROBE)
mcm = devmg.get_device_or(devkeys.MONOCHROMATOR_PROBE)
kymera = devmg.get_device_or(devkeys.MONOCHROMATOR_DETECTOR)
led = devmg.get_device_or(devkeys.LED_CONTROLLER)
temp_sensor = devmg.get_device_or(devkeys.TEMPERATURE_SENSOR)
# TODO: device whether to proceed here based on dummy devices
# 2. APPLY SETTINGS
self.set_status("Applying settings on devices.")
# set correct pump power settings
power_calib_key = measurement_settings.power_calibration_key
if power_calib_key:
log.info(f"Power calibration key is '{power_calib_key}'")
power_calib_db = power_calibration_db.get()
if power_calib_key not in power_calib_db:
raise KeyError(self.tr("Invalid power calibration setting: '{power_calib_key}'.").format(power_calib_key=power_calib_key))
power_calib: PowerCalibration = power_calib_db[power_calib_key]
laser_f = power_calib.power_density_mWcm2_to_laser_power_mW if power_calib.light_source_type == "laser" else None
led_f = power_calib.power_density_mWcm2_to_led_current_A if power_calib.light_source_type == "led" else None
else:
log.info(f"No power calibration key set.")
led_f = None
laser_f = None
measurement_settings.prepare(
power_density_to_laser_power_mW=laser_f,
power_density_to_led_current_A=led_f,
)
# Determine mode and merge settings
log.info(f"The following modes will be measured: {modes}")
for mode, mdevs in mode_devices.items():
if not isinstance(mdevs["lock-in"], LockInAmp):
raise RuntimeError(f"Lock-in for mode '{mode}' must be a LockInAmp but is '{type(mdevs['lock-in'])}'. Please connect a lock-in amplifier with key 'lock-in_{mode}'.")
measurement_settings.modes[mode].lock_in.apply(mdevs["lock-in"])
pre_amp_settings = measurement_settings.modes[mode].pre_amplifier
if pre_amp_settings is not None and isinstance(mdevs["amplifier"], Amplifier):
pre_amp_settings.apply(mdevs["amplifier"])
# must come after lockin settings, since its using get_wait_time_s
if measurement_settings.general.wait_time_s is None:
# we know mdevs["lock-in"] is a LockInAmp, otherwise RuntimeError is raised in the block above
if len(mode_devices) > 0:
wait_time_s = max([mdevs["lock-in"].read_get_wait_time_s() for mdevs in mode_devices.values()])
else:
wait_time_s = 0
wait_time_s += measurement_settings.general.additional_wait_time_s
measurement_settings.general.wait_time_s = wait_time_s
monochromators: list[Monochromator] = []
for i, m in enumerate([mcm, kymera]):
if i >= len(measurement_settings.monochromators):
break
if isinstance(m, Monochromator):
measurement_settings.monochromators[i].apply(m)
monochromators.append(m)
set_pump = False
pump_source = measurement_settings.general.pump_source
if pump_source == "led":
if measurement_settings.led is None:
raise RuntimeError(f"Pump source = 'led', but there are no settings for the LED (settings.led is None)")
if not isinstance(led, LedController):
raise RuntimeError(f"Pump source = 'led', but the LED controller is not connected (led_controller = {led})")
measurement_settings.led.apply(led)
if led: led.on()
set_pump = True
if pump_source == "laser":
if measurement_settings.laser is None:
raise RuntimeError(f"Pump source = 'laser', but there are no settings for the Laser (settings.laser is None)")
laser = devmg.get_device("laser")
if not isinstance(laser, Laser):
raise RuntimeError(f"Pump source = 'laser', but there the Laser is not connected (laser = {laser})")
measurement_settings.laser.apply(laser)
if laser: laser.on()
set_pump = True
if not set_pump:
log.warning(f"Unknown pump source setting: '{pump_source}' -> pump source must be set up manually. (Valid are 'led' and 'laser')")
else:
log.info(f"Pump light is on!")
time_estimate_per_wavelength = get_time_estimate_per_wavelength(devmg, measurement_settings)
max_time = int(round(max(time_estimate_per_wavelength.values())))
wavelengths = measurement_settings.general.get_wavelengths_nm_with_offsets(max_time)
# 3. INIT DATA DIRECTORY
self.set_status("Initializing data directory")
name = "GUITEST"
dirname = None
if dirname is None:
dirname = get_date_time_for_filename() + "_" + name
dirname = sanitize_filename(dirname)
cache_data_dir = AppConfig.get_path(AppConfig.MAIN_CFG.data_dir_cache)
# Make sure data_path is a non-existing, valid path
data_path = get_next_filename(os.path.join(cache_data_dir, dirname))
# METADATA
metadata: dict = AppConfig.CACHE_CFG.global_metadata_input.copy()
# versions might be helpful when the data collection is changed
metadata["prsctrl_version"] = prsctrl_version
metadata["prsctrl-gui_version"] = prsctrl_gui_version
metadata["name"] = name
sample_key = measurement_settings.sample_key
if sample_key:
sample_db_instance = sample_db.get()
if sample_key in sample_db_instance:
metadata["sample"] = sample_db_instance[sample_key].as_dict()
else:
metadata["sample"] = sample_key
# if do_ref and do_tra and AppConfig.MEAS_CFG.get_or("put_ref_tra_in_subdir", True):
# makedirs(data_path)
# data_path = os.path.join(data_path, dirname)
if isinstance(temp_sensor, TemperatureSensor) and not isinstance(temp_sensor, DummyDevice):
metadata["environment"] = {}
try:
metadata["environment"]["temperature_C"] = temp_sensor.get_temperature_C()
metadata["environment"]["humidity"] = temp_sensor.get_humidity_percent()
except Exception as e:
show_exception("Failed to get temperature or humidity", e, log, "warning")
# apply last to take precedence
metadata |= measurement_settings.additional_metadata
log.debug(f"Data path: {data_path}")
# 4. START MEASUREMENT
self.set_status("Starting measurement")
# PREPARE MEASUREMENT
# data_collector.clear()
data_queue = multiprocessing.Queue()
command_queue = multiprocessing.Queue()
measurement_handler = MeasurementHandler(
measurement_parameters=measurement_settings.general,
wavelengths=wavelengths,
command_queue=command_queue,
data_queue=data_queue,
monochromators=monochromators,
shutter=shutter,
add_measurement_info_to_metadata=True,
compress_data=AppConfig.MAIN_CFG.compress_data,
)
self._datas = {}
for mode in modes:
mode_lockin_settings = measurement_settings.modes[mode].lock_in
mode_md = metadata.copy()
mode_md["parameters"] = measurement_settings.get_metadata(mode)
self._datas[mode] = PrsData(data={}, metadata=mode_md, data_path=data_path + "-" + mode, data_name=name, file_mode="w", exp_mode=mode)
measurement_handler.add_measurement_mode(mode, mode_devices[mode]["lock-in"], mode_lockin_settings, self._datas[mode], pre_amp=mode_devices[mode]["amplifier"])
live_data.set_datas.emit(modes)
for mode in self._datas:
set_mode_status_text(mode, self.tr("Preparing measurement..."))
measurement_handler.prepare()
auto_determine_gain = measurement_settings.auto_set_init_gain
if auto_determine_gain != False:
for mode in modes: set_mode_status_text(mode, self.tr("Automatically determining optimum gain value"))
if isinstance(auto_determine_gain, int) and not isinstance(auto_determine_gain, bool):
N_vals = auto_determine_gain
else:
N_vals = 8
measurement_handler.auto_determine_gain(N_test=N_vals)
for mode in self._datas:
set_mode_status_text(mode, self.tr("Starting measurement..."))
# BEGIN
time_start = time.time()
# Argument order must match the definition
proc_measure = Thread(target=measurement_handler.run)
self._state = _State(
proc_measure=proc_measure,
measurement_handler=measurement_handler,
command_queue=command_queue,
data_queue=data_queue
)
proc_measure.start()
try:
self.w_progress.setMaximum(len(wavelengths))
except Exception as e:
log.error(f"Failed to set progress maximum: {e}")
return True
[docs]
def update_task(self) -> Optional[bool]:
"""Process messages from the :class:`.MeasurementHandler`.
:return: True if the measurement process has finished, else None
"""
if self._state is None:
raise RuntimeError(self.tr(f"Can not stop measurement, it is not running"))
if not self._state.proc_measure.is_alive():
return True
while not self._state.data_queue.empty():
msg = self._state.data_queue.get(block=False)
# (data_queue.qsize(), "\n\n")
log.debug(f"Processing queue packet: {msg}")
# if msg[0] == "data": # "data", mode, spectrum_data
# mode = msg[1]
# # data_for_monitor = [msg[2][0,PrsData.default_spectrum_columns.index(key)] for key in ["wl"] + monitor_y_rows]
# # monitors[mode.update(*data_for_monitor)]
# live_data.data_updated.emit(mode)
# set_mode_status_text(mode, "Waiting...") # done getting data
if msg[0] == "data_raw": # "data", mode, dict with raw data
mode = msg[1]
# data_for_monitor = [msg[2][0,PrsData.default_spectrum_columns.index(key)] for key in ["wl"] + monitor_y_rows]
# monitors[mode.update(*data_for_monitor)]
wl = msg[2]
data_raw = msg[3]
live_data.data_update.emit(mode, wl, data_raw)
live_data.data_updated.emit(mode)
set_mode_status_text(mode, "Waiting...") # done getting data
elif msg[0] == "data":
pass
elif msg[0] == "offset": # msg 2 and 3 are R and theta offset
mode = msg[1]
set_mode_status_text(mode, f"Offsets: $R={msg[2]}$ V, $\\theta={msg[3]}$°")
elif msg[0] == "set_wavelength": # wavelength
for mode in self.task.modes:
set_mode_status_text(mode, f"$\\lambda = {msg[1]}$ nm")
val = self.w_progress.value()
val += 1
# self._current_wl = msg[1]
self.w_progress.setValue(val)
elif msg[0] == "close_shutter":
for mode in self.task.modes:
set_mode_status_text(mode, f"Probe light off (closed shutter)")
elif msg[0] == "wait_stable": # time
for mode in self.task.modes:
set_mode_status_text(mode, f"Waiting for signal to stabilize ({msg[1]} s)")
elif msg[0] == "get_data": # mode
mode = msg[1]
set_mode_status_text(mode, "Getting data...")
elif msg[0] == "measuring":
for mode in self.task.modes:
set_mode_status_text(mode, f"Measuring...")
elif msg[0] == "warning": # warning
for mode in self.task.modes:
set_mode_status_text(mode, f"WARNING: {msg[1]}")
else:
log.error(f"Invalid tuple received from measurement thread: {msg}")
return None
[docs]
def stop_task(self) -> bool:
if self._state is None:
# raise RuntimeError(f"Can not stop measurement, it is not running")
log.error(f"stop_task requested while measurement was not running.")
return False
log.debug(f"Stopping measurement")
measurement_settings = self.task
modes = sorted(measurement_settings.modes.keys())
self._state.command_queue.put("stop")
self._state.proc_measure.join(timeout=120) # TODO: Use wait_time_s + measurement_time_s
if self._state.proc_measure.is_alive():
log.critical("Timed out waiting for measurement thread to finish")
for mode in modes: set_mode_status_text(mode, "Stopping Measurement")
time_stop = time.time()
# Plots and CSV
save_spectrum = True
if save_spectrum:
for data in self._datas.values():
try:
data.remove_calculated_values() # make sure all data is calculated using correct offset correction
data.save_csv()
title = prsdata.util.get_data_name_template(data.dirname, template="<sample_type> <sample_batch> <info> <mode>")
_ = prsdata.plot.plot_data(data, os.path.basename(data.dirpath), data.dirpath, title=title, error_shade=True)
except Exception as e:
log.error(f"Exception while saving/plotting ({data.mode}) data: {type(e)}: {e}")
# Error repor
for data in self._datas.values():
try:
report = prsdata.check.sanity_check_data_gen_report(data)
with open(os.path.join(data.dirpath, data.dirname + "-report.md"), "w") as file:
file.write(report)
except Exception as e:
log.error(f"Exception while generating report for ({data.mode}) data: {type(e)}: {e}")
# Archive data
for data in self._datas.values():
try:
n_wavelengths_threshold = AppConfig.MAIN_CFG.archive_file_count_threshold
archive_dirs = AppConfig.get_path(AppConfig.MAIN_CFG.data_dir_archive)
prsdata.util.archive_data(data.dirpath, archive_dirs=archive_dirs, n_wavelengths_threshold=n_wavelengths_threshold)
except Exception as e:
log.error(f"Exception while archiving data: {e}")
do_ref = REFLECTION in self.task.modes
do_tra = TRANSMISSION in self.task.modes
# Copy final data to other directories
if do_ref and do_tra:
try:
if do_ref and do_tra:
copy_data_dirs = AppConfig.get_path(AppConfig.MAIN_CFG.data_dir)
norm_data_dir = AppConfig.get_path(AppConfig.MAIN_CFG.reference_data_dir)
prsdata.util.save_data_rta(self._datas[REFLECTION], self._datas[TRANSMISSION], copy_data_dirs, norm_data_dir=norm_data_dir, error_shade=True, max_ylim=(-75, 75))
except Exception as e:
log.error(f"Exception while copying ref and tra data: {type(e)}: {e}")
else:
for data in self._datas.values():
try:
copy_data_dirs = AppConfig.get_path(AppConfig.MAIN_CFG.data_dir)
prsdata.util.save_data(data.dirpath, copy_data_dirs)
except Exception as e:
log.error(f"Exception while copying ({data.mode}) data: {type(e)}: {e}")
duration = time_stop - self.time_start
time_estimate = 0 # TODO: this is a useful log message, get estimate if possible
for mode in modes:
set_mode_status_text(mode, f"Measurement took {duration_to_string(duration)} (estimate was {duration_to_string(time_estimate)})")
log.info(f"Measurement took {duration_to_string(duration)} (estimate was {duration_to_string(time_estimate)})")
self._state = None
return True