Source code for ess.powder.grouping

# SPDX-License-Identifier: BSD-3-Clause
# Copyright (c) 2025 Scipp contributors (https://github.com/scipp)
"""Grouping and merging of pixels / voxels."""

import numpy as np
import scipp as sc

from .types import (
    CountsDspacing,
    DspacingBins,
    FocussedDataDspacing,
    FocussedDataDspacingTwoTheta,
    KeepEvents,
    ReducedCountsDspacing,
    RunType,
    ScaledCountsDspacing,
    TwoThetaBins,
)


def _reconstruct_wavelength(
    dspacing_bins: DspacingBins, two_theta_bins: TwoThetaBins
) -> sc.Variable:
    dspacing = dspacing_bins
    two_theta = sc.midpoints(two_theta_bins)
    return (2 * dspacing * sc.sin(two_theta / 2)).to(unit='angstrom')


[docs] def focus_data_dspacing_and_two_theta( data: CountsDspacing[RunType], dspacing_bins: DspacingBins, keep_events: KeepEvents[RunType], ) -> ReducedCountsDspacing[RunType]: """ Reduce the pixel-based data to d-spacing and two-theta dimensions. The two-theta binning does not use :py:class:`TwoThetaBins` but instead computes the two-theta bins from the 'two_theta' coordinate of the input data. This is necessary to ensure that we have sufficiently high wavelength resolution when performing a monitor normalization in a follow-up workflow step. If we were to use :py:class:`TwoThetaBins` we would be influenced by and limited to the two-theta binning the user requests for the end result, which may not be sufficient. Parameters ---------- data: The input data to be reduced, which must have 'wavelength', 'dspacing', 'two_theta' coordinates. dspacing_bins: The bins to use for the d-spacing dimension. keep_events: Whether to keep the events in the output. If `False`, the output will be histogrammed instead of binned. Returns ------- : The reduced data with 'dspacing' and 'two_theta' dimensions. """ ttheta = data.coords['two_theta'] ttheta_min = ttheta.nanmin() ttheta_max = ttheta.nanmax() ttheta_max.value = np.nextafter(ttheta_max.value, np.inf) twotheta_bins = sc.linspace( 'two_theta', start=ttheta_min, stop=ttheta_max, num=1024, unit=ttheta.unit, ) args = {twotheta_bins.dim: twotheta_bins, dspacing_bins.dim: dspacing_bins} if keep_events.value: result = data.bin(args) else: # Reconstructing the wavelength results in an inconsistency if dspacing was # computed with a calibration table. Another option would be to use, e.g., the # mean wavelength in each bin, but this leads to random wavelength values that # break stream processing. result = data.hist(args).assign_coords( wavelength=_reconstruct_wavelength( dspacing_bins=dspacing_bins, two_theta_bins=twotheta_bins ) ) return ReducedCountsDspacing[RunType](result)
[docs] def integrate_two_theta( data: ScaledCountsDspacing[RunType], ) -> FocussedDataDspacing[RunType]: """Integrate the two-theta dimension of the data.""" if 'two_theta' not in data.dims: raise sc.DimensionError("Data does not have a 'two_theta' dimension.") return FocussedDataDspacing[RunType]( data.nansum(dim='two_theta') if data.bins is None else data.bins.concat('two_theta') )
[docs] def group_two_theta( data: ScaledCountsDspacing[RunType], two_theta_bins: TwoThetaBins, ) -> FocussedDataDspacingTwoTheta[RunType]: """Group the data by two-theta bins.""" if 'two_theta' not in data.dims: raise ValueError("Data does not have a 'two_theta' dimension.") data = data.assign_coords(two_theta=sc.midpoints(data.coords['two_theta'])) return FocussedDataDspacingTwoTheta[RunType]( data.groupby('two_theta', bins=two_theta_bins).nansum('two_theta') if data.bins is None else data.bin(two_theta=two_theta_bins) )
[docs] def collect_detectors(*detectors: sc.DataArray) -> sc.DataGroup: """Store all inputs in a single data group. This function is intended to be used to reduce a workflow which was mapped over detectors. Parameters ---------- detectors: Data arrays for each detector bank. All arrays must have a scalar "detector" coord containing a ``str``. Returns ------- : The inputs as a data group with the "detector" coord as the key. """ return sc.DataGroup({da.coords.pop('detector').value: da for da in detectors})
providers = ( focus_data_dspacing_and_two_theta, integrate_two_theta, group_two_theta, ) """Sciline providers for grouping pixels."""