"""
DSSC-related sub-routines.
comment: contributions should comply with pep8 code structure guidelines.
"""
import logging
import numpy as np
import xarray as xr
from imageio import imread
import extra_data as ed
from .xgm import get_xgm
from .digitizers import get_tim_peaks
__all__ = [
'create_dssc_bins',
'get_xgm_formatted',
'load_dssc_info',
'load_mask',
'quickmask_DSSC_ASIC',
]
log = logging.getLogger(__name__)
[docs]def load_dssc_info(proposal, run_nr):
"""
Loads the first data file for DSSC module 0 (this is hardcoded)
and returns the detector_info dictionary
Parameters
----------
proposal: str, int
number of proposal
run_nr: str, int
number of run
Returns
-------
info : dictionary
{'dims': tuple, 'frames_per_train': int, 'total_frames': int}
"""
module = ed.open_run(proposal, run_nr, include='*DSSC01*')
info = module.detector_info('SCS_DET_DSSC1M-1/DET/1CH0:xtdf')
info["number_of_trains"] = len(module.train_ids)
info["trainIds"] = module.train_ids
log.debug("Fetched information for DSSC module nr. 1.")
return info
[docs]def create_dssc_bins(name, coordinates, bins):
"""
Creates a single entry for the dssc binner dictionary. The produced xarray
data-array will later be used to perform grouping operations according to
the given bins.
Parameters
----------
name: str
name of the coordinate to be binned.
coordinates: numpy.ndarray
the original coordinate values (1D)
bins: numpy.ndarray
the bins according to which the corresponding dimension should
be grouped.
Returns
-------
da: xarray.DataArray
A pre-formatted xarray.DataArray relating the specified dimension with
its bins.
Examples
--------
>>> import toolbox_scs as tb
>>> run = tb.open_run(2212, 235, include='*DA*')
1.) binner along 'pulse' dimension. Group data into two bins.
>>> bins_pulse = ['pumped', 'unpumped'] * 10
>>> binner_pulse = tb.create_dssc_bins("pulse",
np.linspace(0,19,20, dtype=int),
bins_pulse)
2.) binner along 'train' dimension. Group data into bins corresponding
to the positions of a delay stage for instance.
>>> bins_trainId = tb.get_array(run, 'PP800_PhaseShifter', 0.04)
>>> binner_train = tb.create_dssc_bins("trainId",
run.trainIds,
bins_trainId.values)
"""
if name in ['trainId', 'pulse', 'x', 'y']:
da = xr.DataArray(bins,
dims=[name],
coords={name: coordinates})
log.debug(f'created dssc bin array for dimension {name}')
return da
log.debug(f'could not construct dssc bin array for dimension {name}')
raise ValueError(f'Invalid name {str(name)}: value should be '
'trainId, x, or y')
def get_tim_formatted(run_obj, tim_names, dssc_frame_coords):
"""
Load the tim data and define coordinates along the pulse dimension.
Parameters
----------
run_obj: extra_data.DataCollection
DataCollection object
tim_names: list of str
a list of valid mnemonics for tim data sources
dssc_frame_coords: int
defines which dssc frames should be normalized using data from the tim.
Returns
-------
tim: xarray.DataArray
tim data with coordinate 'pulse'.
"""
log.debug('load tim data')
tim = get_tim_peaks(run_obj, tim_names)
# average over all tim sources
tim = -tim.to_array().mean(dim='variable')
pulse_dim = [d for d in tim.dims if d != 'trainId'][0]
tim = tim.rename({pulse_dim: 'pulse'})
if type(dssc_frame_coords) == int:
dssc_frame_coords = np.arange(tim.sizes['pulse']*dssc_frame_coords,
step=dssc_frame_coords, dtype=np.uint64)
tim['pulse'] = dssc_frame_coords
log.info('loaded formatted tim data.')
return tim
[docs]def quickmask_DSSC_ASIC(poslist):
'''
Returns a mask for the given DSSC geometry with ASICs given in poslist
blanked. poslist is a list of (module, row, column) tuples. Each module
consists of 2 rows and 8 columns of individual ASICS.
Copyright (c) 2019, Michael Schneider
Copyright (c) 2020, SCS-team
license: BSD 3-Clause License (see LICENSE_BSD for more info)
'''
mask = np.ones([16, 128, 512], dtype=float) # need floats to use NaN
for (module, row, col) in poslist:
mask[module, 64 * row:64 * (row + 1), 64 * col:64 * (col + 1)] = \
np.nan
return mask
[docs]def load_mask(fname, dssc_mask):
"""
Load a DSSC mask file.
Copyright (c) 2019, Michael Schneider
Copyright (c) 2020, SCS-team
license: BSD 3-Clause License (see LICENSE_BSD for more info)
Parameters
----------
fname: str
string of the filename of the mask file
Returns
-------
dssc_mask:
"""
mask = imread(fname)
mask = dssc_mask.astype(float)[..., 0] // 255
mask[dssc_mask == 0] = np.nan
return mask