XSpect Controller

`SpectroscopyAnalysis`

A class to perform analysis on spectroscopy data.

Source code in XSpect/XSpect_Analysis.py

class SpectroscopyAnalysis:
    """
    A class to perform analysis on spectroscopy data.
    """
    def __init__(self):
        pass

    def bin_uniques(self,run,key):
        """
        Bins unique values for a given key within a run.

        Parameters
        ----------

        run : spectroscopy_run
            The spectroscopy run instance.

        key : str
            The key for which unique values are to be binned.

        """
        vals = getattr(run,key)
        bins = np.unique(vals)
        addon = (bins[-1] - bins[-2])/2 # add on energy 
        bins2 = np.append(bins,bins[-1]+addon) # elist2 will be elist with dummy value at end
        bins_center = np.empty_like(bins2)
        for ii in np.arange(bins.shape[0]):
            if ii == 0:
                bins_center[ii] = bins2[ii] - (bins2[ii+1] - bins2[ii])/2
            else:
                bins_center[ii] = bins2[ii] - (bins2[ii] - bins2[ii-1])/2
        bins_center[-1] = bins2[-1]

        setattr(run,'scanvar_indices',np.digitize(vals,bins_center))
        setattr(run,'scanvar_bins',bins_center)

    def filter_shots(self, run,shot_mask_key, filter_key='ipm', threshold=1.0E4):
        """
        Filters shots based on a given threshold.
        For example, if we filter: xray,ipm,1E4 then X-ray shots will be filtered out if the ipm is below 1E4.

        Parameters
        ----------

        run : spectroscopy_run
            The spectroscopy run instance.

        shot_mask_key : str
            The key corresponding to the shot mask. An example being [xray,simultaneous,laser] for all x-ray shots

        filter_key : str, optional
            The key corresponding to the filter data (default is 'ipm'). 

        threshold : float, optional
            The threshold value for filtering (default is 1.0E4).

        """
        shot_mask=getattr(run,shot_mask_key)
        count_before=np.sum(shot_mask)
        filter_mask=getattr(run,filter_key)
        nan_mask = np.isnan(filter_mask)
        filtered_shot_mask=shot_mask * (filter_mask>threshold)* (~nan_mask)
        count_after=np.sum(filtered_shot_mask)
        setattr(run,shot_mask_key,filtered_shot_mask)
        run.update_status('Mask: %s has been filtered on %s by minimum threshold: %0.3f\nShots removed: %d' % (shot_mask_key,filter_key,threshold,count_before-count_after))

    def filter_nan(self, run,shot_mask_key, filter_key='ipm'):
        """
        A specific filtering implementation for Nans due to various DAQ issues. 
        Filters out shots with NaN values in the specified filter.

        Parameters
        ----------

        run : spectroscopy_run
            The spectroscopy run instance.

        shot_mask_key : str
            The key corresponding to the shot mask.

        filter_key : str, optional
            The key corresponding to the filter data (default is 'ipm').

        """
        shot_mask=getattr(run,shot_mask_key)
        count_before=np.sum(shot_mask)
        filter_mask=getattr(run,filter_key)
        filtered_shot_mask=shot_mask * (filter_mask>threshold)
        count_after=np.sum(filtered_shot_mask)
        setattr(run,shot_mask_key,filtered_shot_mask)
        run.update_status('Mask: %s has been filtered on %s by minimum threshold: %0.3f\nShots removed: %d' % (shot_mask_key,filter_key,threshold,count_before-count_after))


    def filter_detector_adu(self,run,detector,adu_threshold=3.0):
        """
        Filters is a misnomer compared to the other filter functions. 
        This sets detector pixel values below a threshold to 0.
        Specifically, to remove 0-photon noise from detectors. 

        Parameters
        ----------

        run : spectroscopy_run
            The spectroscopy run instance.

        detector : str
            The key corresponding to the detector data.

        adu_threshold : float or list of float, optional
            The ADU threshold for filtering. Can be a single value or a range (default is 3.0).

        Returns
        -------

        np.ndarray
            The filtered detector data.

        """
        detector_images=getattr(run,detector)
        if isinstance(adu_threshold,list):
            detector_images_adu = detector_images * (detector_images > adu_threshold[0])
            detector_images_adu = detector_images_adu * (detector_images_adu < adu_threshold[1])
            run.update_status('Key: %s has been adu filtered by thresholds: %f,%f' % (detector,adu_threshold[0],adu_threshold[1]))
        else:
            detector_images_adu = detector_images * (detector_images > adu_threshold)
            run.update_status('Key: %s has been adu filtered by threshold: %f' % (detector,adu_threshold))

        setattr(run,detector,detector_images_adu)

        return detector_images_adu

    def purge_keys(self,run,keys):
        """
        Purges specific keys from the run to save memory.
        This is specifically to remove the epix key immediately after processing it from the hdf5 file.
        To avoid OOM. This is different than the purge all keys method which is used to purge many of the larger analysis steps.

        Parameters
        ----------

        run : spectroscopy_run
            The spectroscopy run instance.

        keys : list of str
            The list of keys to purge.

        """
        for detector_key in keys:
            setattr(run, detector_key, None)
            run.update_status(f"Purged key to save room: {detector_key}")

    def reduce_detector_shots(self, run, detector_key,reduction_function=np.sum,  purge=True,new_key=False):
        detector = getattr(run, detector_key)
        reduced_data=reduction_function(detector,axis=0)
        run.update_status(f"Reduced detector by shots: {detector_key} with number of shots: {np.shape(detector)}")
        if new_key:
            target_key=f"{detector_key}_summed"
        else:
            target_key=detector_key
        setattr(run, target_key, reduced_data)
        if purge:
            setattr(run, detector_key,None)
            run.update_status(f"Purged key to save room: {detector_key}")

    def reduce_detector_spatial(self, run, detector_key, shot_range=[0, None], rois=[[0, None]], reduction_function=np.sum,  purge=True, combine=True):
        """
        Reduces the spatial dimension of detector data based on specified ROIs.

        Parameters
        ----------

        run : spectroscopy_run
            The spectroscopy run instance.

        detector_key : str
            The key corresponding to the detector data.

        shot_range : list, optional
            The range of shots to consider (default is [0, None]).

        rois : list of lists, optional
            The list of ROIs (regions of interest) as pixel ranges (default is [[0, None]]).

        reduction_function : function, optional
            The function to apply for reduction (default is np.sum).

        purge : bool, optional
            Whether to purge the original detector data after reduction (default is True).

        combine : bool, optional
            Whether to combine ROIs (default is True).

        """
        detector = getattr(run, detector_key)
        if combine:

            roi_combined = [rois[0][0], rois[-1][1]]  # Combined ROI spanning the first and last ROI
            mask = np.zeros(detector.shape[-1], dtype=bool)
            for roi in rois:
                mask[roi[0]:roi[1]] = True
            if detector.ndim==3:
                masked_data = detector[shot_range[0]:shot_range[1], :, :][:, :, mask]
            elif detector.ndim==2:
                masked_data = detector[:, mask]
            elif detector.ndim==1:
                masked_data = detector[mask]
            reduced_data = reduction_function(masked_data, axis=-1)
            roi_indices = ', '.join([f"{roi[0]}-{roi[1]}" for roi in rois])
            run.update_status(f"Spatially reduced detector: {detector_key} with combined ROI indices: {roi_indices}")
            setattr(run, f"{detector_key}_ROI_1", reduced_data)
        else:
            for idx, roi in enumerate(rois):
                data_chunk = detector[shot_range[0]:shot_range[1], roi[0]:roi[1]]
                reduced_data = reduction_function(data_chunk, **kwargs)
            if roi[1] is None:
                roi[1] = detector.shape[1] - 1
                run.update_status(f"Spatially reduced detector: {detector_key} with ROI: {roi[0]}, {roi[1]}")
                setattr(run, f"{detector_key}_ROI_{idx+1}", reduced_data)
        if purge:
            #pass
            setattr(run, detector_key,None)
            #delattr(run, detector_key)
            #del run.detector_key
            run.update_status(f"Purged key after spatial reduction to save room: {detector_key}")

    def time_binning(self,run,bins,lxt_key='lxt_ttc',fast_delay_key='encoder',tt_correction_key='time_tool_correction'):
        """
        Bins data in time based on specified bins.

        Parameters
        ----------

        run : spectroscopy_run
            The spectroscopy run instance.

        bins : array-like
            The bins to use for time binning.

        lxt_key : str, optional
            The key for the laser time delay data (default is 'lxt_ttc').

        fast_delay_key : str, optional
            The key for the fast delay data (default is 'encoder').

        tt_correction_key : str, optional
            The key for the time tool correction data (default is 'time_tool_correction').

        """
        if lxt_key==None:
            run.delays = 0+ getattr(run,fast_delay_key)  + getattr(run,tt_correction_key)
        else:
            run.delays = getattr(run,lxt_key)*1.0e12 + getattr(run,fast_delay_key)  + getattr(run,tt_correction_key)
        run.time_bins=bins
        run.timing_bin_indices=np.digitize(run.delays, bins)[:]
        run.update_status('Generated timing bins from %f to %f in %d steps.' % (np.min(bins),np.max(bins),len(bins)))
    def union_shots(self, run, detector_key, filter_keys,new_key=True):
        """
        Combines shots across multiple filters into a single array. 
        So union_shots(f,'timing_bin_indices',['simultaneous','laser'])
        means go through the timing_bin_indices and find the ones that correspond to X-rays and laser shots.

        Parameters
        ----------

        run : spectroscopy_run
            The spectroscopy run instance.

        detector_key : str
            The key corresponding to the detector data.

        filter_keys : list of str
            The list of filter keys to combine.

        """
        detector = getattr(run, detector_key)

        if isinstance(filter_keys, list):
            mask = np.logical_and.reduce([getattr(run, k) for k in filter_keys])
        else:
            mask = getattr(run, filter_keys)
        filtered_detector = detector[mask]
        if new_key:
            target_key=detector_key + '_' + '_'.join(filter_keys)
        else:
            target_key=detector_key
        setattr(run, target_key, filtered_detector)
        run.update_status('Shots combined for detector %s on filters: %s and %s into %s'%(detector_key, filter_keys[0],filter_keys[1],target_key))

    def separate_shots(self, run, detector_key, filter_keys):
        """
        Separates shots into different datasets based on filters.
        separate_shots(f,'epix_ROI_1',['xray','laser']) means find me the epix_ROI_1 images in shots that were X-ray but NOT laser.
        If you wanted the inverse you would switch the order of the filter_keys.

        Parameters
        ----------

        run : spectroscopy_run
            The spectroscopy run instance.

        detector_key : str
            The key corresponding to the detector data.

        filter_keys : list of str
            The list of filter keys to separate.

        """
        detector = getattr(run, detector_key)
        if isinstance(filter_keys, list):
            mask1 = getattr(run, filter_keys[0])
            mask2 = np.logical_not(getattr(run, filter_keys[1]))
            mask = np.logical_and(mask1, mask2)
        else:
            mask = getattr(run, filter_keys)
        filtered_detector = detector[mask]
        setattr(run, detector_key + '_' +filter_keys[0]+'_not_'+filter_keys[1], filtered_detector)
        run.update_status('Shots (%d) separated for detector %s on filters: %s and %s into %s'%(np.sum(mask),detector_key,filter_keys[0],filter_keys[1],detector_key + '_' + '_'.join(filter_keys)))

    def reduce_detector_temporal(self, run, detector_key, timing_bin_key_indices,average=False):
        """
        Reduces the temporal dimension of detector data based on timing bins.

        Parameters
        ----------

        run : spectroscopy_run
            The spectroscopy run instance.

        detector_key : str
            The key corresponding to the detector data.

        timing_bin_key_indices : str
            The key corresponding to the timing bin indices.

        average : bool, optional
            Whether to average the data within each bin (default is False).

        """
        detector = getattr(run, detector_key)
        indices = getattr(run, timing_bin_key_indices)
        expected_length = len(run.time_bins)+1
        if len(detector.shape) < 2:
            reduced_array = np.zeros((expected_length))
        elif len(detector.shape) < 3:
            reduced_array = np.zeros((expected_length, detector.shape[1]))
        elif len(detector.shape) == 3:
            reduced_array = np.zeros((expected_length, detector.shape[1], detector.shape[2]))

        counts = np.bincount(indices)
        if average:
            np.add.at(reduced_array, indices, detector)
            reduced_array /= counts[:, None]
        else:
            np.add.at(reduced_array, indices, detector)
        setattr(run, detector_key+'_time_binned', reduced_array)
        run.update_status('Detector %s binned in time into key: %s from detector shape: %s to reduced shape: %s'%(detector_key,detector_key+'_time_binned', detector.shape,reduced_array.shape) )
    def patch_pixels(self,run,detector_key,  mode='average', patch_range=4, deg=1, poly_range=6,axis=1):
        """
        Patches multiple pixels in detector data.

        Parameters
        ----------

        run : spectroscopy_run
            The spectroscopy run instance.

        detector_key : str
            The key corresponding to the detector data.

        mode : str, optional
            The mode of patching ('average', 'polynomial', or 'interpolate').

        patch_range : int, optional
            The range around the pixel to use for patching (default is 4).

        deg : int, optional
            The degree of the polynomial for polynomial patching (default is 1).

        poly_range : int, optional
            The range of pixels to use for polynomial or interpolation patching (default is 6).

        axis : int, optional
            The axis along which to apply the patching (default is 1).

        """
        for pixel in self.pixels_to_patch:
            self.patch_pixel(run,detector_key,pixel,mode,patch_range,deg,poly_range,axis=axis)


    def patch_pixel(self, run, detector_key, pixel, mode='average', patch_range=4, deg=1, poly_range=6, axis=1):
        """
        EPIX detector pixel patching.
        TODO: extend to patch regions instead of per pixel.

        Parameters
        ----------

        data : array_like
            Array of shots

        pixel : integer
            Pixel point to be patched

        mode : string
            Determines which mode to use for patching the pixel. Averaging works well.

        patch_range : integer
            Pixels away from the pixel to be patched to be used for patching. Needed if multiple pixels in a row are an issue.

        deg : integer
            Degree of polynomial if polynomial patching is used.

        poly_range : integer
            Number of pixels to include in the polynomial or interpolation fitting

        Returns
        -------

        float
            The original data with the new patch values.

        """
        data = getattr(run, detector_key)

        def get_neighbor_values(data, pixel, patch_range, axis):
            axis_slice = [slice(None)] * data.ndim
            start_index = max(pixel - patch_range, 0)
            end_index = min(pixel + patch_range + 1, data.shape[axis])
            axis_slice[axis] = slice(start_index, end_index)
            return data[tuple(axis_slice)]

        def patch_value_average(data, pixel, patch_range, axis):
            neighbor_values = get_neighbor_values(data, pixel, patch_range, axis)
            neighbor_values = np.moveaxis(neighbor_values, axis, 0)
            new_val = np.mean(neighbor_values, axis=0)
            return new_val

        def patch_value_polynomial(data, pixel, patch_range, poly_range, deg, axis):
            patch_x = np.arange(pixel - patch_range - poly_range, pixel + patch_range + poly_range + 1)
            patch_range_weights = np.ones(len(patch_x))
            patch_range_weights[patch_range:-patch_range] = 0.001

            neighbor_values = get_neighbor_values(data, pixel, patch_range + poly_range, axis)
            neighbor_values = np.moveaxis(neighbor_values, axis, 0)

            new_vals = []
            for idx in range(neighbor_values.shape[1]): 
                ys = neighbor_values[:, idx]
                coeffs = np.polyfit(patch_x, ys, deg, w=patch_range_weights)
                new_vals.append(np.polyval(coeffs, pixel))
            return np.array(new_vals)

        def patch_value_interpolate(data, pixel, patch_range, poly_range, axis):
            patch_x = np.arange(pixel - patch_range - poly_range, pixel + patch_range + poly_range + 1)
            neighbor_values = get_neighbor_values(data, pixel, patch_range + poly_range, axis)
            neighbor_values = np.moveaxis(neighbor_values, axis, 0)

            new_vals = []
            for idx in range(neighbor_values.shape[1]):
                ys = neighbor_values[:, idx]
                interp_func = interp1d(patch_x, ys, kind='quadratic')
                new_vals.append(interp_func(pixel))
            return np.array(new_vals)

        if mode == 'average':
            new_val = patch_value_average(data, pixel, patch_range, axis)
        elif mode == 'polynomial':
            new_val = patch_value_polynomial(data, pixel, patch_range, poly_range, deg, axis)
        elif mode == 'interpolate':
            new_val = patch_value_interpolate(data, pixel, patch_range, poly_range, axis)
        else:
            raise ValueError(f"Unsupported mode: {mode}")

        patch_slice = [slice(None)] * data.ndim
        patch_slice[axis] = pixel
        data[tuple(patch_slice)] = new_val

        setattr(run, detector_key, data)
        run.update_status(f"Detector {detector_key} pixel {pixel} patched. Old value.")

    def patch_pixels_1d(self,run,detector_key,  mode='average', patch_range=4, deg=1, poly_range=6):
        """
        Patches multiple pixels in 1D detector data.

        Parameters
        ----------
        run : spectroscopy_run
            The spectroscopy run instance.
        detector_key : str
            The key corresponding to the detector data.
        mode : str, optional
            The mode of patching ('average', 'polynomial', or 'interpolate').
        patch_range : int, optional
            The range around the pixel to use for patching (default is 4).
        deg : int, optional
            The degree of the polynomial for polynomial patching (default is 1).
        poly_range : int, optional
            The range of pixels to use for polynomial or interpolation patching (default is 6).
        """
        for pixel in self.pixels_to_patch:
            self.patch_pixel_1d(run,detector_key,pixel,mode,patch_range,deg,poly_range)
    def patch_pixel_1d(self, run, detector_key, pixel, mode='average', patch_range=4, deg=1, poly_range=6):
        """
        EPIX detector pixel patching.
        TODO: extend to patch regions instead of per pixel.
        Parameters
        ----------
        data : array_like
            Array of shots
        pixel : integer
            Pixel point to be patched
        mode : string
            Determined which mode to use for patching the pixel. Averaging works well.
        patch_range : integer
            pixels away from the pixel to be patched to be used for patching. Needed if multiple pixels in a row are an issue.
        deg : integer
            Degree of polynomial if polynomial patching is used.
        poly_range : integer
            Number of pixels to include in the polynomial or interpolation fitting
        Returns
        -------
        float
            The original data with the new patch values.
        """
        data = getattr(run, detector_key)
        if mode == 'average':
            neighbor_values = data[:, pixel - patch_range:pixel + patch_range + 1]
            data[:, pixel] = np.sum(neighbor_values, axis=1) / neighbor_values.shape[1]
        elif mode == 'polynomial':
            patch_x = np.arange(pixel - patch_range - poly_range, pixel + patch_range + poly_range + 1, 1)
            patch_range_weights = np.ones(len(patch_x))
            patch_range_weights[pixel - patch_range - poly_range:pixel + patch_range + poly_range] = 0.001
            coeffs = np.polyfit(patch_x, data[pixel - patch_range - poly_range:pixel + patch_range + poly_range + 1], deg,
                                w=patch_range_weights)
            data[pixel, :] = np.polyval(coeffs, pixel)
        elif mode == 'interpolate':
            patch_x = np.arange(pixel - patch_range - poly_range, pixel + patch_range + poly_range + 1, 1)
            interp = interp1d(patch_x, data[pixel - patch_range - poly_range:pixel + patch_range + poly_range + 1, :],
                              kind='quadratic')
            data[pixel, :] = interp(pixel)
        setattr(run,detector_key,data)
        run.update_status('Detector %s pixel %d patched in mode %s'%(detector_key, pixel,mode ))

