preprocess

crosspeak.preprocess

reference_spectrum(series)

The reference spectrum subtracted to form the dynamic spectrum.

Returns the perturbation-mean spectrum — the average intensity at each wavenumber across all perturbation points. This is the reference that mean_center subtracts, and the choice that makes the correlation intensities pure covariances.

Noda's formalism allows any reference (the first spectrum, an external one), but the mean is the conventional pick: it centres the series so that every deviation the correlation sees is a genuine departure from the average state, not an artefact of which spectrum you happened to choose as the baseline.

Parameters:

Name	Type	Description	Default
series		A SpectralSeries.	required

Returns:

Type	Description
ndarray	The mean spectrum, shape (n_wavenumbers,).

Source code in src/crosspeak/preprocess.py

def reference_spectrum(series):
    """The reference spectrum subtracted to form the dynamic spectrum.

    Returns the perturbation-mean spectrum — the average intensity at each
    wavenumber across all perturbation points. This is the reference that
    `mean_center` subtracts, and the choice that makes the correlation
    intensities pure covariances.

    Noda's formalism allows any reference (the first spectrum, an external
    one), but the mean is the conventional pick: it centres the series so that
    every deviation the correlation sees is a genuine departure from the
    average state, not an artefact of which spectrum you happened to choose as
    the baseline.

    Parameters
    ----------
    series
        A `SpectralSeries`.

    Returns
    -------
    np.ndarray
        The mean spectrum, shape `(n_wavenumbers,)`.
    """
    return series.intensities.mean(axis=0)

mean_center(series)

Subtract the reference spectrum to form the dynamic spectrum.

Replaces each spectrum with its deviation from the perturbation mean — the dynamic spectrum Ỹ in Noda's notation. This is the first step of every correlation calculation: synchronous and asynchronous mean-centre internally, so you rarely call this yourself, but it is exposed for when you want the centred series in hand to inspect or plot.

Parameters:

Name	Type	Description	Default
series		A SpectralSeries.	required

Returns:

Type	Description
SpectralSeries	A new series of mean-centred intensities; wavenumbers, perturbations, and name preserved. The original is untouched.

Source code in src/crosspeak/preprocess.py

def mean_center(series):
    """Subtract the reference spectrum to form the dynamic spectrum.

    Replaces each spectrum with its deviation from the perturbation mean — the
    dynamic spectrum Ỹ in Noda's notation. This is the first step of every
    correlation calculation: `synchronous` and `asynchronous` mean-centre
    internally, so you rarely call this yourself, but it is exposed for when you
    want the centred series in hand to inspect or plot.

    Parameters
    ----------
    series
        A `SpectralSeries`.

    Returns
    -------
    SpectralSeries
        A new series of mean-centred intensities; wavenumbers, perturbations,
        and name preserved. The original is untouched.
    """
    reference = reference_spectrum(series)
    dynamic = series.intensities - reference
    return SpectralSeries(
        wavenumbers=series.wavenumbers,
        perturbations=series.perturbations,
        intensities=dynamic,
        name=series.name,
    )

crop_region(series, low, high)

Return a new SpectralSeries restricted to a wavenumber range Both bounds are inclusive. Does not modify original series. Wavenumber axis direction is maintained

Source code in src/crosspeak/preprocess.py

def crop_region(series: SpectralSeries, low: float, high: float) -> SpectralSeries:
    """Return a new SpectralSeries restricted to a wavenumber range
    Both bounds are inclusive. Does not modify original series.
    Wavenumber axis direction is maintained
    """

    if low > high:
        raise ValueError(f"low ({low}) must be <= high ({high})")

    wn = series.wavenumbers
    mask = (wn >= low) & (wn <= high)

    if not mask.any():
        raise ValueError(
            f"requested range [{low}, {high}] has no overlap "
            f"with data range [{wn.min()}, {wn.max()}]"
        )

    return SpectralSeries(
        wavenumbers=wn[mask],
        perturbations=series.perturbations,
        intensities=series.intensities[:, mask],
        name=series.name,
    )

savgol_smooth(series, window_length=13, polyorder=3, **kwargs)

Apply SavitzkyGolay smoothing along the wavenumber axis.

