From 939d32b98fb1443423c4749dd1e76f34a885ed4b Mon Sep 17 00:00:00 2001
From: Laurent Mercadier <laurent.mercadier@xfel.eu>
Date: Thu, 20 May 2021 21:43:19 +0200
Subject: [PATCH] Separate FastADC and ADQ412 for retrieving peak params,
support mnemonics versions
---
src/toolbox_scs/detectors/digitizers.py | 174 ++++++++++++++++--------
1 file changed, 116 insertions(+), 58 deletions(-)
diff --git a/src/toolbox_scs/detectors/digitizers.py b/src/toolbox_scs/detectors/digitizers.py
index b677323..8d42063 100644
--- a/src/toolbox_scs/detectors/digitizers.py
+++ b/src/toolbox_scs/detectors/digitizers.py
@@ -346,10 +346,9 @@ def get_peaks(run,
extra_dim)
peaks = peaks.where(mask_on, drop=True)
return peaks.assign_coords({extra_dim: pid})
-
-
-def channel_peak_params(run, source, key=None, digitizer=None,
- channel=None, board=None):
+
+
+def channel_peak_params(run, source, key=None, channel=None, board=None):
"""
Extract peak-integration parameters used by a channel of the digitizer.
@@ -364,13 +363,12 @@ def channel_peak_params(run, source, key=None, digitizer=None,
key: str
optional, used in combination of source (if source is not a ToolBox
mnemonics) instead of digitizer, channel and board.
- digitizer: {"FastADC", "ADQ412"} str
- Type of digitizer. If None, inferred from the source mnemonic.
channel: int or str
The digitizer channel for which to retrieve the parameters. If None,
- inferred from the source mnemonic.
+ inferred from the source mnemonic or source + key arguments.
board: int
- Board of the ADQ412. If None, inferred from the source mnemonic.
+ Board of the ADQ412. If None, inferred from the source mnemonic or
+ source + key arguments.
Returns
-------
@@ -381,58 +379,118 @@ def channel_peak_params(run, source, key=None, digitizer=None,
m = run_mnemonics[source]
source = m['source']
key = m['key']
- if key is not None:
- if 'network' in source:
- digitizer = 'ADQ412'
- ch_to_int = {'A': 0, 'B': 1, 'C': 2, 'D': 3}
- k = key.split('.')[1].split('_')
- channel = ch_to_int[k[2]]
- board = k[1]
- else:
- digitizer = 'FastADC'
- channel = int(source.split(':')[1].split('.')[0].split('_')[1])
- if digitizer is None:
- raise ValueError('digitizer argument is missing.')
+ if 'network' in source:
+ return adq412_channel_peak_params(run, source, key, channel, board)
+ else:
+ return fastADC_channel_peak_params(run, source, channel)
+
+
+def fastADC_channel_peak_params(run, source, channel=None):
+ """
+ Extract peak-integration parameters used by a channel of the Fast ADC.
+
+ Parameters
+ ----------
+ run: extra_data.DataCollection
+ DataCollection containing the digitizer data.
+ source: str
+ Name of Fast ADC source, e.g. 'SCS_UTC1_MCP/ADC/1:channel_5.output'.
+ channel: int
+ The Fast ADC channel for which to retrieve the parameters. If None,
+ inferred from the source.
+
+ Returns
+ -------
+ dict with peak integration parameters.
+ """
+ fastADC_keys = {
+ 'baseStart': ('baselineSettings.baseStart.value',
+ 'baseStart.value'),
+ 'baseStop': ('baseStop.value',),
+ 'baseLength': ('baselineSettings.length.value',),
+ 'enable': ('enablePeakComputation.value',),
+ 'pulseStart': ('initialDelay.value',),
+ 'pulseLength': ('peakSamples.value',),
+ 'npulses': ('numPulses.value',),
+ 'period': ('pulsePeriod.value',)
+ }
if channel is None:
- raise ValueError('channel argument is missing.')
- if isinstance(channel, str):
- ch_to_int = {'A': 0, 'B': 1, 'C': 2, 'D': 3}
- channel = ch_to_int[channel]
- if board is None and digitizer == 'ADQ412':
- raise ValueError('board argument is missing.')
- keys = None
- if digitizer == 'ADQ412':
- baseKey = f'board{board}.apd.channel_{channel}.'
- keys = ['baseStart', 'baseStop', 'pulseStart',
- 'pulseStop', 'initialDelay', 'upperLimit',
- 'enable']
- keys = {k: baseKey + k + '.value' for k in keys}
- keys['npulses'] = f'board{board}.apd.numberOfPulses.value'
- if digitizer == 'FastADC':
- if ":" in source:
- baseKey = source.split(':')[1].split('.')[0]+'.'
- else:
- baseKey = f'channel_{channel}.'
- keys = ['baseStart', 'baseStop', 'initialDelay',
- 'peakSamples', 'numPulses', 'pulsePeriod',
- 'enablePeakComputation']
- keys = {k: baseKey + k + '.value' for k in keys}
- if ":" in source:
- source = source.split(':')[0]
- tid, data = run.select(source).train_from_index(0)
- result = [data[source][k] for k in keys.values()]
- result = dict(zip(keys.keys(), result))
- if digitizer == 'ADQ412':
- result['period'] = result.pop('upperLimit') - \
- result.pop('initialDelay')
- if digitizer == 'FastADC':
- result['period'] = result.pop('pulsePeriod')
- result['npulses'] = result.pop('numPulses')
- result['pulseStart'] = result['initialDelay']
- result['pulseStop'] = result.pop('initialDelay') + \
- result.pop('peakSamples')
- result['enable'] = result.pop('enablePeakComputation')
+ channel = int(source.split(':')[1].split('.')[0].split('_')[1])
+ baseName = f'channel_{channel}.'
+ source = source.split(':')[0]
+ data = run.select(source).train_from_index(0)[1][source]
+ result = {}
+ for mnemo, versions in fastADC_keys.items():
+ for v in versions:
+ key = baseName + v
+ if key in data:
+ result[mnemo] = data[key]
+ if 'baseLength' in result:
+ result['baseStop'] = result['baseStart'] + \
+ result.pop('baseLength')
+ if 'pulseLength' in result:
+ result['pulseStop'] = result['pulseStart'] + \
+ result.pop('pulseLength')
+ return result
+
+
+def adq412_channel_peak_params(run, source, key=None,
+ channel=None, board=None):
+ """
+ Extract peak-integration parameters used by a channel of the ADQ412.
+
+ Parameters
+ ----------
+ run: extra_data.DataCollection
+ DataCollection containing the digitizer data.
+ source: str
+ Nname of ADQ412 source, e.g. 'SCS_UTC1_ADQ/ADC/1:network'.
+ key: str
+ optional, e.g. 'digitizers.channel_1_D.apd.pulseIntegral', used in
+ combination of source instead of channel and board.
+ channel: int or str
+ The ADQ412 channel for which to retrieve the parameters. If None,
+ inferred from the source mnemonic or source + key arguments.
+ board: int
+ Board of the ADQ412. If None, inferred from the source mnemonic or
+ source + key arguments.
+ Returns
+ -------
+ dict with peak integration parameters.
+ """
+ if key is None:
+ if channel is None or board is None:
+ raise ValueError('key or channel + board arguments are '
+ 'missing.')
+ else:
+ k = key.split('.')[1].split('_')
+ ch_to_int = {'A': 0, 'B': 1, 'C': 2, 'D': 3}
+ channel = ch_to_int[k[2]]
+ board = k[1]
+ baseName = f'board{board}.apd.channel_{channel}.'
+ source = source.split(':')[0]
+
+ adq412_keys = {
+ 'baseStart': (baseName + 'baseStart.value',),
+ 'baseStop': (baseName + 'baseStop.value',),
+ 'enable': (baseName + 'enable.value',),
+ 'pulseStart': (baseName + 'pulseStart.value',),
+ 'pulseStop': (baseName + 'pulseStop.value',),
+ 'npulses': (baseName + 'numPulses.value',),
+ 'initialDelay': (baseName + 'initialDelay.value',),
+ 'upperLimit': (baseName + 'upperLimit.value',),
+ 'npulses': (f'board{board}.apd.numberOfPulses.value',)
+ }
+
+ data = run.select(source).train_from_index(0)[1][source]
+ result = {}
+ for mnemo, versions in adq412_keys.items():
+ for key in versions:
+ if key in data:
+ result[mnemo] = data[key]
+ result['period'] = result.pop('upperLimit') - \
+ result.pop('initialDelay')
return result
--
GitLab