Source code for toolbox_scs.detectors.dssc_misc

"""
    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')
[docs]def get_xgm_formatted(run_obj, xgm_name, dssc_frame_coords): """ Load the xgm data and define coordinates along the pulse dimension. Parameters ---------- run_obj: extra_data.DataCollection DataCollection object providing access to the xgm data to be loaded xgm_name: str valid mnemonic of a xgm source dssc_frame_coords: int, list defines which dssc frames should be normalized using data from the xgm. Returns ------- xgm: xarray.DataArray xgm data with coordinate 'pulse'. """ log.debug('load raw xgm data') xgm = get_xgm(run_obj, xgm_name)[xgm_name] pulse_dim = [d for d in xgm.dims if d != 'trainId'][0] xgm = xgm.rename({pulse_dim: 'pulse'}) if type(dssc_frame_coords) == int: dssc_frame_coords = np.arange(xgm.sizes['pulse']*dssc_frame_coords, step=dssc_frame_coords, dtype=np.uint64) xgm['pulse'] = dssc_frame_coords log.info('loaded formatted xgm data.') return xgm
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