Each spectrum (row) is smoothed independently. The original series is not modified.

Parameters:

Name	Type	Description	Default
series	SpectralSeries	Input SpectralSeries.	required
window_length	int	Length of the filter window. Must be odd, greater than polyorder, and no larger than the number of wavenumber points. Default 13.	13
polyorder	int	Order of the polynomial fit. Must be less than window_length. Default 3.	3
**kwargs		Additional keyword arguments passed through to scipy.signal.savgol_filter (e.g. deriv, delta, mode, cval). Do not pass axis — it is fixed to the wavenumber axis.	{}

Returns:

Type	Description
SpectralSeries	New SpectralSeries with smoothed intensities. Wavenumbers, perturbations, and name are preserved.

Source code in src/crosspeak/preprocess.py

def savgol_smooth(
    series: SpectralSeries,
    window_length: int = 13,
    polyorder: int = 3,
    **kwargs,
) -> SpectralSeries:
    """Apply SavitzkyGolay smoothing along the wavenumber axis.

    Each spectrum (row) is smoothed independently. The original series is
    not modified.

    Parameters
    ----------
    series
        Input SpectralSeries.
    window_length
        Length of the filter window. Must be odd, greater than `polyorder`,
        and no larger than the number of wavenumber points. Default 13.
    polyorder
        Order of the polynomial fit. Must be less than `window_length`.
        Default 3.
    **kwargs
        Additional keyword arguments passed through to
        `scipy.signal.savgol_filter` (e.g. `deriv`, `delta`, `mode`, `cval`).
        Do not pass `axis` — it is fixed to the wavenumber axis.

    Returns
    -------
    SpectralSeries
        New SpectralSeries with smoothed intensities. Wavenumbers,
        perturbations, and name are preserved.
    """
    if window_length % 2 == 0:
        raise ValueError(f"window_length must be odd, got {window_length}")

    smoothed = savgol_filter(
        series.intensities,
        window_length=window_length,
        polyorder=polyorder,
        axis=-1,
        **kwargs,
    )

    return SpectralSeries(
        wavenumbers=series.wavenumbers,
        perturbations=series.perturbations,
        intensities=smoothed,
        name=series.name,
    )

area_normalize(series, target_area=1.0, reference_region=None)

Normalize each spectrum to a target area under the curve.

Parameters:

Name	Type	Description	Default
series	SpectralSeries	Input SpectralSeries.	required
target_area	float	Desired area under the curve for each spectrum after normalization. Default is 1.0.	1.0
reference_region	tuple[float, float] | None	Optional tuple specifying a wavenumber range (low, high) to use for calculating the area. If None, the entire wavenumber range is used.	None

Returns:

Type	Description
SpectralSeries	New SpectralSeries with normalized intensities. Wavenumbers, perturbations, and name are preserved.

Raises:

Type	Description
ValueError	If any row's integrated area is zero, or if reference_region is invalid (inverted bounds or no overlap with data — delegated to crop_region).

Source code in src/crosspeak/preprocess.py

def area_normalize(
    series: SpectralSeries,
    target_area: float = 1.0,
    reference_region: tuple[float, float] | None = None,
) -> SpectralSeries:
    """Normalize each spectrum to a target area under the curve.

    Parameters
    ----------
    series
        Input SpectralSeries.
    target_area
        Desired area under the curve for each spectrum after normalization.
        Default is 1.0.
    reference_region
        Optional tuple specifying a wavenumber range (low, high) to use for
        calculating the area. If None, the entire wavenumber range is used.

    Returns
    -------
    SpectralSeries
        New SpectralSeries with normalized intensities. Wavenumbers,
        perturbations, and name are preserved.

    Raises
    ------
    ValueError
        If any row's integrated area is zero, or if `reference_region` is
        invalid (inverted bounds or no overlap with data — delegated to
        `crop_region`).
    """
    if reference_region is not None:
        ref = crop_region(series, *reference_region)
        wn_ref = ref.wavenumbers
        intens_ref = ref.intensities
    else:
        wn_ref = series.wavenumbers
        intens_ref = series.intensities

    areas = np.abs(np.trapezoid(intens_ref, x=wn_ref, axis=-1))

    if (areas == 0).any():
        zero_rows = np.where(areas == 0)[0].tolist()
        raise ValueError(f"zero integrated area on perturbation rows {zero_rows}; cannot normalize")

    scale = target_area / areas
    normalized = series.intensities * scale[:, None]

    return SpectralSeries(
        wavenumbers=series.wavenumbers,
        perturbations=series.perturbations,
        intensities=normalized,
        name=series.name,
    )