`bin_uniques(run, key)`

Bins unique values for a given key within a run.

Parameters:

Name	Type	Description	Default
`run`	`spectroscopy_run`	The spectroscopy run instance.	required
`key`	`str`	The key for which unique values are to be binned.	required

Source code in XSpect/XSpect_Analysis.py

def bin_uniques(self,run,key):
    """
    Bins unique values for a given key within a run.

    Parameters
    ----------

    run : spectroscopy_run
        The spectroscopy run instance.

    key : str
        The key for which unique values are to be binned.

    """
    vals = getattr(run,key)
    bins = np.unique(vals)
    addon = (bins[-1] - bins[-2])/2 # add on energy 
    bins2 = np.append(bins,bins[-1]+addon) # elist2 will be elist with dummy value at end
    bins_center = np.empty_like(bins2)
    for ii in np.arange(bins.shape[0]):
        if ii == 0:
            bins_center[ii] = bins2[ii] - (bins2[ii+1] - bins2[ii])/2
        else:
            bins_center[ii] = bins2[ii] - (bins2[ii] - bins2[ii-1])/2
    bins_center[-1] = bins2[-1]

    setattr(run,'scanvar_indices',np.digitize(vals,bins_center))
    setattr(run,'scanvar_bins',bins_center)

`filter_detector_adu(run, detector, adu_threshold=3.0)`

