Source code for prsctrl.measurement.time_estimate

import logging
log = logging.getLogger(__name__)

from .settings import PrsSettings
from devctrl.devices.device_manager import DeviceManager
from devctrl.devices import DummyDevice, LockInAmp, Monochromator



def get_time_estimate_per_wavelength(devmg: DeviceManager, measurement_settings: PrsSettings, extra_wait_time_s=10) -> dict[str, float]:
    """
    :return: Number of seconds a measurement step will take (step = at one wavelength) for each mode.
    """
    # Set wavelength, not parallel so times for each monochromator add up
    monochromator_set_time = 0
    for k in devmg:
        if not k.startswith("monochromator"): continue
        if not devmg.is_connected(k): continue
        mcm = devmg.get_device(k)
        # TODO: measure times
        if "TMC" in mcm.get_device_name():  # Bentham
            monochromator_set_time += 5
        elif "Kymera" in mcm.get_device_name():
            monochromator_set_time += 5
        elif isinstance(mcm, DummyDevice):
            log.warning(f"Dummy monochromator device '{mcm.get_device_name()}', can not give good time estimate.")
            monochromator_set_time += 5
        else:
            log.warning(f"Unknown monochromator device '{mcm.get_device_name()}', can not give good time estimate.")
            monochromator_set_time += 5

    ms: PrsSettings = measurement_settings

    mode_times = {}
    for mode_key, mode_set in ms.modes.items():
        lock_in_key = f"lock-in_{mode_key}"
        if not lock_in_key in devmg:
            log.warning(f"Time prediction impossible for mode '{lock_in_key}', as it does not have an associated lock-in amplifier.")
            continue

        if not lock_in_key in devmg:
            log.warning(f"Time prediction impossible for mode '{lock_in_key}', as it does not have an associated lock-in amplifier.")
            continue

        if not devmg.is_connected(lock_in_key):
            log.warning(f"Time prediction impossible for mode '{lock_in_key}', as its lock-in amplifier is not connected.")
            continue

        lockin = devmg.get_device(lock_in_key, LockInAmp)

        mode_time = 0
        if measurement_settings.general.wait_time_s is not None:
            wait_time_s = measurement_settings.general.wait_time_s + extra_wait_time_s
        else:
            wait_time_s = lockin.get_wait_time_s(mode_set.lock_in.time_constant_s, mode_set.lock_in.filter_slope) + extra_wait_time_s
        measurement_time_s = ms.general.measurement_time_s
        log.info(f"Wait_time_s={wait_time_s}, measurement_time_s={measurement_time_s}")
        # Data Transfer + Processing
        if "830" in lockin.get_device_name():
            vals_per_sec = 666  # assuming two channels (Benchmark from 14.06.2025)
            max_size = lockin.buffer_get_max_size(2)
        elif "Model7260" in lockin.get_device_name():
            vals_per_sec = 196  # assuming two or three channels (Benchmark from 14.06.2025)
            max_size = lockin.buffer_get_max_size(3)
        else:
            log.warning(f"Dummy or unknown lock-in device '{lockin.get_device_name()}', can not give good time estimate.")
            vals_per_sec = 300
            max_size = 16383
        srat = mode_set.lock_in.sample_rate_Hz
        if srat == "max":
            srat = max_size/measurement_time_s  # perfect sample rate to fully fill the buffer in the measurement time
        srat = lockin.get_next_valid_sample_rate_Hz(srat)
        n_values = measurement_time_s * srat
        processing_time = 20
        get_data_time = n_values / vals_per_sec + processing_time

        # Time to set wavelength and wait until stable
        set_wavelength_time = monochromator_set_time + wait_time_s

        mode_time += measurement_time_s + max(get_data_time, set_wavelength_time)
        mode_times[mode_key] = mode_time
    return mode_times