jungfraulib

JungfrauCtrl

Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/jungfrau/jungfraulib.py

class JungfrauCtrl():
    def __init__(
        self,
        run_dc: extra_data.DataCollection,
        ctrl_src: str,
    ):
        """Read slow data from RUN source.
        :param run_dir: EXtra-data RunDirectory DataCollection object.
        :param ctrl_src: Control source name for accessing run slow data.
        """
        self.run_dc = run_dc
        self.ctrl_src = ctrl_src
        # run settings and mode are read as raw settings
        self.run_settings = self._get_settings()
        self.run_mode = self._get_mode()

    def _get_settings(self) -> str:
        """Get JUNGFRAU run settings and mode."""
        try:
            return self.run_dc.get_run_value(self.ctrl_src, "settings")
        except extra_data.PropertyNameError:
            print("\'settings.value\' key "
                  "is not available for this run. "
                  "`run_settings` is set to `dynamicgain`")
            return "dynamicgain"  # old data are set to `dynamicgain`.

    def _get_mode(self) -> str:
        """Get run mode from `gainMode` dataset."""
        try:
            return self.run_dc.get_run_value(self.ctrl_src, "gainMode")
        except extra_data.PropertyNameError:
            return self.run_settings

    def get_memory_cells(self) -> Tuple[int, int]:
        n_storage_cells = int(self.run_dc.get_run_value(
            self.ctrl_src, "storageCells.value")) + 1
        storage_cell_st = int(self.run_dc.get_run_value(
            self.ctrl_src, "storageCellStart.value"))
        return n_storage_cells, storage_cell_st

    def get_bias_voltage(self) -> int:
        """Get Bias voltage value from RUN source."""
        if "highVoltage" in self.run_dc[self.ctrl_src]:
            return int(
                self.run_dc.get_run_value(self.ctrl_src, "highVoltage")[0])
        else:  # Old dataset "vHighVoltage" before end of 2022.
            return int(
                self.run_dc.get_run_value(self.ctrl_src, "vHighVoltage")[0])

    def get_integration_time(self) -> float:
        return(float(self.run_dc.get_run_value(
            self.ctrl_src, "exposureTime.value")) * 1e6)

    def get_gain_setting(self) -> int:
        """Get run gain settings to identify if run is in
        High CDS or Low CDS.
        - `1` if run_settings = highgain0.
        - `0` if run_settings = gain0 or None.
        """
        # Check if `run_settings` is of an old settings value
        # to convert into new settings value.
        if self.run_settings in [s.value for s in JungfrauSettings]:
            settings = self.run_settings
        else:
            settings = _old_settings_to_new(self.run_settings, 0)

        if settings == JungfrauSettings.HIGH_GAIN_0.value:
            return 1
        else:  # JungfrauSettings.GAIN_0
            return 0

    def get_gain_mode_str(self):
        # Check if run_mode is of an old settings to convert
        # into new mode value.
        if self.run_mode in [m.value for m in JGM]:
            return self.run_mode
        else:
            return _old_settings_to_new(self.run_mode, 1)

    def get_gain_mode(self) -> int:
        """Get gain mode value. Fixed `1` or Adaptive `0`.
        Returns:
            (int): gain mode parameter condition
        """
        gm_enum = self.get_gain_mode_str()

        if gm_enum in [JGM.FIX_GAIN_1.value, JGM.FIX_GAIN_2.value]:
            return 1
        else:  # DYNAMIC, FORCE_SWITCH_G1, FORCE_SWITCH_G2 or None
            return 0

__init__(run_dc, ctrl_src)

Read slow data from RUN source. :param run_dir: EXtra-data RunDirectory DataCollection object. :param ctrl_src: Control source name for accessing run slow data.

Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/jungfrau/jungfraulib.py

def __init__(
    self,
    run_dc: extra_data.DataCollection,
    ctrl_src: str,
):
    """Read slow data from RUN source.
    :param run_dir: EXtra-data RunDirectory DataCollection object.
    :param ctrl_src: Control source name for accessing run slow data.
    """
    self.run_dc = run_dc
    self.ctrl_src = ctrl_src
    # run settings and mode are read as raw settings
    self.run_settings = self._get_settings()
    self.run_mode = self._get_mode()

get_bias_voltage()

Get Bias voltage value from RUN source.

Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/jungfrau/jungfraulib.py

def get_bias_voltage(self) -> int:
    """Get Bias voltage value from RUN source."""
    if "highVoltage" in self.run_dc[self.ctrl_src]:
        return int(
            self.run_dc.get_run_value(self.ctrl_src, "highVoltage")[0])
    else:  # Old dataset "vHighVoltage" before end of 2022.
        return int(
            self.run_dc.get_run_value(self.ctrl_src, "vHighVoltage")[0])