Filters is a misnomer compared to the other filter functions. This sets detector pixel values below a threshold to 0. Specifically, to remove 0-photon noise from detectors.

Parameters:

Name	Type	Description	Default
`run`	`spectroscopy_run`	The spectroscopy run instance.	required
`detector`	`str`	The key corresponding to the detector data.	required
`adu_threshold`	`float or list of float`	The ADU threshold for filtering. Can be a single value or a range (default is 3.0).	`3.0`

Returns:

Type	Description
`ndarray`	The filtered detector data.

Source code in XSpect/XSpect_Analysis.py

def filter_detector_adu(self,run,detector,adu_threshold=3.0):
    """
    Filters is a misnomer compared to the other filter functions. 
    This sets detector pixel values below a threshold to 0.
    Specifically, to remove 0-photon noise from detectors. 

    Parameters
    ----------

    run : spectroscopy_run
        The spectroscopy run instance.

    detector : str
        The key corresponding to the detector data.

    adu_threshold : float or list of float, optional
        The ADU threshold for filtering. Can be a single value or a range (default is 3.0).

    Returns
    -------

    np.ndarray
        The filtered detector data.

    """
    detector_images=getattr(run,detector)
    if isinstance(adu_threshold,list):
        detector_images_adu = detector_images * (detector_images > adu_threshold[0])
        detector_images_adu = detector_images_adu * (detector_images_adu < adu_threshold[1])
        run.update_status('Key: %s has been adu filtered by thresholds: %f,%f' % (detector,adu_threshold[0],adu_threshold[1]))
    else:
        detector_images_adu = detector_images * (detector_images > adu_threshold)
        run.update_status('Key: %s has been adu filtered by threshold: %f' % (detector,adu_threshold))

    setattr(run,detector,detector_images_adu)

    return detector_images_adu

`filter_nan(run, shot_mask_key, filter_key='ipm')`

A specific filtering implementation for Nans due to various DAQ issues. Filters out shots with NaN values in the specified filter.

Parameters:

Name	Type	Description	Default
`run`	`spectroscopy_run`	The spectroscopy run instance.	required
`shot_mask_key`	`str`	The key corresponding to the shot mask.	required
`filter_key`	`str`	The key corresponding to the filter data (default is 'ipm').	`'ipm'`

Source code in XSpect/XSpect_Analysis.py

def filter_nan(self, run,shot_mask_key, filter_key='ipm'):
    """
    A specific filtering implementation for Nans due to various DAQ issues. 
    Filters out shots with NaN values in the specified filter.

    Parameters
    ----------

    run : spectroscopy_run
        The spectroscopy run instance.

    shot_mask_key : str
        The key corresponding to the shot mask.

    filter_key : str, optional
        The key corresponding to the filter data (default is 'ipm').

    """
    shot_mask=getattr(run,shot_mask_key)
    count_before=np.sum(shot_mask)
    filter_mask=getattr(run,filter_key)
    filtered_shot_mask=shot_mask * (filter_mask>threshold)
    count_after=np.sum(filtered_shot_mask)
    setattr(run,shot_mask_key,filtered_shot_mask)
    run.update_status('Mask: %s has been filtered on %s by minimum threshold: %0.3f\nShots removed: %d' % (shot_mask_key,filter_key,threshold,count_before-count_after))

`filter_shots(run, shot_mask_key, filter_key='ipm', threshold=10000.0)`

Filters shots based on a given threshold. For example, if we filter: xray,ipm,1E4 then X-ray shots will be filtered out if the ipm is below 1E4.

Parameters:

Name	Type	Description	Default
`run`	`spectroscopy_run`	The spectroscopy run instance.	required
`shot_mask_key`	`str`	The key corresponding to the shot mask. An example being [xray,simultaneous,laser] for all x-ray shots	required
`filter_key`	`str`	The key corresponding to the filter data (default is 'ipm').	`'ipm'`
`threshold`	`float`	The threshold value for filtering (default is 1.0E4).	`10000.0`

Source code in XSpect/XSpect_Analysis.py

def filter_shots(self, run,shot_mask_key, filter_key='ipm', threshold=1.0E4):
    """
    Filters shots based on a given threshold.
    For example, if we filter: xray,ipm,1E4 then X-ray shots will be filtered out if the ipm is below 1E4.

    Parameters
    ----------

    run : spectroscopy_run
        The spectroscopy run instance.

    shot_mask_key : str
        The key corresponding to the shot mask. An example being [xray,simultaneous,laser] for all x-ray shots

    filter_key : str, optional
        The key corresponding to the filter data (default is 'ipm'). 

    threshold : float, optional
        The threshold value for filtering (default is 1.0E4).

    """
    shot_mask=getattr(run,shot_mask_key)
    count_before=np.sum(shot_mask)
    filter_mask=getattr(run,filter_key)
    nan_mask = np.isnan(filter_mask)
    filtered_shot_mask=shot_mask * (filter_mask>threshold)* (~nan_mask)
    count_after=np.sum(filtered_shot_mask)
    setattr(run,shot_mask_key,filtered_shot_mask)
    run.update_status('Mask: %s has been filtered on %s by minimum threshold: %0.3f\nShots removed: %d' % (shot_mask_key,filter_key,threshold,count_before-count_after))

`patch_pixel(run, detector_key, pixel, mode='average', patch_range=4, deg=1, poly_range=6, axis=1)`

EPIX detector pixel patching. TODO: extend to patch regions instead of per pixel.

Parameters:

Name	Type	Description	Default
`data`	`array_like`	Array of shots	required
`pixel`	`integer`	Pixel point to be patched	required
`mode`	`string`	Determines which mode to use for patching the pixel. Averaging works well.	`'average'`
`patch_range`	`integer`	Pixels away from the pixel to be patched to be used for patching. Needed if multiple pixels in a row are an issue.	`4`
`deg`	`integer`	Degree of polynomial if polynomial patching is used.	`1`
`poly_range`	`integer`	Number of pixels to include in the polynomial or interpolation fitting	`6`

Returns:

Type	Description
`float`	The original data with the new patch values.

Source code in XSpect/XSpect_Analysis.py

def patch_pixel(self, run, detector_key, pixel, mode='average', patch_range=4, deg=1, poly_range=6, axis=1):
    """
    EPIX detector pixel patching.
    TODO: extend to patch regions instead of per pixel.

    Parameters
    ----------

    data : array_like
        Array of shots

    pixel : integer
        Pixel point to be patched

    mode : string
        Determines which mode to use for patching the pixel. Averaging works well.

    patch_range : integer
        Pixels away from the pixel to be patched to be used for patching. Needed if multiple pixels in a row are an issue.

    deg : integer
        Degree of polynomial if polynomial patching is used.

    poly_range : integer
        Number of pixels to include in the polynomial or interpolation fitting

    Returns
    -------

    float
        The original data with the new patch values.

    """
    data = getattr(run, detector_key)

    def get_neighbor_values(data, pixel, patch_range, axis):
        axis_slice = [slice(None)] * data.ndim
        start_index = max(pixel - patch_range, 0)
        end_index = min(pixel + patch_range + 1, data.shape[axis])
        axis_slice[axis] = slice(start_index, end_index)
        return data[tuple(axis_slice)]

    def patch_value_average(data, pixel, patch_range, axis):
        neighbor_values = get_neighbor_values(data, pixel, patch_range, axis)
        neighbor_values = np.moveaxis(neighbor_values, axis, 0)
        new_val = np.mean(neighbor_values, axis=0)
        return new_val

    def patch_value_polynomial(data, pixel, patch_range, poly_range, deg, axis):
        patch_x = np.arange(pixel - patch_range - poly_range, pixel + patch_range + poly_range + 1)
        patch_range_weights = np.ones(len(patch_x))
        patch_range_weights[patch_range:-patch_range] = 0.001

        neighbor_values = get_neighbor_values(data, pixel, patch_range + poly_range, axis)
        neighbor_values = np.moveaxis(neighbor_values, axis, 0)

        new_vals = []
        for idx in range(neighbor_values.shape[1]): 
            ys = neighbor_values[:, idx]
            coeffs = np.polyfit(patch_x, ys, deg, w=patch_range_weights)
            new_vals.append(np.polyval(coeffs, pixel))
        return np.array(new_vals)

    def patch_value_interpolate(data, pixel, patch_range, poly_range, axis):
        patch_x = np.arange(pixel - patch_range - poly_range, pixel + patch_range + poly_range + 1)
        neighbor_values = get_neighbor_values(data, pixel, patch_range + poly_range, axis)
        neighbor_values = np.moveaxis(neighbor_values, axis, 0)

        new_vals = []
        for idx in range(neighbor_values.shape[1]):
            ys = neighbor_values[:, idx]
            interp_func = interp1d(patch_x, ys, kind='quadratic')
            new_vals.append(interp_func(pixel))
        return np.array(new_vals)

    if mode == 'average':
        new_val = patch_value_average(data, pixel, patch_range, axis)
    elif mode == 'polynomial':
        new_val = patch_value_polynomial(data, pixel, patch_range, poly_range, deg, axis)
    elif mode == 'interpolate':
        new_val = patch_value_interpolate(data, pixel, patch_range, poly_range, axis)
    else:
        raise ValueError(f"Unsupported mode: {mode}")

    patch_slice = [slice(None)] * data.ndim
    patch_slice[axis] = pixel
    data[tuple(patch_slice)] = new_val

    setattr(run, detector_key, data)
    run.update_status(f"Detector {detector_key} pixel {pixel} patched. Old value.")

