Source code for toolbox_scs.mnemonics_machinery

""" Handling ToolBox mnemonics sub-routines

    Copyright (2021) SCS Team.

    (contributions preferrably comply with pep8 code structure
    guidelines.)
"""

import logging

from .constants import mnemonics as _mnemonics
from extra_data import open_run
from copy import deepcopy

__all__ = [
    'mnemonics_for_run'
]

log = logging.getLogger(__name__)


[docs]def mnemonics_for_run(prop_or_run, runNB=None): """ Returns the availble ToolBox mnemonics for a give extra_data DataCollection, or a given proposal + run number. Parameters ---------- prop_or_run: extra_data DataCollection or int The run (DataCollection) to check for mnemonics. Alternatively, the proposal number (int), for which the runNB is also required. runNB: int The run number. Only used if the first argument is the proposal number. Returns ------- mnemonics: dict The dictionnary of mnemonics that are available in the run. Example ------- >>> import toolbox_scs as tb >>> tb.mnemonics_for_run(2212, 213) """ run = prop_or_run if runNB is not None: run = open_run(prop_or_run, runNB) result = {} for m in _mnemonics: version = mnemo_version_index(run, m) if version != -1: result[m] = _mnemonics[m][version] return result
def mnemo_version_index(run, mnemonic): """ Given a mnemonic and a DataCollection, checks the existence of the mnemonic versions within the tuple value of the mnemonic and returns the valid version index. Parameters ---------- run: extra_data DataCollection The run that contains the data source corresponding to the mnemonic menmonic: str A ToolBox mnemonic Returns ------- index: int The index of the tuple. If no valid version is found, returns -1. """ if len(_mnemonics[mnemonic]) == 1: if _mnemonics[mnemonic][0]['source'] in run.all_sources: return 0 return -1 for i, v in enumerate(_mnemonics[mnemonic]): if (v['source'] in run.all_sources and v['key'] in run.keys_for_source(v['source'])): log.debug(f'Found version {i} for mnemonic "{mnemonic}": {v}') return i return -1 def mnemonics_to_process(mnemo_list, merge_with, detector, func=None): """ Finds the list of mnemonics, within mnemo_list and merge_with, that correspond to arrays that are not yet loaded and/or processed by a detector function. Removes the mnemonics of the already processed arrays from the list. Parameters ---------- mnemo_list: str or list of str ToolBox mnemonics of pulse-resolved detector arrays merge_with: xarray Dataset Dataset that may contain non-processed arrays detector: str One in {'ADQ412', 'FastADC', 'FastADC2', 'XGM', 'BAM', 'PES'} func: function function that takes one argument, an unprocessed mnemonic string, and converts it into a processed one, i.e. from 'MCP2apd' to 'MCP2peaks'. If None, the function returns the input mnemonic. Returns ------- mnemonics: list of str the mnemonics to process """ if func is None: def func(x): return x det_list = ['ADQ412', 'FastADC', 'FastADC2', 'XGM', 'BAM', 'PES'] if detector not in det_list: raise ValueError(f"Detector not supported. Expecting one in {det_list}") if detector == 'BAM': det_mnemos = [m for m in _mnemonics if 'BAM' in m] default_mnemo = 'BAM1932M' default_processed = 'BAM1932M' if detector == 'XGM': det_mnemos = ['XTD10_XGM', 'XTD10_XGM_sigma', 'XTD10_SA3', 'XTD10_SA3_sigma', 'XTD10_SA1', 'XTD10_SA1_sigma', 'SCS_XGM', 'SCS_XGM_sigma', 'SCS_SA1', 'SCS_SA1_sigma', 'SCS_SA3', 'SCS_SA3_sigma'] default_mnemo = 'SCS_SA3' default_processed = 'SCS_SA3' if detector == 'ADQ412': det_mnemos = [m for m in _mnemonics if 'MCP' in m and 'XTD10_' not in m] default_mnemo = 'MCP2apd' default_processed = 'MCP2peaks' if detector == 'FastADC': det_mnemos = [m for m in _mnemonics if 'FastADC' in m\ and 'FastADC2_' not in m] default_mnemo = 'FastADC5raw' default_processed = 'FastADC5peaks' if detector == 'FastADC2': det_mnemos = [m for m in _mnemonics if 'FastADC2_' in m] default_mnemo = 'FastADC2_5raw' default_processed = 'FastADC2_5peaks' if detector == 'PES': det_mnemos = [m for m in _mnemonics if 'PES' in m and 'raw' in m] default_mnemo = 'PES_W_raw' default_processed = 'PES_W_tof' dig_dims = list(set([_mnemonics[m][0]['dim'][0] for m in det_mnemos])) processed_mnemos = list(set([func(m) for m in det_mnemos])) # create a list of mnemonics to process from the provided mnemonics and # merge_with Dataset mw_mnemos = [] mw_processed = [] if bool(merge_with): mw_mnemos = [m for m in merge_with if m in det_mnemos and any(dim in merge_with[m].dims for dim in dig_dims)] mw_processed = [m for m in merge_with if m in processed_mnemos and any(dim in merge_with[m].dims for dim in dig_dims) is False] if mnemo_list is None: mnemonics = [] if len(mw_mnemos) == 0 and default_processed not in mw_processed: mnemonics = [default_mnemo] else: mnemonics = [mnemo_list] if isinstance(mnemo_list, str) else mnemo_list mnemonics = list(set(mnemonics + mw_mnemos)) for m in mnemonics[:]: if func(m) in mw_processed: mnemonics.remove(m) return mnemonics