Source code for prsctrl.measurement.settings2

"""
Measurement settings
"""
from devctrl.devices import *  # all device classes

from typing import Any

mset = dict[str, Any]

import logging
log = logging.getLogger(__name__)




def assert_subdict_exists(d, key):
    """
    Create a sub-dictionary if it does not exist
    """
    if key in d:
        if type(d[key]) != dict:
            raise ValueError(f"Value to key '{key}' is not a dict, but a {type(d[key])}")
    else:
        d[key] = {}




def get_subdict(d, key):
    assert_subdict_exists(d, key)
    return d[key]




def extract_settings_dict(settings:dict, type_key:str, initial: dict|None) -> dict:
    """
    If both are dictionaries, raise an error.
    If None are dictionaries, return {}
    Else return initial or settings[type_key], which ever is a dictionary
    If settings[type_key] is returned, it is removed from settings
    """
    if type(initial) == dict and type_key in settings:
            raise RuntimeError(f"Device settings must either be passed directly or be in the measurement_settings, but got both for '{type_key}'")
    if type(initial) == dict:
        return initial
    if type_key in settings:
        if type(settings[type_key]) != dict:
            raise ValueError(f"Value to key '{type_key}' is not a dict, but a {type(settings[type_key])}")
        initial = settings[type_key].copy()
        del settings[type_key]
        return initial
    return {}




def _apply_settings(
        device: Device|None, 
        device_settings: mset|None,
        measurement_settings: mset,
        device_key: str,
        key_to_setter: dict,
        ignore: list[str],
        defaults={},
        add_name=True,
        raise_when_invalid=True,
        ignore_error=True,
        **kwargs
    ) -> dict:
    dev_settings = defaults | extract_settings_dict(measurement_settings, device_key, device_settings) | kwargs
    log.debug(f"{device_key}: {device} <- {dev_settings}")
    if dev_settings and (device is None):
        log.warning(f"Found {device_key} settings, but the device is None!")
    elif device and (not dev_settings):
        log.warning(f"No settings for device '{device_key}'")
    if device is not None:
        for k, v in dev_settings.items():
            if k in key_to_setter:
                try:
                    key_to_setter[k](device, v)
                except Exception as e:
                    if ignore_error:
                        log.error(f"Device={device_key}: failed to set '{k}' to '{v}': {e}")
                    else:
                        raise e
            elif k in ignore:
                pass
            elif raise_when_invalid:
                raise KeyError(f"Invalid key for {device_key}: '{k}'")
            else:
                log.warning(f"Invalid key for {device_key}: '{k}'")
    if add_name:
        dev_settings["name"] = str(device)
        if isinstance(device, Device):
            dev_settings["name_model"] = device.get_device_name()
    return dev_settings


"""
Each of the apply_settings_xxx function takes
- a device
- a device settings dictionary, that only contains keys associated with the device
- a measurement dictionary, containing general measurement settings. Device settings are stored
    in a sub-dictionary with a given key. Device settings that are applied are removed from this dictionary.
Only one method must be used: If the device settings are given directly, the device key must appear in measurement_settings.
If the device are given via the sub-dictionary in measurement_settings, device_settings must be None.

The function returns a dictionary of all the settings that were applied to the device.
"""



def apply_settings_amplifier(amplifier: Amplifier|None, device_settings: mset|None, measurement_settings: mset, **kwargs):
    key_to_setter = {
        "coupling": lambda dev, v: dev.set_coupling(v),
        "gain":     lambda dev, v: dev.set_gain(v)
    }
    settings =  _apply_settings(amplifier, device_settings, measurement_settings, "pre-amp", key_to_setter, [], **kwargs)
    try:
        mode = amplifier.get_mode()
        settings["mode"] = mode
    except Exception as e:
        log.debug(f"Failed to get amplifier mode: {e}")
    return settings





def apply_settings_monochromator(mcm: Monochromator|None, device_settings: mset|None, measurement_settings: mset, **kwargs):
    key_to_setter = {
        "bandwidth_nm": lambda dev, v: dev.set_resolution_nm(v),
        "bandwidth_eV": lambda dev, v: dev.set_resolution_eV(v),
    }
    return _apply_settings(mcm, device_settings, measurement_settings, "monochromator", key_to_setter, [], **kwargs)





def apply_settings_led(led: LedController|None, device_settings: mset|None, measurement_settings: mset, **kwargs):
    key_to_setter = {
        "mode":      lambda dev, v: dev.set_mode(v),
        "current_A": lambda dev, v: dev.set_current_A(v),
    }
    settings = _apply_settings(led, device_settings, measurement_settings, "led", key_to_setter, [], **kwargs)
    # Get the name of the LED lamp that is connected to the controller
    if isinstance(led, LedController):
        try:
            settings["name_led"] = led.get_led_name(),
        except Exception as e:
            log.error(f"Failed to get led name: {e}")
    return settings




def apply_settings_laser(laser: Laser|None, device_settings: mset|None, measurement_settings: mset, **kwargs):
    key_to_setter = {
        "mode":      lambda dev, v: dev.set_mode(v),
        "power_mW": lambda dev, v: dev.set_power_mW(v),
    }
    return _apply_settings(laser, device_settings, measurement_settings, "laser", key_to_setter, [], **kwargs)




def apply_settings_lockin(lockin: LockInAmp|None, device_settings: mset|None, measurement_settings: mset, defaults: mset={}, **kwargs):
    key_to_setter = {
        "time_constant_s":   lambda dev, v: dev.set_time_constant_s(v),
        "filter_slope":      lambda dev, v: dev.set_filter_slope(v),
        "sync_filter":       lambda dev, v: dev.set_sync_filter(v),
        "reserve":           lambda dev, v: dev.set_reserve(v),  # not all lockins have it
        "sensitivity_volt":  lambda dev, v: dev.set_sensitivity_volt(v),
        "frequency_Hz":      lambda dev, v: dev.set_frequency_Hz(v),
        "reference":         lambda dev, v: dev.set_reference(v),
        "reference_trigger": lambda dev, v: dev.set_reference_trigger(v),
    }
    ignore = ["aux_DC", "sample_rate_Hz"]
    return _apply_settings(lockin, device_settings, measurement_settings, "lock-in-amp", key_to_setter, ignore=ignore, defaults=defaults, **kwargs)