`patch_pixel_1d(run, detector_key, pixel, mode='average', patch_range=4, deg=1, poly_range=6)`

EPIX detector pixel patching. TODO: extend to patch regions instead of per pixel.

Parameters:

Name	Type	Description	Default
`data`	`array_like`	Array of shots	required
`pixel`	`integer`	Pixel point to be patched	required
`mode`	`string`	Determined which mode to use for patching the pixel. Averaging works well.	`'average'`
`patch_range`	`integer`	pixels away from the pixel to be patched to be used for patching. Needed if multiple pixels in a row are an issue.	`4`
`deg`	`integer`	Degree of polynomial if polynomial patching is used.	`1`
`poly_range`	`integer`	Number of pixels to include in the polynomial or interpolation fitting	`6`

Returns:

Type	Description
`float`	The original data with the new patch values.

Source code in XSpect/XSpect_Analysis.py

def patch_pixel_1d(self, run, detector_key, pixel, mode='average', patch_range=4, deg=1, poly_range=6):
    """
    EPIX detector pixel patching.
    TODO: extend to patch regions instead of per pixel.
    Parameters
    ----------
    data : array_like
        Array of shots
    pixel : integer
        Pixel point to be patched
    mode : string
        Determined which mode to use for patching the pixel. Averaging works well.
    patch_range : integer
        pixels away from the pixel to be patched to be used for patching. Needed if multiple pixels in a row are an issue.
    deg : integer
        Degree of polynomial if polynomial patching is used.
    poly_range : integer
        Number of pixels to include in the polynomial or interpolation fitting
    Returns
    -------
    float
        The original data with the new patch values.
    """
    data = getattr(run, detector_key)
    if mode == 'average':
        neighbor_values = data[:, pixel - patch_range:pixel + patch_range + 1]
        data[:, pixel] = np.sum(neighbor_values, axis=1) / neighbor_values.shape[1]
    elif mode == 'polynomial':
        patch_x = np.arange(pixel - patch_range - poly_range, pixel + patch_range + poly_range + 1, 1)
        patch_range_weights = np.ones(len(patch_x))
        patch_range_weights[pixel - patch_range - poly_range:pixel + patch_range + poly_range] = 0.001
        coeffs = np.polyfit(patch_x, data[pixel - patch_range - poly_range:pixel + patch_range + poly_range + 1], deg,
                            w=patch_range_weights)
        data[pixel, :] = np.polyval(coeffs, pixel)
    elif mode == 'interpolate':
        patch_x = np.arange(pixel - patch_range - poly_range, pixel + patch_range + poly_range + 1, 1)
        interp = interp1d(patch_x, data[pixel - patch_range - poly_range:pixel + patch_range + poly_range + 1, :],
                          kind='quadratic')
        data[pixel, :] = interp(pixel)
    setattr(run,detector_key,data)
    run.update_status('Detector %s pixel %d patched in mode %s'%(detector_key, pixel,mode ))

`patch_pixels(run, detector_key, mode='average', patch_range=4, deg=1, poly_range=6, axis=1)`

Patches multiple pixels in detector data.

Parameters:

Name	Type	Description	Default
`run`	`spectroscopy_run`	The spectroscopy run instance.	required
`detector_key`	`str`	The key corresponding to the detector data.	required
`mode`	`str`	The mode of patching ('average', 'polynomial', or 'interpolate').	`'average'`
`patch_range`	`int`	The range around the pixel to use for patching (default is 4).	`4`
`deg`	`int`	The degree of the polynomial for polynomial patching (default is 1).	`1`
`poly_range`	`int`	The range of pixels to use for polynomial or interpolation patching (default is 6).	`6`
`axis`	`int`	The axis along which to apply the patching (default is 1).	`1`

Source code in XSpect/XSpect_Analysis.py

def patch_pixels(self,run,detector_key,  mode='average', patch_range=4, deg=1, poly_range=6,axis=1):
    """
    Patches multiple pixels in detector data.

    Parameters
    ----------

    run : spectroscopy_run
        The spectroscopy run instance.

    detector_key : str
        The key corresponding to the detector data.

    mode : str, optional
        The mode of patching ('average', 'polynomial', or 'interpolate').

    patch_range : int, optional
        The range around the pixel to use for patching (default is 4).

    deg : int, optional
        The degree of the polynomial for polynomial patching (default is 1).

    poly_range : int, optional
        The range of pixels to use for polynomial or interpolation patching (default is 6).

    axis : int, optional
        The axis along which to apply the patching (default is 1).

    """
    for pixel in self.pixels_to_patch:
        self.patch_pixel(run,detector_key,pixel,mode,patch_range,deg,poly_range,axis=axis)

`patch_pixels_1d(run, detector_key, mode='average', patch_range=4, deg=1, poly_range=6)`

Patches multiple pixels in 1D detector data.

Parameters:

Name	Type	Description	Default
`run`	`spectroscopy_run`	The spectroscopy run instance.	required
`detector_key`	`str`	The key corresponding to the detector data.	required
`mode`	`str`	The mode of patching ('average', 'polynomial', or 'interpolate').	`'average'`
`patch_range`	`int`	The range around the pixel to use for patching (default is 4).	`4`
`deg`	`int`	The degree of the polynomial for polynomial patching (default is 1).	`1`
`poly_range`	`int`	The range of pixels to use for polynomial or interpolation patching (default is 6).	`6`

Source code in XSpect/XSpect_Analysis.py

def patch_pixels_1d(self,run,detector_key,  mode='average', patch_range=4, deg=1, poly_range=6):
    """
    Patches multiple pixels in 1D detector data.

    Parameters
    ----------
    run : spectroscopy_run
        The spectroscopy run instance.
    detector_key : str
        The key corresponding to the detector data.
    mode : str, optional
        The mode of patching ('average', 'polynomial', or 'interpolate').
    patch_range : int, optional
        The range around the pixel to use for patching (default is 4).
    deg : int, optional
        The degree of the polynomial for polynomial patching (default is 1).
    poly_range : int, optional
        The range of pixels to use for polynomial or interpolation patching (default is 6).
    """
    for pixel in self.pixels_to_patch:
        self.patch_pixel_1d(run,detector_key,pixel,mode,patch_range,deg,poly_range)

`purge_keys(run, keys)`

Purges specific keys from the run to save memory. This is specifically to remove the epix key immediately after processing it from the hdf5 file. To avoid OOM. This is different than the purge all keys method which is used to purge many of the larger analysis steps.

Parameters:

Name	Type	Description	Default
`run`	`spectroscopy_run`	The spectroscopy run instance.	required
`keys`	`list of str`	The list of keys to purge.	required

Source code in XSpect/XSpect_Analysis.py

def purge_keys(self,run,keys):
    """
    Purges specific keys from the run to save memory.
    This is specifically to remove the epix key immediately after processing it from the hdf5 file.
    To avoid OOM. This is different than the purge all keys method which is used to purge many of the larger analysis steps.

    Parameters
    ----------

    run : spectroscopy_run
        The spectroscopy run instance.

    keys : list of str
        The list of keys to purge.

    """
    for detector_key in keys:
        setattr(run, detector_key, None)
        run.update_status(f"Purged key to save room: {detector_key}")

`reduce_detector_spatial(run, detector_key, shot_range=[0, None], rois=[[0, None]], reduction_function=np.sum, purge=True, combine=True)`

Reduces the spatial dimension of detector data based on specified ROIs.

Parameters:

Name	Type	Description	Default
`run`	`spectroscopy_run`	The spectroscopy run instance.	required
`detector_key`	`str`	The key corresponding to the detector data.	required
`shot_range`	`list`	The range of shots to consider (default is [0, None]).	`[0, None]`
`rois`	`list of lists`	The list of ROIs (regions of interest) as pixel ranges (default is [[0, None]]).	`[[0, None]]`
`reduction_function`	`function`	The function to apply for reduction (default is np.sum).	`sum`
`purge`	`bool`	Whether to purge the original detector data after reduction (default is True).	`True`
`combine`	`bool`	Whether to combine ROIs (default is True).	`True`

Source code in XSpect/XSpect_Analysis.py

def reduce_detector_spatial(self, run, detector_key, shot_range=[0, None], rois=[[0, None]], reduction_function=np.sum,  purge=True, combine=True):
    """
    Reduces the spatial dimension of detector data based on specified ROIs.

    Parameters
    ----------

    run : spectroscopy_run
        The spectroscopy run instance.

    detector_key : str
        The key corresponding to the detector data.

    shot_range : list, optional
        The range of shots to consider (default is [0, None]).

    rois : list of lists, optional
        The list of ROIs (regions of interest) as pixel ranges (default is [[0, None]]).

    reduction_function : function, optional
        The function to apply for reduction (default is np.sum).

    purge : bool, optional
        Whether to purge the original detector data after reduction (default is True).

    combine : bool, optional
        Whether to combine ROIs (default is True).

    """
    detector = getattr(run, detector_key)
    if combine:

        roi_combined = [rois[0][0], rois[-1][1]]  # Combined ROI spanning the first and last ROI
        mask = np.zeros(detector.shape[-1], dtype=bool)
        for roi in rois:
            mask[roi[0]:roi[1]] = True
        if detector.ndim==3:
            masked_data = detector[shot_range[0]:shot_range[1], :, :][:, :, mask]
        elif detector.ndim==2:
            masked_data = detector[:, mask]
        elif detector.ndim==1:
            masked_data = detector[mask]
        reduced_data = reduction_function(masked_data, axis=-1)
        roi_indices = ', '.join([f"{roi[0]}-{roi[1]}" for roi in rois])
        run.update_status(f"Spatially reduced detector: {detector_key} with combined ROI indices: {roi_indices}")
        setattr(run, f"{detector_key}_ROI_1", reduced_data)
    else:
        for idx, roi in enumerate(rois):
            data_chunk = detector[shot_range[0]:shot_range[1], roi[0]:roi[1]]
            reduced_data = reduction_function(data_chunk, **kwargs)
        if roi[1] is None:
            roi[1] = detector.shape[1] - 1
            run.update_status(f"Spatially reduced detector: {detector_key} with ROI: {roi[0]}, {roi[1]}")
            setattr(run, f"{detector_key}_ROI_{idx+1}", reduced_data)
    if purge:
        #pass
        setattr(run, detector_key,None)
        #delattr(run, detector_key)
        #del run.detector_key
        run.update_status(f"Purged key after spatial reduction to save room: {detector_key}")