get_gain_mode()

Get gain mode value. Fixed 1 or Adaptive 0.

Returns:

Type	Description
int	gain mode parameter condition

Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/jungfrau/jungfraulib.py

def get_gain_mode(self) -> int:
    """Get gain mode value. Fixed `1` or Adaptive `0`.
    Returns:
        (int): gain mode parameter condition
    """
    gm_enum = self.get_gain_mode_str()

    if gm_enum in [JGM.FIX_GAIN_1.value, JGM.FIX_GAIN_2.value]:
        return 1
    else:  # DYNAMIC, FORCE_SWITCH_G1, FORCE_SWITCH_G2 or None
        return 0

get_gain_setting()

Get run gain settings to identify if run is in High CDS or Low CDS. - 1 if run_settings = highgain0. - 0 if run_settings = gain0 or None.

Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/jungfrau/jungfraulib.py

def get_gain_setting(self) -> int:
    """Get run gain settings to identify if run is in
    High CDS or Low CDS.
    - `1` if run_settings = highgain0.
    - `0` if run_settings = gain0 or None.
    """
    # Check if `run_settings` is of an old settings value
    # to convert into new settings value.
    if self.run_settings in [s.value for s in JungfrauSettings]:
        settings = self.run_settings
    else:
        settings = _old_settings_to_new(self.run_settings, 0)

    if settings == JungfrauSettings.HIGH_GAIN_0.value:
        return 1
    else:  # JungfrauSettings.GAIN_0
        return 0

sort_runs_by_gain(raw_folder, runs, ctrl_src, modes_order=MODES_ORDER, expected_run_order=EXPECTED_RUN_ORDER)

Validate the 3 dark runs given for Jungfrau.

Parameters:

Name	Type	Description	Default
raw_folder	str	RAW folder for the validated dark runs.	required
runs	list	[High run, Medium run, Low run].	required
ctrl_src	str	Control source path for slow data.	required
modes_order	dict	Gain modes order to sort the runs by.	MODES_ORDER
expected_run_order	list	Expected dark runs order to process.	EXPECTED_RUN_ORDER

Raises:

Type	Description
ValueError	Wrong given dark runs

Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/cal_tools/jungfrau/jungfraulib.py

def sort_runs_by_gain(
    raw_folder,
    runs,
    ctrl_src,
    modes_order=MODES_ORDER,
    expected_run_order=EXPECTED_RUN_ORDER
):
    """Validate the 3 dark runs given for Jungfrau.

    Args:
        raw_folder (str): RAW folder for the validated dark runs.
        runs (list): [High run, Medium run, Low run].
        ctrl_src (str): Control source path for slow data.
        modes_order (dict): Gain modes order to sort the runs by.
        expected_run_order (list):Expected dark runs order to process.
    Raises:
        ValueError: Wrong given dark runs
    """
    assert len(runs) == 3, "Wrong number of runs. expected a list of 3 runs."

    run_gm_mapping = dict()
    for run in runs:
        ctrl_data = JungfrauCtrl(
            extra_data.RunDirectory(f"{raw_folder}/r{run:04d}/"),
            ctrl_src)
        gm = ctrl_data.get_gain_mode_str()
        run_gm_mapping[run] = gm

    # 1st legacy case before having run.settings in data.
    if all(value is None for value in run_gm_mapping.values()):
        warning("run.settings is not stored in the data "
                f"to read. Hence assuming gain_mode = {gm}"
                " for adaptive old data.")
        return runs

    run_gm_mapping = dict(sorted(
        run_gm_mapping.items(),
        key=lambda item: modes_order[item[1]]
        ))
    if list(run_gm_mapping.keys()) != runs:
        warning("Given dark runs are unsorted. "
                f"Runs will be sorted from {runs} of gain modes "
                f"{list(run_gm_mapping.values())} to "
                f"{list(run_gm_mapping.keys())}")

    runs = list(run_gm_mapping.keys())
    modes = list(run_gm_mapping.values())

    legacy_adaptive = [
        JGM.DYNAMIC.value,
        JGM.DYNAMIC.value,
        JGM.FORCE_SWITCH_HG1.value
    ]

    # 2nd legacy case with CTRL/MDL bug resulting in wrong run settings.
    if modes == legacy_adaptive:
        warning(f"run.settings for medium and low gain runs"
                f" are wrong {modes[1:]}. This is an expected "
                f"bug for old raw data. "
                "Assuming this is an adaptive gain runs.")
    elif not modes in expected_run_order:
        raise ValueError("Wrong dark runs are given. "
                         f"The given three runs are {runs} with "
                         f"wrong gain modes {modes}."
                         "Please verify the selected 3 dark runs to process.")

    return runs