""" 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