`reduce_detector_temporal(run, detector_key, timing_bin_key_indices, average=False)`

Reduces the temporal dimension of detector data based on timing bins.

Parameters:

Name	Type	Description	Default
`run`	`spectroscopy_run`	The spectroscopy run instance.	required
`detector_key`	`str`	The key corresponding to the detector data.	required
`timing_bin_key_indices`	`str`	The key corresponding to the timing bin indices.	required
`average`	`bool`	Whether to average the data within each bin (default is False).	`False`

Source code in XSpect/XSpect_Analysis.py

def reduce_detector_temporal(self, run, detector_key, timing_bin_key_indices,average=False):
    """
    Reduces the temporal dimension of detector data based on timing bins.

    Parameters
    ----------

    run : spectroscopy_run
        The spectroscopy run instance.

    detector_key : str
        The key corresponding to the detector data.

    timing_bin_key_indices : str
        The key corresponding to the timing bin indices.

    average : bool, optional
        Whether to average the data within each bin (default is False).

    """
    detector = getattr(run, detector_key)
    indices = getattr(run, timing_bin_key_indices)
    expected_length = len(run.time_bins)+1
    if len(detector.shape) < 2:
        reduced_array = np.zeros((expected_length))
    elif len(detector.shape) < 3:
        reduced_array = np.zeros((expected_length, detector.shape[1]))
    elif len(detector.shape) == 3:
        reduced_array = np.zeros((expected_length, detector.shape[1], detector.shape[2]))

    counts = np.bincount(indices)
    if average:
        np.add.at(reduced_array, indices, detector)
        reduced_array /= counts[:, None]
    else:
        np.add.at(reduced_array, indices, detector)
    setattr(run, detector_key+'_time_binned', reduced_array)
    run.update_status('Detector %s binned in time into key: %s from detector shape: %s to reduced shape: %s'%(detector_key,detector_key+'_time_binned', detector.shape,reduced_array.shape) )

`separate_shots(run, detector_key, filter_keys)`

Separates shots into different datasets based on filters. separate_shots(f,'epix_ROI_1',['xray','laser']) means find me the epix_ROI_1 images in shots that were X-ray but NOT laser. If you wanted the inverse you would switch the order of the filter_keys.

Parameters:

Name	Type	Description	Default
`run`	`spectroscopy_run`	The spectroscopy run instance.	required
`detector_key`	`str`	The key corresponding to the detector data.	required
`filter_keys`	`list of str`	The list of filter keys to separate.	required

Source code in XSpect/XSpect_Analysis.py

def separate_shots(self, run, detector_key, filter_keys):
    """
    Separates shots into different datasets based on filters.
    separate_shots(f,'epix_ROI_1',['xray','laser']) means find me the epix_ROI_1 images in shots that were X-ray but NOT laser.
    If you wanted the inverse you would switch the order of the filter_keys.

    Parameters
    ----------

    run : spectroscopy_run
        The spectroscopy run instance.

    detector_key : str
        The key corresponding to the detector data.

    filter_keys : list of str
        The list of filter keys to separate.

    """
    detector = getattr(run, detector_key)
    if isinstance(filter_keys, list):
        mask1 = getattr(run, filter_keys[0])
        mask2 = np.logical_not(getattr(run, filter_keys[1]))
        mask = np.logical_and(mask1, mask2)
    else:
        mask = getattr(run, filter_keys)
    filtered_detector = detector[mask]
    setattr(run, detector_key + '_' +filter_keys[0]+'_not_'+filter_keys[1], filtered_detector)
    run.update_status('Shots (%d) separated for detector %s on filters: %s and %s into %s'%(np.sum(mask),detector_key,filter_keys[0],filter_keys[1],detector_key + '_' + '_'.join(filter_keys)))

`time_binning(run, bins, lxt_key='lxt_ttc', fast_delay_key='encoder', tt_correction_key='time_tool_correction')`

Bins data in time based on specified bins.

Parameters:

Name	Type	Description	Default
`run`	`spectroscopy_run`	The spectroscopy run instance.	required
`bins`	`array - like`	The bins to use for time binning.	required
`lxt_key`	`str`	The key for the laser time delay data (default is 'lxt_ttc').	`'lxt_ttc'`
`fast_delay_key`	`str`	The key for the fast delay data (default is 'encoder').	`'encoder'`
`tt_correction_key`	`str`	The key for the time tool correction data (default is 'time_tool_correction').	`'time_tool_correction'`

Source code in XSpect/XSpect_Analysis.py

def time_binning(self,run,bins,lxt_key='lxt_ttc',fast_delay_key='encoder',tt_correction_key='time_tool_correction'):
    """
    Bins data in time based on specified bins.

    Parameters
    ----------

    run : spectroscopy_run
        The spectroscopy run instance.

    bins : array-like
        The bins to use for time binning.

    lxt_key : str, optional
        The key for the laser time delay data (default is 'lxt_ttc').

    fast_delay_key : str, optional
        The key for the fast delay data (default is 'encoder').

    tt_correction_key : str, optional
        The key for the time tool correction data (default is 'time_tool_correction').

    """
    if lxt_key==None:
        run.delays = 0+ getattr(run,fast_delay_key)  + getattr(run,tt_correction_key)
    else:
        run.delays = getattr(run,lxt_key)*1.0e12 + getattr(run,fast_delay_key)  + getattr(run,tt_correction_key)
    run.time_bins=bins
    run.timing_bin_indices=np.digitize(run.delays, bins)[:]
    run.update_status('Generated timing bins from %f to %f in %d steps.' % (np.min(bins),np.max(bins),len(bins)))

`union_shots(run, detector_key, filter_keys, new_key=True)`

Combines shots across multiple filters into a single array. So union_shots(f,'timing_bin_indices',['simultaneous','laser']) means go through the timing_bin_indices and find the ones that correspond to X-rays and laser shots.

Parameters:

Name	Type	Description	Default
`run`	`spectroscopy_run`	The spectroscopy run instance.	required
`detector_key`	`str`	The key corresponding to the detector data.	required
`filter_keys`	`list of str`	The list of filter keys to combine.	required

Source code in XSpect/XSpect_Analysis.py

def union_shots(self, run, detector_key, filter_keys,new_key=True):
    """
    Combines shots across multiple filters into a single array. 
    So union_shots(f,'timing_bin_indices',['simultaneous','laser'])
    means go through the timing_bin_indices and find the ones that correspond to X-rays and laser shots.

    Parameters
    ----------

    run : spectroscopy_run
        The spectroscopy run instance.

    detector_key : str
        The key corresponding to the detector data.

    filter_keys : list of str
        The list of filter keys to combine.

    """
    detector = getattr(run, detector_key)

    if isinstance(filter_keys, list):
        mask = np.logical_and.reduce([getattr(run, k) for k in filter_keys])
    else:
        mask = getattr(run, filter_keys)
    filtered_detector = detector[mask]
    if new_key:
        target_key=detector_key + '_' + '_'.join(filter_keys)
    else:
        target_key=detector_key
    setattr(run, target_key, filtered_detector)
    run.update_status('Shots combined for detector %s on filters: %s and %s into %s'%(detector_key, filter_keys[0],filter_keys[1],target_key))

`XASAnalysis`

Bases: SpectroscopyAnalysis

Source code in XSpect/XSpect_Analysis.py

class XASAnalysis(SpectroscopyAnalysis):
    def __init__(self):
        pass;
    def trim_ccm(self,run,threshold=120):
        """
        Trim CCM values to remove bins with fewer shots than a specified threshold.

        Parameters
        ----------

        run : object
            The spectroscopy run instance.

        threshold : int, optional
            The minimum number of shots required to keep a CCM value (default is 120).

        """

        ccm_bins=getattr(run,'ccm_bins',elist_center)
        ccm_energies=getattr(run,'ccm_energies',elist)
        counts = np.bincount(bins)
        trimmed_ccm=ccm_energies[counts[:-1]>120]
        self.make_ccm_axis(run,ccm_energies)

    def make_ccm_axis(self,run,energies):
        """
        Generate CCM bins and centers from given energy values.

        Parameters
        ----------

        run : object
            The spectroscopy run instance.

        energies : array-like
            Array of energy values to be used for creating CCM bins.

        """
        elist=energies