vector_normalize(series)

Scale each spectrum to unit Euclidean (L2) norm.

For each perturbation row, divides the intensities by their L2 norm, removing per-spectrum multiplicative scaling without assuming a stable reference band. Corrects ATR sample-loading and contact variation before correlation.

Parameters:

Name	Type	Description	Default
series	SpectralSeries	Input SpectralSeries. The norm is computed over the full wavenumber range of the passed series; crop first to restrict it.	required

Returns:

Type	Description
SpectralSeries	New SpectralSeries with unit-norm rows. Metadata preserved.

Raises:

Type	Description
ValueError	If any row has zero norm (an all-zero spectrum).

Source code in src/crosspeak/preprocess.py

def vector_normalize(series: SpectralSeries) -> SpectralSeries:
    """Scale each spectrum to unit Euclidean (L2) norm.

    For each perturbation row, divides the intensities by their L2 norm,
    removing per-spectrum multiplicative scaling without assuming a stable
    reference band. Corrects ATR sample-loading and contact variation before
    correlation.

    Parameters
    ----------
    series
        Input SpectralSeries. The norm is computed over the full wavenumber
        range of the passed series; crop first to restrict it.

    Returns
    -------
    SpectralSeries
        New SpectralSeries with unit-norm rows. Metadata preserved.

    Raises
    ------
    ValueError
        If any row has zero norm (an all-zero spectrum).
    """
    norms = np.linalg.norm(series.intensities, axis=-1)

    if (norms == 0).any():
        zero_rows = np.where(norms == 0)[0].tolist()
        raise ValueError(f"zero norm on perturbation rows {zero_rows}; cannot normalize")

    normalized = series.intensities / norms[:, None]

    return SpectralSeries(
        wavenumbers=series.wavenumbers,
        perturbations=series.perturbations,
        intensities=normalized,
        name=series.name,
    )

snv(series)

Apply the standard normal variate (SNV) transform to each spectrum.

For each perturbation row, subtracts the row mean and divides by the row sample standard deviation (ddof=1). Removes both an additive baseline offset and a multiplicative scale per spectrum, without assuming a stable reference band.

Parameters:

Name	Type	Description	Default
series	SpectralSeries	Input SpectralSeries. The transform is computed over the full wavenumber range of the passed series; crop first to restrict it.	required

Returns:

Type	Description
SpectralSeries	New SpectralSeries with SNV-transformed rows. Metadata preserved.

Raises:

Type	Description
ValueError	If any row has zero standard deviation (a flat spectrum).

Source code in src/crosspeak/preprocess.py

def snv(series: SpectralSeries) -> SpectralSeries:
    """Apply the standard normal variate (SNV) transform to each spectrum.

    For each perturbation row, subtracts the row mean and divides by the row
    sample standard deviation (ddof=1). Removes both an additive baseline
    offset and a multiplicative scale per spectrum, without assuming a stable
    reference band.

    Parameters
    ----------
    series
        Input SpectralSeries. The transform is computed over the full
        wavenumber range of the passed series; crop first to restrict it.

    Returns
    -------
    SpectralSeries
        New SpectralSeries with SNV-transformed rows. Metadata preserved.

    Raises
    ------
    ValueError
        If any row has zero standard deviation (a flat spectrum).
    """
    means = series.intensities.mean(axis=-1, keepdims=True)
    stds = series.intensities.std(axis=-1, ddof=1, keepdims=True)

    if (stds == 0).any():
        zero_rows = np.where(stds.ravel() == 0)[0].tolist()
        raise ValueError(
            f"zero standard deviation on perturbation rows {zero_rows}; cannot apply SNV"
        )

    transformed = (series.intensities - means) / stds

    return SpectralSeries(
        wavenumbers=series.wavenumbers,
        perturbations=series.perturbations,
        intensities=transformed,
        name=series.name,
    )