#         addon = (elist[-1] - elist[-2])/2 # add on energy 
#         elist2 = np.append(elist,elist[-1]+addon) # elist2 will be elist with dummy value at end
#         elist_center = np.empty_like(elist2)
#         for ii in np.arange(elist.shape[0]):
#             if ii == 0:
#                 elist_center[ii] = elist2[ii] - (elist2[ii+1] - elist2[ii])/2
#             else:
#                 elist_center[ii] = elist2[ii] - (elist2[ii] - elist2[ii-1])/2
#                 elist_center[-1] = elist2[-1]
        addon = (elist[-1] - elist[-2])/2
        elist2 = np.append(elist,elist[-1]+addon)
        elist_center = np.empty_like(elist)

        for ii in np.arange(elist_center.shape[0]):
            if ii == elist_center.shape[0]:
                elist_center[ii] = elist[-1]+addon
            else:
                elist_center[ii] = elist2[ii+1] - (elist2[ii+1] - elist2[ii])/2    

        setattr(run,'ccm_bins',elist_center)
        setattr(run,'ccm_energies',elist)
    def reduce_detector_ccm_temporal(self, run, detector_key, timing_bin_key_indices,ccm_bin_key_indices,average=True):
        """
        Reduce detector data temporally and by CCM bins.

        Parameters
        ----------
        run : object
            The spectroscopy run instance.
        detector_key : str
            The key corresponding to the detector data.
        timing_bin_key_indices : str
            The key corresponding to the timing bin indices.
        ccm_bin_key_indices : str
            The key corresponding to the CCM bin indices.
        average : bool, optional
            Whether to average the reduced data (default is True).
        """
        detector = getattr(run, detector_key)
        timing_indices = getattr(run, timing_bin_key_indices)#digitized indices from detector
        ccm_indices = getattr(run, ccm_bin_key_indices)#digitized indices from detector
        reduced_array = np.zeros((np.shape(run.time_bins)[0]+1, np.shape(run.ccm_bins)[0]))
        unique_indices =np.column_stack((timing_indices, ccm_indices))
        np.add.at(reduced_array, (unique_indices[:, 0], unique_indices[:, 1]), detector)
        reduced_array = reduced_array[:-1,:]
        setattr(run, detector_key+'_time_energy_binned', reduced_array)
        run.update_status('Detector %s binned in time into key: %s'%(detector_key,detector_key+'_time_energy_binned') )

    def reduce_detector_ccm(self, run, detector_key, ccm_bin_key_indices, average = False, not_ccm=False):
        """
        Reduce detector data by CCM bins.

        Parameters
        ----------

        run : object
            The spectroscopy run instance.

        detector_key : str
            The key corresponding to the detector data.

        ccm_bin_key_indices : str
            The key corresponding to the CCM bin indices.

        average : bool, optional
            Whether to average the reduced data (default is False).

        not_ccm : bool, optional
            Whether to indicate that CCM is not being used (default is False).

        """
        detector = getattr(run, detector_key)

        ccm_indices = getattr(run, ccm_bin_key_indices)#digitized indices from detector
        if not_ccm:
            reduced_array = np.zeros(np.max(ccm_indices)+1 )
        else:
            reduced_array = np.zeros(np.shape(run.ccm_bins)[0]) 
        np.add.at(reduced_array, ccm_indices, detector)
        setattr(run, detector_key+'_energy_binned', reduced_array)

        run.update_status('Detector %s binned in energy into key: %s'%(detector_key,detector_key+'_energy_binned') )

    def reduce_detector_temporal(self, run, detector_key, timing_bin_key_indices, average=False):
        """
        Reduce detector data temporally. Specifically the 1d detector output for XAS data.

        Parameters
        ----------

        run : object
            The spectroscopy run instance.

        detector_key : str
            The key corresponding to the detector data.

        timing_bin_key_indices : str
            The key corresponding to the timing bin indices.

        average : bool, optional
            Whether to average the reduced data (default is False).

        """
        detector = getattr(run, detector_key)
        time_bins=run.time_bins
        timing_indices = getattr(run, timing_bin_key_indices)#digitized indices from detector
        reduced_array = np.zeros(np.shape(time_bins)[0]+1)
        np.add.at(reduced_array, timing_indices, detector)
        setattr(run, detector_key+'_time_binned', reduced_array)
        run.update_status('Detector %s binned in time into key: %s'%(detector_key,detector_key+'_time_binned') )

    def ccm_binning(self,run,ccm_bins_key,ccm_key='ccm'):
        """
        Generate CCM bin indices from CCM data and bins.

        Parameters
        ----------

        run : object
            The spectroscopy run instance.

        ccm_bins_key : str
            The key corresponding to the CCM bins.

        ccm_key : str, optional
            The key corresponding to the CCM data (default is 'ccm').

        """
        ccm=getattr(run,ccm_key)
        bins=getattr(run,ccm_bins_key)
        run.ccm_bin_indices=np.digitize(ccm, bins)
        run.update_status('Generated ccm bins from %f to %f in %d steps.' % (np.min(bins),np.max(bins),len(bins)))

`ccm_binning(run, ccm_bins_key, ccm_key='ccm')`

Generate CCM bin indices from CCM data and bins.

Parameters:

Name	Type	Description	Default
`run`	`object`	The spectroscopy run instance.	required
`ccm_bins_key`	`str`	The key corresponding to the CCM bins.	required
`ccm_key`	`str`	The key corresponding to the CCM data (default is 'ccm').	`'ccm'`

Source code in XSpect/XSpect_Analysis.py

def ccm_binning(self,run,ccm_bins_key,ccm_key='ccm'):
    """
    Generate CCM bin indices from CCM data and bins.

    Parameters
    ----------

    run : object
        The spectroscopy run instance.

    ccm_bins_key : str
        The key corresponding to the CCM bins.

    ccm_key : str, optional
        The key corresponding to the CCM data (default is 'ccm').

    """
    ccm=getattr(run,ccm_key)
    bins=getattr(run,ccm_bins_key)
    run.ccm_bin_indices=np.digitize(ccm, bins)
    run.update_status('Generated ccm bins from %f to %f in %d steps.' % (np.min(bins),np.max(bins),len(bins)))

`make_ccm_axis(run, energies)`

Generate CCM bins and centers from given energy values.

Parameters:

Name	Type	Description	Default
`run`	`object`	The spectroscopy run instance.	required
`energies`	`array - like`	Array of energy values to be used for creating CCM bins.	required

Source code in XSpect/XSpect_Analysis.py

    def make_ccm_axis(self,run,energies):
        """
        Generate CCM bins and centers from given energy values.

        Parameters
        ----------

        run : object
            The spectroscopy run instance.

        energies : array-like
            Array of energy values to be used for creating CCM bins.

        """
        elist=energies
#         addon = (elist[-1] - elist[-2])/2 # add on energy 
#         elist2 = np.append(elist,elist[-1]+addon) # elist2 will be elist with dummy value at end
#         elist_center = np.empty_like(elist2)
#         for ii in np.arange(elist.shape[0]):
#             if ii == 0:
#                 elist_center[ii] = elist2[ii] - (elist2[ii+1] - elist2[ii])/2
#             else:
#                 elist_center[ii] = elist2[ii] - (elist2[ii] - elist2[ii-1])/2
#                 elist_center[-1] = elist2[-1]
        addon = (elist[-1] - elist[-2])/2
        elist2 = np.append(elist,elist[-1]+addon)
        elist_center = np.empty_like(elist)

        for ii in np.arange(elist_center.shape[0]):
            if ii == elist_center.shape[0]:
                elist_center[ii] = elist[-1]+addon
            else:
                elist_center[ii] = elist2[ii+1] - (elist2[ii+1] - elist2[ii])/2    

        setattr(run,'ccm_bins',elist_center)
        setattr(run,'ccm_energies',elist)

`reduce_detector_ccm(run, detector_key, ccm_bin_key_indices, average=False, not_ccm=False)`

Reduce detector data by CCM bins.

Parameters:

Name	Type	Description	Default
`run`	`object`	The spectroscopy run instance.	required
`detector_key`	`str`	The key corresponding to the detector data.	required
`ccm_bin_key_indices`	`str`	The key corresponding to the CCM bin indices.	required
`average`	`bool`	Whether to average the reduced data (default is False).	`False`
`not_ccm`	`bool`	Whether to indicate that CCM is not being used (default is False).	`False`

Source code in XSpect/XSpect_Analysis.py

def reduce_detector_ccm(self, run, detector_key, ccm_bin_key_indices, average = False, not_ccm=False):
    """
    Reduce detector data by CCM bins.

    Parameters
    ----------

    run : object
        The spectroscopy run instance.

    detector_key : str
        The key corresponding to the detector data.

    ccm_bin_key_indices : str
        The key corresponding to the CCM bin indices.

    average : bool, optional
        Whether to average the reduced data (default is False).

    not_ccm : bool, optional
        Whether to indicate that CCM is not being used (default is False).

    """
    detector = getattr(run, detector_key)

    ccm_indices = getattr(run, ccm_bin_key_indices)#digitized indices from detector
    if not_ccm:
        reduced_array = np.zeros(np.max(ccm_indices)+1 )
    else:
        reduced_array = np.zeros(np.shape(run.ccm_bins)[0]) 
    np.add.at(reduced_array, ccm_indices, detector)
    setattr(run, detector_key+'_energy_binned', reduced_array)

    run.update_status('Detector %s binned in energy into key: %s'%(detector_key,detector_key+'_energy_binned') )

`reduce_detector_ccm_temporal(run, detector_key, timing_bin_key_indices, ccm_bin_key_indices, average=True)`

Reduce detector data temporally and by CCM bins.

Parameters:

Name	Type	Description	Default
`run`	`object`	The spectroscopy run instance.	required
`detector_key`	`str`	The key corresponding to the detector data.	required
`timing_bin_key_indices`	`str`	The key corresponding to the timing bin indices.	required
`ccm_bin_key_indices`	`str`	The key corresponding to the CCM bin indices.	required
`average`	`bool`	Whether to average the reduced data (default is True).	`True`

Source code in XSpect/XSpect_Analysis.py

def reduce_detector_ccm_temporal(self, run, detector_key, timing_bin_key_indices,ccm_bin_key_indices,average=True):
    """
    Reduce detector data temporally and by CCM bins.

    Parameters
    ----------
    run : object
        The spectroscopy run instance.
    detector_key : str
        The key corresponding to the detector data.
    timing_bin_key_indices : str
        The key corresponding to the timing bin indices.
    ccm_bin_key_indices : str
        The key corresponding to the CCM bin indices.
    average : bool, optional
        Whether to average the reduced data (default is True).
    """
    detector = getattr(run, detector_key)
    timing_indices = getattr(run, timing_bin_key_indices)#digitized indices from detector
    ccm_indices = getattr(run, ccm_bin_key_indices)#digitized indices from detector
    reduced_array = np.zeros((np.shape(run.time_bins)[0]+1, np.shape(run.ccm_bins)[0]))
    unique_indices =np.column_stack((timing_indices, ccm_indices))
    np.add.at(reduced_array, (unique_indices[:, 0], unique_indices[:, 1]), detector)
    reduced_array = reduced_array[:-1,:]
    setattr(run, detector_key+'_time_energy_binned', reduced_array)
    run.update_status('Detector %s binned in time into key: %s'%(detector_key,detector_key+'_time_energy_binned') )

`reduce_detector_temporal(run, detector_key, timing_bin_key_indices, average=False)`

Reduce detector data temporally. Specifically the 1d detector output for XAS data.

Parameters:

Name	Type	Description	Default
`run`	`object`	The spectroscopy run instance.	required
`detector_key`	`str`	The key corresponding to the detector data.	required
`timing_bin_key_indices`	`str`	The key corresponding to the timing bin indices.	required
`average`	`bool`	Whether to average the reduced data (default is False).	`False`

Source code in XSpect/XSpect_Analysis.py

def reduce_detector_temporal(self, run, detector_key, timing_bin_key_indices, average=False):
    """
    Reduce detector data temporally. Specifically the 1d detector output for XAS data.

    Parameters
    ----------

    run : object
        The spectroscopy run instance.

    detector_key : str
        The key corresponding to the detector data.

    timing_bin_key_indices : str
        The key corresponding to the timing bin indices.

    average : bool, optional
        Whether to average the reduced data (default is False).

    """
    detector = getattr(run, detector_key)
    time_bins=run.time_bins
    timing_indices = getattr(run, timing_bin_key_indices)#digitized indices from detector
    reduced_array = np.zeros(np.shape(time_bins)[0]+1)
    np.add.at(reduced_array, timing_indices, detector)
    setattr(run, detector_key+'_time_binned', reduced_array)
    run.update_status('Detector %s binned in time into key: %s'%(detector_key,detector_key+'_time_binned') )

`trim_ccm(run, threshold=120)`

Trim CCM values to remove bins with fewer shots than a specified threshold.

Parameters:

Name	Type	Description	Default
`run`	`object`	The spectroscopy run instance.	required
`threshold`	`int`	The minimum number of shots required to keep a CCM value (default is 120).	`120`

Source code in XSpect/XSpect_Analysis.py

def trim_ccm(self,run,threshold=120):
    """
    Trim CCM values to remove bins with fewer shots than a specified threshold.

    Parameters
    ----------

    run : object
        The spectroscopy run instance.

    threshold : int, optional
        The minimum number of shots required to keep a CCM value (default is 120).

    """

    ccm_bins=getattr(run,'ccm_bins',elist_center)
    ccm_energies=getattr(run,'ccm_energies',elist)
    counts = np.bincount(bins)
    trimmed_ccm=ccm_energies[counts[:-1]>120]
    self.make_ccm_axis(run,ccm_energies)

`XESAnalysis`

Bases: SpectroscopyAnalysis

Source code in XSpect/XSpect_Analysis.py

class XESAnalysis(SpectroscopyAnalysis):
    def __init__(self,xes_line='kbeta'):
        self.xes_line=xes_line
        pass
    def normalize_xes(self,run,detector_key,pixel_range=[300,550]):
        """
        Normalize XES data by summing the signal over a specified pixel range.

        Parameters
        ----------

        run : object
            The spectroscopy run instance.

        detector_key : str
            The key corresponding to the detector data.

        pixel_range : list of int, optional
            The pixel range to sum over for normalization (default is [300, 550]).

        """
        detector = getattr(run, detector_key)
        row_sum = np.sum(detector[:, pixel_range[0]:pixel_range[1]], axis=1)
        normed_main = np.divide(detector, row_sum[:,np.newaxis])
        setattr(run, detector_key+'_normalized', normed_main)
    def make_energy_axis(self, run,energy_axis_length, A, R,  mm_per_pixel=0.05, d=0.895):
        """
        Determination of energy axis by pixels and crystal configuration

        Parameters
        ----------

        A : float
            The detector to vH distance (mm) and can roughly float. This will affect the spectral offset.

        R : float
            The vH crystal radii (mm) and should not float. This will affect the spectral stretch.

        pixel_array : array-like
            Array of pixels to determine the energy of.

        d : float
            Crystal d-spacing. To calculate, visit: spectra.tools/bin/controller.pl?body=Bragg_Angle_Calculator

        """
        pix = mm_per_pixel
        gl = np.arange(energy_axis_length, dtype=np.float64)
        gl *= pix
        ll = gl / 2 - (np.amax(gl) - np.amin(gl)) / 4
        factor = 1.2398e4
        xaxis = factor / (2.0 * d * np.sin(np.arctan(R / (ll + A))))

        setattr(run,self.xes_line+'_energy',xaxis[::-1])
        run.update_status('XES energy axis generated for %s'%(self.xes_line))

    def reduce_det_scanvar(self, run, detector_key, scanvar_key, scanvar_bins_key):
        """
        Reduce detector data by binning according to an arbitrary scan variable.

        This method bins the detector data based on a specified scan variable and its corresponding bins. 
        The result is stored in the `run` object under a new attribute.

        Parameters
        ----------

        run : object
            The spectroscopy run instance.

        detector_key : str
            The key corresponding to the detector data within the run object.

        scanvar_key : str
            The key corresponding to the scan variable indices.

        scanvar_bins_key : str
            The key corresponding to the scan variable bins.

        Returns
        -------

        None
            The reduced data is stored in the `run` object with the key formatted as `{detector_key}_scanvar_reduced`.

        """

        detector = getattr(run, detector_key)

        scanvar_indices = getattr(run, scanvar_key)  # Shape: (4509,)
        scanvar_bins=getattr(run, scanvar_bins_key)

        n_bins = len(scanvar_bins)  # Number of bins

        # Initialize reduced_array with the correct shape (number of bins, 699, 50)
        reduced_array = np.zeros((n_bins, detector.shape[1], detector.shape[2]))

        # Iterate over the images and accumulate them into reduced_array based on timing_indices
        for i in range(detector.shape[0]):
            np.add.at(reduced_array, (scanvar_indices[i],), detector[i])

        # Store the reduced_array in the object, replace 'key_name' with the actual key
        setattr(run,  f"{detector_key}_scanvar_reduced", reduced_array)

        # Update status
        run.update_status(f'Detector binned in time into key: {detector_key}_scanvar_reduced')

`make_energy_axis(run, energy_axis_length, A, R, mm_per_pixel=0.05, d=0.895)`

Determination of energy axis by pixels and crystal configuration

Parameters:

Name	Type	Description	Default
`A`	`float`	The detector to vH distance (mm) and can roughly float. This will affect the spectral offset.	required
`R`	`float`	The vH crystal radii (mm) and should not float. This will affect the spectral stretch.	required
`pixel_array`	`array - like`	Array of pixels to determine the energy of.	required
`d`	`float`	Crystal d-spacing. To calculate, visit: spectra.tools/bin/controller.pl?body=Bragg_Angle_Calculator	`0.895`

Source code in XSpect/XSpect_Analysis.py

def make_energy_axis(self, run,energy_axis_length, A, R,  mm_per_pixel=0.05, d=0.895):
    """
    Determination of energy axis by pixels and crystal configuration

    Parameters
    ----------

    A : float
        The detector to vH distance (mm) and can roughly float. This will affect the spectral offset.

    R : float
        The vH crystal radii (mm) and should not float. This will affect the spectral stretch.

    pixel_array : array-like
        Array of pixels to determine the energy of.

    d : float
        Crystal d-spacing. To calculate, visit: spectra.tools/bin/controller.pl?body=Bragg_Angle_Calculator

    """
    pix = mm_per_pixel
    gl = np.arange(energy_axis_length, dtype=np.float64)
    gl *= pix
    ll = gl / 2 - (np.amax(gl) - np.amin(gl)) / 4
    factor = 1.2398e4
    xaxis = factor / (2.0 * d * np.sin(np.arctan(R / (ll + A))))

    setattr(run,self.xes_line+'_energy',xaxis[::-1])
    run.update_status('XES energy axis generated for %s'%(self.xes_line))

`normalize_xes(run, detector_key, pixel_range=[300, 550])`

Normalize XES data by summing the signal over a specified pixel range.

Parameters:

Name	Type	Description	Default
`run`	`object`	The spectroscopy run instance.	required
`detector_key`	`str`	The key corresponding to the detector data.	required
`pixel_range`	`list of int`	The pixel range to sum over for normalization (default is [300, 550]).	`[300, 550]`

Source code in XSpect/XSpect_Analysis.py

def normalize_xes(self,run,detector_key,pixel_range=[300,550]):
    """
    Normalize XES data by summing the signal over a specified pixel range.

    Parameters
    ----------

    run : object
        The spectroscopy run instance.

    detector_key : str
        The key corresponding to the detector data.

    pixel_range : list of int, optional
        The pixel range to sum over for normalization (default is [300, 550]).

    """
    detector = getattr(run, detector_key)
    row_sum = np.sum(detector[:, pixel_range[0]:pixel_range[1]], axis=1)
    normed_main = np.divide(detector, row_sum[:,np.newaxis])
    setattr(run, detector_key+'_normalized', normed_main)

`reduce_det_scanvar(run, detector_key, scanvar_key, scanvar_bins_key)`

Reduce detector data by binning according to an arbitrary scan variable.

This method bins the detector data based on a specified scan variable and its corresponding bins. The result is stored in the run object under a new attribute.

Parameters:

Name	Type	Description	Default
`run`	`object`	The spectroscopy run instance.	required
`detector_key`	`str`	The key corresponding to the detector data within the run object.	required
`scanvar_key`	`str`	The key corresponding to the scan variable indices.	required
`scanvar_bins_key`	`str`	The key corresponding to the scan variable bins.	required

Returns:

Type	Description
`None`	The reduced data is stored in the `run` object with the key formatted as `{detector_key}_scanvar_reduced`.

Source code in XSpect/XSpect_Analysis.py

def reduce_det_scanvar(self, run, detector_key, scanvar_key, scanvar_bins_key):
    """
    Reduce detector data by binning according to an arbitrary scan variable.

    This method bins the detector data based on a specified scan variable and its corresponding bins. 
    The result is stored in the `run` object under a new attribute.

    Parameters
    ----------

    run : object
        The spectroscopy run instance.

    detector_key : str
        The key corresponding to the detector data within the run object.

    scanvar_key : str
        The key corresponding to the scan variable indices.

    scanvar_bins_key : str
        The key corresponding to the scan variable bins.

    Returns
    -------

    None
        The reduced data is stored in the `run` object with the key formatted as `{detector_key}_scanvar_reduced`.

    """

    detector = getattr(run, detector_key)

    scanvar_indices = getattr(run, scanvar_key)  # Shape: (4509,)
    scanvar_bins=getattr(run, scanvar_bins_key)

    n_bins = len(scanvar_bins)  # Number of bins

    # Initialize reduced_array with the correct shape (number of bins, 699, 50)
    reduced_array = np.zeros((n_bins, detector.shape[1], detector.shape[2]))

    # Iterate over the images and accumulate them into reduced_array based on timing_indices
    for i in range(detector.shape[0]):
        np.add.at(reduced_array, (scanvar_indices[i],), detector[i])

    # Store the reduced_array in the object, replace 'key_name' with the actual key
    setattr(run,  f"{detector_key}_scanvar_reduced", reduced_array)

    # Update status
    run.update_status(f'Detector binned in time into key: {detector_key}_scanvar_reduced')

`experiment`

Source code in XSpect/XSpect_Analysis.py

class experiment:
    def __init__(self, lcls_run, hutch, experiment_id):
        """
        Initializes an experiment instance.

        Parameters
        ----------

        lcls_run : str
            LCLS run identifier. The LCLS run not the scan/run. Example: 21

        hutch : str
            Hutch name. Example: xcs

        experiment_id : str
            Experiment identifier. Example: xcsl1004021

        """
        self.lcls_run = lcls_run
        self.hutch = hutch
        self.experiment_id = experiment_id
        self.get_experiment_directory()
    def get_experiment_directory(self):
        """
        Determines and returns the directory of the experiment based on the hutch and experiment ID. 
        It attempts the various paths LCLS has had over the years with recent S3DF paths being the first attempt.

        Returns
        -------

        str
            The directory of the experiment.

        Raises
        ------

        Exception
            If the directory cannot be found.

        """
        experiment_directories = [
        '/sdf/data/lcls/ds/%s/%s/hdf5/smalldata',
        '/reg/data/drpsrcf/%s/%s/scratch/hdf5/smalldata',
        '/cds/data/drpsrcf/%s/%s/scratch/hdf5/smalldata',
        '/reg/d/psdm/%s/%s/hdf5/smalldata'
        ]
        for directory in experiment_directories:
            experiment_directory = directory % (self.hutch, self.experiment_id)
            if os.path.exists(experiment_directory) and os.listdir(experiment_directory):
                self.experiment_directory=experiment_directory
                return experiment_directory
        raise Exception("Unable to find experiment directory.")

`init(lcls_run, hutch, experiment_id)`

Initializes an experiment instance.

Parameters:

Name	Type	Description	Default
`lcls_run`	`str`	LCLS run identifier. The LCLS run not the scan/run. Example: 21	required
`hutch`	`str`	Hutch name. Example: xcs	required
`experiment_id`	`str`	Experiment identifier. Example: xcsl1004021	required

Source code in XSpect/XSpect_Analysis.py

def __init__(self, lcls_run, hutch, experiment_id):
    """
    Initializes an experiment instance.

    Parameters
    ----------

    lcls_run : str
        LCLS run identifier. The LCLS run not the scan/run. Example: 21

    hutch : str
        Hutch name. Example: xcs

    experiment_id : str
        Experiment identifier. Example: xcsl1004021

    """
    self.lcls_run = lcls_run
    self.hutch = hutch
    self.experiment_id = experiment_id
    self.get_experiment_directory()

`get_experiment_directory()`

Determines and returns the directory of the experiment based on the hutch and experiment ID. It attempts the various paths LCLS has had over the years with recent S3DF paths being the first attempt.

Returns:

Type	Description
`str`	The directory of the experiment.

Raises:

Type	Description
`Exception`	If the directory cannot be found.

Source code in XSpect/XSpect_Analysis.py

def get_experiment_directory(self):
    """
    Determines and returns the directory of the experiment based on the hutch and experiment ID. 
    It attempts the various paths LCLS has had over the years with recent S3DF paths being the first attempt.

    Returns
    -------

    str
        The directory of the experiment.

    Raises
    ------

    Exception
        If the directory cannot be found.

    """
    experiment_directories = [
    '/sdf/data/lcls/ds/%s/%s/hdf5/smalldata',
    '/reg/data/drpsrcf/%s/%s/scratch/hdf5/smalldata',
    '/cds/data/drpsrcf/%s/%s/scratch/hdf5/smalldata',
    '/reg/d/psdm/%s/%s/hdf5/smalldata'
    ]
    for directory in experiment_directories:
        experiment_directory = directory % (self.hutch, self.experiment_id)
        if os.path.exists(experiment_directory) and os.listdir(experiment_directory):
            self.experiment_directory=experiment_directory
            return experiment_directory
    raise Exception("Unable to find experiment directory.")

`spectroscopy_experiment`

Bases: experiment

A class to represent a spectroscopy experiment. Trying to integrate methods that incorporate meta parameters of the experiment but did not follow through.

Source code in XSpect/XSpect_Analysis.py

class spectroscopy_experiment(experiment):
    """
    A class to represent a spectroscopy experiment. 
    Trying to integrate methods that incorporate meta parameters of the experiment but did not follow through.
    """
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
    def add_detector(self, detector_name, detector_dimensions):
        self.detector_name = detector_name
        self.detector_dimensions = detector_dimensions

XSpect Controller

SpectroscopyAnalysis

bin_uniques(run, key)

filter_detector_adu(run, detector, adu_threshold=3.0)

filter_nan(run, shot_mask_key, filter_key='ipm')

filter_shots(run, shot_mask_key, filter_key='ipm', threshold=10000.0)

patch_pixel(run, detector_key, pixel, mode='average', patch_range=4, deg=1, poly_range=6, axis=1)

patch_pixel_1d(run, detector_key, pixel, mode='average', patch_range=4, deg=1, poly_range=6)

patch_pixels(run, detector_key, mode='average', patch_range=4, deg=1, poly_range=6, axis=1)

patch_pixels_1d(run, detector_key, mode='average', patch_range=4, deg=1, poly_range=6)

purge_keys(run, keys)

reduce_detector_spatial(run, detector_key, shot_range=[0, None], rois=[[0, None]], reduction_function=np.sum, purge=True, combine=True)

reduce_detector_temporal(run, detector_key, timing_bin_key_indices, average=False)

separate_shots(run, detector_key, filter_keys)

time_binning(run, bins, lxt_key='lxt_ttc', fast_delay_key='encoder', tt_correction_key='time_tool_correction')

union_shots(run, detector_key, filter_keys, new_key=True)

XASAnalysis

ccm_binning(run, ccm_bins_key, ccm_key='ccm')

make_ccm_axis(run, energies)

reduce_detector_ccm(run, detector_key, ccm_bin_key_indices, average=False, not_ccm=False)

reduce_detector_ccm_temporal(run, detector_key, timing_bin_key_indices, ccm_bin_key_indices, average=True)

reduce_detector_temporal(run, detector_key, timing_bin_key_indices, average=False)

trim_ccm(run, threshold=120)

XESAnalysis

make_energy_axis(run, energy_axis_length, A, R, mm_per_pixel=0.05, d=0.895)

normalize_xes(run, detector_key, pixel_range=[300, 550])

reduce_det_scanvar(run, detector_key, scanvar_key, scanvar_bins_key)

experiment

__init__(lcls_run, hutch, experiment_id)

get_experiment_directory()

spectroscopy_experiment

`SpectroscopyAnalysis`

`bin_uniques(run, key)`

`filter_detector_adu(run, detector, adu_threshold=3.0)`

`filter_nan(run, shot_mask_key, filter_key='ipm')`

`filter_shots(run, shot_mask_key, filter_key='ipm', threshold=10000.0)`

`patch_pixel(run, detector_key, pixel, mode='average', patch_range=4, deg=1, poly_range=6, axis=1)`

`patch_pixel_1d(run, detector_key, pixel, mode='average', patch_range=4, deg=1, poly_range=6)`

`patch_pixels(run, detector_key, mode='average', patch_range=4, deg=1, poly_range=6, axis=1)`

`patch_pixels_1d(run, detector_key, mode='average', patch_range=4, deg=1, poly_range=6)`

`purge_keys(run, keys)`

`reduce_detector_spatial(run, detector_key, shot_range=[0, None], rois=[[0, None]], reduction_function=np.sum, purge=True, combine=True)`

`reduce_detector_temporal(run, detector_key, timing_bin_key_indices, average=False)`

`separate_shots(run, detector_key, filter_keys)`

`time_binning(run, bins, lxt_key='lxt_ttc', fast_delay_key='encoder', tt_correction_key='time_tool_correction')`

`union_shots(run, detector_key, filter_keys, new_key=True)`

`XASAnalysis`

`ccm_binning(run, ccm_bins_key, ccm_key='ccm')`

`make_ccm_axis(run, energies)`

`reduce_detector_ccm(run, detector_key, ccm_bin_key_indices, average=False, not_ccm=False)`

`reduce_detector_ccm_temporal(run, detector_key, timing_bin_key_indices, ccm_bin_key_indices, average=True)`

`reduce_detector_temporal(run, detector_key, timing_bin_key_indices, average=False)`

`trim_ccm(run, threshold=120)`

`XESAnalysis`

`make_energy_axis(run, energy_axis_length, A, R, mm_per_pixel=0.05, d=0.895)`

`normalize_xes(run, detector_key, pixel_range=[300, 550])`

`reduce_det_scanvar(run, detector_key, scanvar_key, scanvar_bins_key)`

`experiment`

`init(lcls_run, hutch, experiment_id)`

`get_experiment_directory()`

`spectroscopy_experiment`