Processing Module / 処理モジュール

The wandas.processing module contains audio operations for temporal, spectral, cepstral, statistical, filter, and effect processing. Each operation's numerical contract is generated from its Google-style docstring.

wandas.processingは時間領域、スペクトル、ケプストラム、統計、filter、effectのaudio operationを提供します。各operationの数値契約はGoogle style docstringから生成されます。

wandas.processing

Audio time series processing operations.

This module provides audio processing operations for time series data.

Attributes

__all__ = ['apply_channel_factors', 'AudioOperation', 'ChannelIndependentAudioOperation', 'create_operation', 'get_operation', 'register_lazy_operation', 'register_operation', 'Astype', 'Cepstrum', 'Lifter', 'SpectralEnvelope', 'SpectrogramCepstrum', 'AWeighting', 'HighPassFilter', 'LowPassFilter', 'CSD', 'Coherence', 'CoherenceFromCSD', 'FFT', 'IFFT', 'ISTFT', 'NOctSpectrum', 'NOctSynthesis', 'STFT', 'SpectrogramCSD', 'TransferFunction', 'TransferFunctionFromCSD', 'Welch', 'ReSampling', 'RmsTrend', 'SoundLevel', 'Trim', 'AddWithSNR', 'HpssHarmonic', 'HpssPercussive', 'ABS', 'ChannelDifference', 'Mean', 'Power', 'Sum', 'LoudnessZwst', 'LoudnessZwtv', 'RoughnessDw', 'RoughnessDwSpec', 'SharpnessDin', 'SharpnessDinSt'] module-attribute

Classes

AudioOperation

Bases: Generic[InputArrayType, OutputArrayType]

Base class for numerical audio operations.

Subclasses may depend on relationships between channels. The default lazy execution graph therefore passes the complete channel-first tensor to the eager :meth:_process kernel as one whole-frame operation.

Use :class:ChannelIndependentAudioOperation instead when every output channel depends only on the corresponding input channel.

Source code in wandas/processing/base.py

class AudioOperation(Generic[InputArrayType, OutputArrayType]):
    """Base class for numerical audio operations.

    Subclasses may depend on relationships between channels. The default lazy
    execution graph therefore passes the complete channel-first tensor to the
    eager :meth:`_process` kernel as one whole-frame operation.

    Use :class:`ChannelIndependentAudioOperation` instead when every output
    channel depends only on the corresponding input channel.
    """

    # Class variable: operation name
    name: ClassVar[str]
    _expected_input_count: ClassVar[int | None] = 1
    # When True, the lazy graph casts each kernel result to calculate_output_dtype(),
    # so a float32 input keeps float32 output even when the kernel computes in float64.
    _keeps_input_precision: ClassVar[bool] = False

    _config: dict[str, Any]
    _process: Callable[..., OutputArrayType] = _unimplemented_process

    def __init_subclass__(cls, **kwargs: Any) -> None:
        """Ensure subclass ``process`` overrides keep the base input contract."""
        super().__init_subclass__(**kwargs)
        process = cls.__dict__.get("process")
        if process is None or getattr(process, "_wandas_validates_process_inputs", False):
            return

        @wraps(process)
        def validated_process(self: "AudioOperation[Any, Any]", data: Any, *inputs: Any) -> Any:
            self._validate_process_inputs(data, *inputs)
            return process(self, data, *inputs)

        setattr(validated_process, "_wandas_validates_process_inputs", True)
        cls.process = cast(Any, validated_process)

    def __init__(self, sampling_rate: float, *, pure: bool = True, **params: Any):
        """
        Initialize AudioOperation.

        Args:
            sampling_rate: float. Sampling rate (Hz)
            pure: bool, default=True. Whether the operation is pure (deterministic with no side effects).
                When True, Dask can cache results for identical inputs.
                Set to False only if the operation has side effects or is non-deterministic.
            **params: Any. Operation-specific parameters
        """
        object.__setattr__(self, "_sampling_rate", float(sampling_rate))
        object.__setattr__(
            self,
            "_config",
            {key: _snapshot_config_value(value) for key, value in params.items()},
        )
        self.pure = pure

        # Validate parameters during initialization
        self.validate_params()

        # Create processor function (lazy initialization possible)
        self._setup_processor()

        logger.debug(f"Initialized {self.__class__.__name__} operation with params: {params}")

    @property
    def sampling_rate(self) -> float:
        """Sampling rate captured at operation construction time."""
        return object.__getattribute__(self, "_sampling_rate")

    @property
    def params(self) -> _DefensiveParamsMapping:
        """Return a read-only defensive snapshot of operation parameters."""
        return _DefensiveParamsMapping(self.to_params())

    def to_params(self) -> Mapping[str, Any]:
        """Return operation parameters used for lineage and display."""
        return self._config_snapshot()

    def _config_snapshot(self) -> dict[str, Any]:
        """Return a defensive copy of base-managed constructor config."""
        return {key: _snapshot_config_value(value) for key, value in self._config.items()}

    def _config_value(self, key: str) -> Any:
        """Return a defensive snapshot for one base-managed config value."""
        return _snapshot_config_value(self._config[key])

    def validate_params(self) -> None:
        """Validate parameters (raises exception if invalid)"""

    def _setup_processor(self) -> None:
        """Set up processor function (implemented by subclasses)"""

    def get_metadata_updates(self) -> dict[str, Any]:
        """
        Get metadata updates to apply after processing.

        This method allows operations to specify how metadata should be
        updated after processing. By default, no metadata is updated.

        Returns:
            dict: Dictionary of metadata updates. Can include:
                - 'sampling_rate': New sampling rate (float)
                - Other metadata keys as needed

        Examples:
            Return empty dict for operations that don't change metadata:

            >>> return {}

            Return new sampling rate for operations that resample:

            >>> return {"sampling_rate": self.target_sr}

        Notes:
            This method is called by the framework after processing to update
            the frame metadata. Subclasses should override this method if they
            need to update metadata (e.g., changing sampling rate).

            Design principle: Operations should use parameters provided at
            initialization (via __init__). All necessary information should be
            available as instance variables.
        """
        return {}

    def get_display_name(self) -> str | None:
        """
        Get display name for the operation for use in channel labels.

        Returns ``_display`` if the subclass sets it, otherwise ``None``
        (which tells the framework to fall back to the ``name`` class
        variable).  Subclasses with dynamic display names can still
        override this method.
        """
        return getattr(self, "_display", None)

    def _validate_process_input_count(self, input_count: int) -> None:
        """Validate process input arity using the operation class contract."""
        expected = self._expected_input_count
        if expected is None or input_count == expected:
            return

        noun = "input" if expected == 1 else "inputs"
        expected_text = "one" if expected == 1 else str(expected)
        raise ValueError(
            f"Expected exactly {expected_text} {noun} for {self.__class__.__name__}; "
            f"got {input_count}. Use an operation-specific method when multiple "
            "runtime inputs are required."
        )

    def _validate_process_inputs(self, data: DaArray, *inputs: DaArray, ndim: int | None = None) -> None:
        """Validate Frame-internal lazy inputs before building a process graph."""
        self._validate_process_input_count(1 + len(inputs))
        _validate_channel_first_array(data, "data", ndim=ndim)
        for index, input_data in enumerate(inputs, start=1):
            _validate_channel_first_array(input_data, f"input {index}")

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        """
        Calculate output data shape after operation.

        The default returns *input_shape* unchanged, which is correct for the
        majority of operations (filters, effects, weighting, etc.).
        Subclasses that alter the shape (e.g. FFT, STFT, resampling) **must**
        override this method.

        Args:
            input_shape: tuple. Input data shape

        Returns:
            tuple: Output data shape
        """
        return input_shape

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        """Calculate output dtype metadata after operation."""
        return np.result_type(input_dtype, *input_dtypes)

    def _build_whole_frame_graph(
        self,
        data: DaArray,
        inputs: tuple[DaArray, ...],
        *,
        output_shape: tuple[int, ...],
        output_dtype: np.dtype[Any],
    ) -> DaArray:
        """Wrap the complete channel-first tensor in one delayed kernel call."""
        delayed_result = delayed(_execute_wandas_operation, name=self.name, pure=self.pure)(self, data, *inputs)
        return _da_from_delayed(delayed_result, shape=output_shape, dtype=output_dtype)

    def _build_execution_graph(
        self,
        data: DaArray,
        inputs: tuple[DaArray, ...],
        *,
        output_shape: tuple[int, ...],
        output_dtype: np.dtype[Any],
    ) -> DaArray:
        """Build this operation's lazy execution graph.

        Subclasses can extend graph construction without adding dispatch branches
        to :meth:`process`. The default preserves the historical whole-frame
        boundary.
        """
        return self._build_whole_frame_graph(
            data,
            inputs,
            output_shape=output_shape,
            output_dtype=output_dtype,
        )

    def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
        """
        Execute operation lazily on Frame-internal channel-first Dask arrays.

        ``data`` must be the lazy array held by a Frame, with a leading channel
        axis such as ``(channels, samples)``. Direct 1-D lazy input is not part
        of this API; use a Frame operation or reshape direct lazy inputs to add
        a channel axis before calling ``process()``. Multi-input operations
        pass additional channel-first Dask arrays through ``*inputs``.
        """
        self._validate_process_inputs(data, *inputs)
        logger.debug("Adding delayed operation to computation graph")
        output_shape = self.calculate_output_shape(data.shape)
        output_dtype = self.calculate_output_dtype(data.dtype, *(input_data.dtype for input_data in inputs))
        return self._build_execution_graph(
            data,
            inputs,
            output_shape=output_shape,
            output_dtype=output_dtype,
        )

Attributes

name class-attribute

pure = pure instance-attribute

sampling_rate property

Sampling rate captured at operation construction time.

params property

Return a read-only defensive snapshot of operation parameters.

Functions

__init_subclass__(**kwargs)

Ensure subclass process overrides keep the base input contract.

Source code in wandas/processing/base.py

def __init_subclass__(cls, **kwargs: Any) -> None:
    """Ensure subclass ``process`` overrides keep the base input contract."""
    super().__init_subclass__(**kwargs)
    process = cls.__dict__.get("process")
    if process is None or getattr(process, "_wandas_validates_process_inputs", False):
        return

    @wraps(process)
    def validated_process(self: "AudioOperation[Any, Any]", data: Any, *inputs: Any) -> Any:
        self._validate_process_inputs(data, *inputs)
        return process(self, data, *inputs)

    setattr(validated_process, "_wandas_validates_process_inputs", True)
    cls.process = cast(Any, validated_process)

__init__(sampling_rate, *, pure=True, **params)

Initialize AudioOperation.

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate (Hz)	required
pure	bool	bool, default=True. Whether the operation is pure (deterministic with no side effects). When True, Dask can cache results for identical inputs. Set to False only if the operation has side effects or is non-deterministic.	True
**params	Any	Any. Operation-specific parameters	{}

Source code in wandas/processing/base.py

def __init__(self, sampling_rate: float, *, pure: bool = True, **params: Any):
    """
    Initialize AudioOperation.

    Args:
        sampling_rate: float. Sampling rate (Hz)
        pure: bool, default=True. Whether the operation is pure (deterministic with no side effects).
            When True, Dask can cache results for identical inputs.
            Set to False only if the operation has side effects or is non-deterministic.
        **params: Any. Operation-specific parameters
    """
    object.__setattr__(self, "_sampling_rate", float(sampling_rate))
    object.__setattr__(
        self,
        "_config",
        {key: _snapshot_config_value(value) for key, value in params.items()},
    )
    self.pure = pure

    # Validate parameters during initialization
    self.validate_params()

    # Create processor function (lazy initialization possible)
    self._setup_processor()

    logger.debug(f"Initialized {self.__class__.__name__} operation with params: {params}")

to_params()

Return operation parameters used for lineage and display.

Source code in wandas/processing/base.py

def to_params(self) -> Mapping[str, Any]:
    """Return operation parameters used for lineage and display."""
    return self._config_snapshot()

validate_params()

Validate parameters (raises exception if invalid)

Source code in wandas/processing/base.py

def validate_params(self) -> None:
    """Validate parameters (raises exception if invalid)"""

get_metadata_updates()

Get metadata updates to apply after processing.

This method allows operations to specify how metadata should be updated after processing. By default, no metadata is updated.

Returns:

Name	Type	Description
dict	dict[str, Any]	Dictionary of metadata updates. Can include: - 'sampling_rate': New sampling rate (float) - Other metadata keys as needed

Examples:

Return empty dict for operations that don't change metadata:

>>> return {}

Return new sampling rate for operations that resample:

>>> return {"sampling_rate": self.target_sr}

Notes

This method is called by the framework after processing to update the frame metadata. Subclasses should override this method if they need to update metadata (e.g., changing sampling rate).

Design principle: Operations should use parameters provided at initialization (via init). All necessary information should be available as instance variables.

Source code in wandas/processing/base.py

def get_metadata_updates(self) -> dict[str, Any]:
    """
    Get metadata updates to apply after processing.

    This method allows operations to specify how metadata should be
    updated after processing. By default, no metadata is updated.

    Returns:
        dict: Dictionary of metadata updates. Can include:
            - 'sampling_rate': New sampling rate (float)
            - Other metadata keys as needed

    Examples:
        Return empty dict for operations that don't change metadata:

        >>> return {}

        Return new sampling rate for operations that resample:

        >>> return {"sampling_rate": self.target_sr}

    Notes:
        This method is called by the framework after processing to update
        the frame metadata. Subclasses should override this method if they
        need to update metadata (e.g., changing sampling rate).

        Design principle: Operations should use parameters provided at
        initialization (via __init__). All necessary information should be
        available as instance variables.
    """
    return {}

get_display_name()

Get display name for the operation for use in channel labels.

Returns _display if the subclass sets it, otherwise None (which tells the framework to fall back to the name class variable). Subclasses with dynamic display names can still override this method.

Source code in wandas/processing/base.py

def get_display_name(self) -> str | None:
    """
    Get display name for the operation for use in channel labels.

    Returns ``_display`` if the subclass sets it, otherwise ``None``
    (which tells the framework to fall back to the ``name`` class
    variable).  Subclasses with dynamic display names can still
    override this method.
    """
    return getattr(self, "_display", None)

calculate_output_shape(input_shape)

Calculate output data shape after operation.

The default returns input_shape unchanged, which is correct for the majority of operations (filters, effects, weighting, etc.). Subclasses that alter the shape (e.g. FFT, STFT, resampling) must override this method.

Parameters:

Name	Type	Description	Default
input_shape	tuple[int, ...]	tuple. Input data shape	required

Returns:

Name	Type	Description
tuple	tuple[int, ...]	Output data shape

Source code in wandas/processing/base.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """
    Calculate output data shape after operation.

    The default returns *input_shape* unchanged, which is correct for the
    majority of operations (filters, effects, weighting, etc.).
    Subclasses that alter the shape (e.g. FFT, STFT, resampling) **must**
    override this method.

    Args:
        input_shape: tuple. Input data shape

    Returns:
        tuple: Output data shape
    """
    return input_shape

calculate_output_dtype(input_dtype, *input_dtypes)

Calculate output dtype metadata after operation.

Source code in wandas/processing/base.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    """Calculate output dtype metadata after operation."""
    return np.result_type(input_dtype, *input_dtypes)

process(data, *inputs)

Execute operation lazily on Frame-internal channel-first Dask arrays.

data must be the lazy array held by a Frame, with a leading channel axis such as (channels, samples). Direct 1-D lazy input is not part of this API; use a Frame operation or reshape direct lazy inputs to add a channel axis before calling process(). Multi-input operations pass additional channel-first Dask arrays through *inputs.

Source code in wandas/processing/base.py

def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
    """
    Execute operation lazily on Frame-internal channel-first Dask arrays.

    ``data`` must be the lazy array held by a Frame, with a leading channel
    axis such as ``(channels, samples)``. Direct 1-D lazy input is not part
    of this API; use a Frame operation or reshape direct lazy inputs to add
    a channel axis before calling ``process()``. Multi-input operations
    pass additional channel-first Dask arrays through ``*inputs``.
    """
    self._validate_process_inputs(data, *inputs)
    logger.debug("Adding delayed operation to computation graph")
    output_shape = self.calculate_output_shape(data.shape)
    output_dtype = self.calculate_output_dtype(data.dtype, *(input_data.dtype for input_data in inputs))
    return self._build_execution_graph(
        data,
        inputs,
        output_shape=output_shape,
        output_dtype=output_dtype,
    )

ChannelIndependentAudioOperation

Bases: AudioOperation[InputArrayType, OutputArrayType]

Base class for operations whose channels are numerically independent.

For every supported input, a conforming operation satisfies the semantic equivalence

op(all_channels) == concatenate(op(channel) for channel in each_channel).

Subclasses must preserve this independence when overriding :meth:_process. The kernel must also accept a complete multi-channel tensor because graph construction can conservatively use whole-frame execution.

The class expresses numerical semantics, not a public scheduler, chunk, or task-topology guarantee. The current implementation may evaluate eligible unary, channel-axis-preserving inputs independently by channel. Unknown or zero channel counts, runtime inputs, and channel-axis-changing outputs use the whole-frame graph without changing the subclass contract.

Cross-channel algorithms, such as common-mode removal, must subclass :class:AudioOperation instead.

Source code in wandas/processing/base.py

class ChannelIndependentAudioOperation(AudioOperation[InputArrayType, OutputArrayType]):
    """Base class for operations whose channels are numerically independent.

    For every supported input, a conforming operation satisfies the semantic
    equivalence

    ``op(all_channels) == concatenate(op(channel) for channel in each_channel)``.

    Subclasses must preserve this independence when overriding :meth:`_process`.
    The kernel must also accept a complete multi-channel tensor because graph
    construction can conservatively use whole-frame execution.

    The class expresses numerical semantics, not a public scheduler, chunk, or
    task-topology guarantee. The current implementation may evaluate eligible
    unary, channel-axis-preserving inputs independently by channel. Unknown or
    zero channel counts, runtime inputs, and channel-axis-changing outputs use
    the whole-frame graph without changing the subclass contract.

    Cross-channel algorithms, such as common-mode removal, must subclass
    :class:`AudioOperation` instead.
    """

    def _build_execution_graph(
        self,
        data: DaArray,
        inputs: tuple[DaArray, ...],
        *,
        output_shape: tuple[int, ...],
        output_dtype: np.dtype[Any],
    ) -> DaArray:
        result = _try_build_channelwise_graph(
            self,
            data,
            inputs,
            output_shape=output_shape,
            output_dtype=output_dtype,
        )
        if result is not None:
            return result
        return super()._build_execution_graph(
            data,
            inputs,
            output_shape=output_shape,
            output_dtype=output_dtype,
        )

Astype

Bases: ChannelIndependentAudioOperation[Any, Any]

Convert a raw Frame tensor to a supported real or complex floating dtype.

The eager kernel is channel-independent, preserves shape, and never mutates its input. :meth:process builds a lazy Dask graph whose dtype metadata is the exact selected target before computation. Real or integer inputs can produce float32/float64; complex inputs can produce complex64/complex128.

Parameters:

Name	Type	Description	Default
sampling_rate	float	Sampling rate in Hz. It is preserved and does not affect the numerical cast.	required
dtype	DTypeLike	Supported target NumPy dtype or equivalent dtype-like value.	required

Raises:

Type	Description
TypeError	If dtype is not understood by NumPy.
ValueError	If the target is unsupported or :meth:process receives an input whose real/complex domain does not match it.

Source code in wandas/processing/conversion.py

class Astype(ChannelIndependentAudioOperation[Any, Any]):
    """Convert a raw Frame tensor to a supported real or complex floating dtype.

    The eager kernel is channel-independent, preserves shape, and never mutates
    its input. :meth:`process` builds a lazy Dask graph whose dtype metadata is
    the exact selected target before computation. Real or integer inputs can
    produce float32/float64; complex inputs can produce complex64/complex128.

    Args:
        sampling_rate: Sampling rate in Hz. It is preserved and does not affect
            the numerical cast.
        dtype: Supported target NumPy dtype or equivalent dtype-like value.

    Raises:
        TypeError: If *dtype* is not understood by NumPy.
        ValueError: If the target is unsupported or :meth:`process` receives an
            input whose real/complex domain does not match it.
    """

    name = "astype"
    _display = "astype"

    def __init__(self, sampling_rate: float, dtype: npt.DTypeLike) -> None:
        """Initialize a dtype conversion with a canonical target dtype."""
        super().__init__(sampling_rate, dtype=_normalize_target_dtype(dtype))

    @property
    def dtype(self) -> str:
        """Return the canonical target dtype name."""
        return self._config_value("dtype")

    def validate_params(self) -> None:
        """Reject unsupported target representations at construction time."""
        if self.dtype not in _SUPPORTED_TARGET_DTYPES:
            raise ValueError(
                "Unsupported dtype for astype\n"
                f"  Got: {self.dtype}\n"
                "  Expected: float32, float64, complex64, or complex128\n"
                "Choose a supported floating representation."
            )

    def calculate_output_dtype(
        self,
        input_dtype: np.dtype[Any],
        *input_dtypes: np.dtype[Any],
    ) -> np.dtype[Any]:
        """Return exact output metadata after validating the source domain."""
        del input_dtypes
        return np.dtype(_normalize_astype_dtype(input_dtype, self.dtype))

    def _build_execution_graph(
        self,
        data: DaArray,
        inputs: tuple[DaArray, ...],
        *,
        output_shape: tuple[int, ...],
        output_dtype: np.dtype[Any],
    ) -> DaArray:
        """Cast each existing Dask block without changing chunk boundaries."""
        del inputs, output_shape
        return data.astype(output_dtype)

    def _process(self, data: np.ndarray[Any, Any]) -> np.ndarray[Any, Any]:
        """Convert one eager channel-first tensor without mutating the input."""
        target = _normalize_astype_dtype(data.dtype, self.dtype)
        return data.astype(target, copy=False)

Attributes

name = 'astype' class-attribute instance-attribute

dtype property

Return the canonical target dtype name.

Functions

__init__(sampling_rate, dtype)

Initialize a dtype conversion with a canonical target dtype.

Source code in wandas/processing/conversion.py

def __init__(self, sampling_rate: float, dtype: npt.DTypeLike) -> None:
    """Initialize a dtype conversion with a canonical target dtype."""
    super().__init__(sampling_rate, dtype=_normalize_target_dtype(dtype))

validate_params()

Reject unsupported target representations at construction time.

Source code in wandas/processing/conversion.py

def validate_params(self) -> None:
    """Reject unsupported target representations at construction time."""
    if self.dtype not in _SUPPORTED_TARGET_DTYPES:
        raise ValueError(
            "Unsupported dtype for astype\n"
            f"  Got: {self.dtype}\n"
            "  Expected: float32, float64, complex64, or complex128\n"
            "Choose a supported floating representation."
        )

calculate_output_dtype(input_dtype, *input_dtypes)

Return exact output metadata after validating the source domain.

Source code in wandas/processing/conversion.py

def calculate_output_dtype(
    self,
    input_dtype: np.dtype[Any],
    *input_dtypes: np.dtype[Any],
) -> np.dtype[Any]:
    """Return exact output metadata after validating the source domain."""
    del input_dtypes
    return np.dtype(_normalize_astype_dtype(input_dtype, self.dtype))

AddWithSNR

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Addition operation considering SNR

Source code in wandas/processing/effects.py

class AddWithSNR(AudioOperation[NDArrayReal, NDArrayReal]):
    """Addition operation considering SNR"""

    name = "add_with_snr"
    _display = "+SNR"
    _expected_input_count = 2

    def __init__(self, sampling_rate: float, snr: float = 1.0):
        """
        Initialize addition operation considering SNR

        Args:
            sampling_rate: float. Sampling rate (Hz)
            snr: float. Signal-to-noise ratio (dB)
        """
        super().__init__(sampling_rate, snr=snr)
        logger.debug(f"Initialized AddWithSNR operation with SNR: {snr} dB")

    @property
    def snr(self) -> float:
        """Signal-to-noise ratio captured at operation construction time."""
        return self._config_value("snr")

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        """Promote SNR mixing to at least float32 precision."""
        return np.result_type(input_dtype, *input_dtypes, np.float32)

    def _process(self, x: NDArrayReal, other: NDArrayReal) -> NDArrayReal:
        """Perform addition processing considering SNR."""
        logger.debug(f"Applying SNR-based addition with shape: {x.shape}")
        output_dtype = self.calculate_output_dtype(x.dtype, other.dtype)
        clean = np.asarray(x, dtype=output_dtype)
        noise = np.asarray(other, dtype=output_dtype)

        clean_rms = util.calculate_rms(clean)
        other_rms = util.calculate_rms(noise)
        desired_noise_rms = util.calculate_desired_noise_rms(clean_rms, self.snr)
        gain = np.zeros_like(desired_noise_rms, dtype=output_dtype)
        np.divide(desired_noise_rms, other_rms, out=gain, where=other_rms != 0)
        result: NDArrayReal = clean + noise * gain
        return np.asarray(result, dtype=output_dtype)

Attributes

name = 'add_with_snr' class-attribute instance-attribute

snr property

Signal-to-noise ratio captured at operation construction time.

Functions

__init__(sampling_rate, snr=1.0)

Initialize addition operation considering SNR

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate (Hz)	required
snr	float	float. Signal-to-noise ratio (dB)	1.0

Source code in wandas/processing/effects.py

def __init__(self, sampling_rate: float, snr: float = 1.0):
    """
    Initialize addition operation considering SNR

    Args:
        sampling_rate: float. Sampling rate (Hz)
        snr: float. Signal-to-noise ratio (dB)
    """
    super().__init__(sampling_rate, snr=snr)
    logger.debug(f"Initialized AddWithSNR operation with SNR: {snr} dB")

calculate_output_dtype(input_dtype, *input_dtypes)

Promote SNR mixing to at least float32 precision.

Source code in wandas/processing/effects.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    """Promote SNR mixing to at least float32 precision."""
    return np.result_type(input_dtype, *input_dtypes, np.float32)

HpssHarmonic

Bases: _HpssBase

HPSS Harmonic operation

Source code in wandas/processing/effects.py

class HpssHarmonic(_HpssBase):
    """HPSS Harmonic operation"""

    name = "hpss_harmonic"
    _extract_func = "harmonic"
    _display = "Hrm"

Attributes

name = 'hpss_harmonic' class-attribute instance-attribute

HpssPercussive

Bases: _HpssBase

HPSS Percussive operation

Source code in wandas/processing/effects.py

class HpssPercussive(_HpssBase):
    """HPSS Percussive operation"""

    name = "hpss_percussive"
    _extract_func = "percussive"
    _display = "Prc"

Attributes

name = 'hpss_percussive' class-attribute instance-attribute

AWeighting

Bases: ChannelIndependentAudioOperation[NDArrayReal, NDArrayReal]

Apply the implemented digital A-frequency-weighting curve.

The output is a linear waveform. This operation does not calculate RMS, convert to dB, or establish sound-level-meter conformance.

Source code in wandas/processing/filters.py

class AWeighting(ChannelIndependentAudioOperation[NDArrayReal, NDArrayReal]):
    """Apply the implemented digital A-frequency-weighting curve.

    The output is a linear waveform. This operation does not calculate RMS,
    convert to dB, or establish sound-level-meter conformance.
    """

    name = "a_weighting"
    _display = "Aw"

    def __init__(self, sampling_rate: float):
        """
        Initialize A-weighting filter

        Args:
            sampling_rate: float. Sampling rate (Hz)
        """
        super().__init__(sampling_rate)

    _keeps_input_precision = True

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        return precision_real_dtype(input_dtype)

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Create processor function for A-weighting filter"""
        logger.debug(f"Applying A-weighting to array with shape: {x.shape}")
        result = A_weight(x, self.sampling_rate)

        # Handle case where A_weight returns a tuple
        if isinstance(result, tuple):
            # Use the first element of the tuple
            result = result[0]

        logger.debug(f"A-weighting applied, returning result with shape: {result.shape}")
        return np.array(result)

Attributes

name = 'a_weighting' class-attribute instance-attribute

Functions

__init__(sampling_rate)

Initialize A-weighting filter

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate (Hz)	required

Source code in wandas/processing/filters.py

def __init__(self, sampling_rate: float):
    """
    Initialize A-weighting filter

    Args:
        sampling_rate: float. Sampling rate (Hz)
    """
    super().__init__(sampling_rate)

calculate_output_dtype(input_dtype, *input_dtypes)

Source code in wandas/processing/filters.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    return precision_real_dtype(input_dtype)

HighPassFilter

Bases: _ButterworthFilter

High-pass filter operation

Source code in wandas/processing/filters.py

class HighPassFilter(_ButterworthFilter):
    """High-pass filter operation"""

    name = "highpass_filter"
    _btype = "high"
    _display = "hpf"

Attributes

name = 'highpass_filter' class-attribute instance-attribute

LowPassFilter

Bases: _ButterworthFilter

Low-pass filter operation

Source code in wandas/processing/filters.py

class LowPassFilter(_ButterworthFilter):
    """Low-pass filter operation"""

    name = "lowpass_filter"
    _btype = "low"
    _display = "lpf"

Attributes

name = 'lowpass_filter' class-attribute instance-attribute

ABS

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Absolute value operation

Source code in wandas/processing/stats.py

class ABS(AudioOperation[NDArrayReal, NDArrayReal]):
    """Absolute value operation"""

    name = "abs"
    _display = "abs"

    def __init__(self, sampling_rate: float):
        """
        Initialize absolute value operation

        Args:
            sampling_rate: float. Sampling rate (Hz)
        """
        super().__init__(sampling_rate)

    def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
        (values,) = integer_arithmetic_as_float64(data)
        return da.abs(values)

Attributes

name = 'abs' class-attribute instance-attribute

Functions

__init__(sampling_rate)

Initialize absolute value operation

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate (Hz)	required

Source code in wandas/processing/stats.py

def __init__(self, sampling_rate: float):
    """
    Initialize absolute value operation

    Args:
        sampling_rate: float. Sampling rate (Hz)
    """
    super().__init__(sampling_rate)

process(data, *inputs)

Source code in wandas/processing/stats.py

def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
    (values,) = integer_arithmetic_as_float64(data)
    return da.abs(values)

ChannelDifference

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Channel difference calculation operation

Source code in wandas/processing/stats.py

class ChannelDifference(AudioOperation[NDArrayReal, NDArrayReal]):
    """Channel difference calculation operation"""

    name = "channel_difference"
    _display = "diff"

    def __init__(self, sampling_rate: float, other_channel: int = 0):
        """
        Initialize channel difference calculation

        Args:
            sampling_rate: float. Sampling rate (Hz)
            other_channel: int. Channel to calculate difference with, default is 0
        """
        super().__init__(sampling_rate, other_channel=other_channel)

    @property
    def other_channel(self) -> int:
        """Other channel index captured at operation construction time."""
        return self._config_value("other_channel")

    def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
        other_channel = self.other_channel
        if not -data.shape[0] <= other_channel < data.shape[0]:
            raise IndexError("Channel index out of range")
        (values,) = integer_arithmetic_as_float64(data)
        return values - values[other_channel]

Attributes

name = 'channel_difference' class-attribute instance-attribute

other_channel property

Other channel index captured at operation construction time.

Functions

__init__(sampling_rate, other_channel=0)

Initialize channel difference calculation

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate (Hz)	required
other_channel	int	int. Channel to calculate difference with, default is 0	0

Source code in wandas/processing/stats.py

def __init__(self, sampling_rate: float, other_channel: int = 0):
    """
    Initialize channel difference calculation

    Args:
        sampling_rate: float. Sampling rate (Hz)
        other_channel: int. Channel to calculate difference with, default is 0
    """
    super().__init__(sampling_rate, other_channel=other_channel)

process(data, *inputs)

Source code in wandas/processing/stats.py

def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
    other_channel = self.other_channel
    if not -data.shape[0] <= other_channel < data.shape[0]:
        raise IndexError("Channel index out of range")
    (values,) = integer_arithmetic_as_float64(data)
    return values - values[other_channel]

Mean

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Mean calculation

Source code in wandas/processing/stats.py

class Mean(AudioOperation[NDArrayReal, NDArrayReal]):
    """Mean calculation"""

    name = "mean"
    _display = "mean"

    def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
        return data.mean(axis=0, keepdims=True)

Attributes

name = 'mean' class-attribute instance-attribute

Functions

process(data, *inputs)

Source code in wandas/processing/stats.py

def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
    return data.mean(axis=0, keepdims=True)

Power

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Power operation

Source code in wandas/processing/stats.py

class Power(AudioOperation[NDArrayReal, NDArrayReal]):
    """Power operation"""

    name = "power"
    _display = "pow"

    def __init__(self, sampling_rate: float, exponent: float):
        """
        Initialize power operation

        Args:
            sampling_rate: float. Sampling rate (Hz)
            exponent: float. Power exponent
        """
        super().__init__(sampling_rate, exponent=exponent)

    @property
    def exponent(self) -> float:
        """Exponent captured at operation construction time."""
        return self._config_value("exponent")

    @property
    def exp(self) -> float:
        """Backward-compatible read-only alias for the captured exponent."""
        return self.exponent

    def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
        values, exponent = integer_arithmetic_as_float64(data, self.exponent)
        return da.power(values, exponent)

Attributes

name = 'power' class-attribute instance-attribute

exponent property

Exponent captured at operation construction time.

exp property

Backward-compatible read-only alias for the captured exponent.

Functions

__init__(sampling_rate, exponent)

Initialize power operation

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate (Hz)	required
exponent	float	float. Power exponent	required

Source code in wandas/processing/stats.py

def __init__(self, sampling_rate: float, exponent: float):
    """
    Initialize power operation

    Args:
        sampling_rate: float. Sampling rate (Hz)
        exponent: float. Power exponent
    """
    super().__init__(sampling_rate, exponent=exponent)

process(data, *inputs)

Source code in wandas/processing/stats.py

def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
    values, exponent = integer_arithmetic_as_float64(data, self.exponent)
    return da.power(values, exponent)

Sum

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Sum calculation

Source code in wandas/processing/stats.py

class Sum(AudioOperation[NDArrayReal, NDArrayReal]):
    """Sum calculation"""

    name = "sum"
    _display = "sum"

    def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
        return data.sum(axis=0, keepdims=True)

Attributes

name = 'sum' class-attribute instance-attribute

Functions

process(data, *inputs)

Source code in wandas/processing/stats.py

def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
    return data.sum(axis=0, keepdims=True)

ReSampling

Bases: ChannelIndependentAudioOperation[NDArrayReal, NDArrayReal]

Resampling operation

Source code in wandas/processing/temporal.py

class ReSampling(ChannelIndependentAudioOperation[NDArrayReal, NDArrayReal]):
    """Resampling operation"""

    name = "resampling"
    _display = "rs"

    def __init__(self, sampling_rate: float, target_sr: float):
        """
        Initialize a resampling operation.

        Args:
            sampling_rate (float): Source sampling rate in Hz.
            target_sr (float): Target sampling rate in Hz.

        Raises:
            ValueError: If ``sampling_rate`` or ``target_sr`` is not positive.
        """
        validate_sampling_rate(sampling_rate, "source sampling rate")
        validate_sampling_rate(target_sr, "target sampling rate")
        super().__init__(sampling_rate, target_sr=target_sr)

    @property
    def target_sr(self) -> float:
        """Target sampling rate captured at operation construction time."""
        return self._config_value("target_sr")

    def get_metadata_updates(self) -> dict[str, Any]:
        """
        Update sampling rate to target sampling rate.

        Returns:
            dict: Metadata updates with the new sampling rate.

        Notes:
            Resampling always produces output at ``target_sr``, regardless of the
            input sampling rate.
        """
        return {"sampling_rate": self.target_sr}

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        """
        Calculate the output data shape after the operation.

        Args:
            input_shape (tuple[int, ...]): Input data shape.

        Returns:
            tuple[int, ...]: Output data shape.
        """
        # Calculate length after resampling using exact decimal sampling-rate ratio.
        ratio = _resampling_fraction(self.sampling_rate, self.target_sr)
        n_samples = _ceil_resampled_length(input_shape[-1], ratio)
        return (*input_shape[:-1], n_samples)

    @staticmethod
    def _output_dtype(input_dtype: np.dtype[Any]) -> np.dtype[Any]:
        dtype = np.dtype(input_dtype)
        if dtype.kind == "f":
            return dtype
        return np.dtype(np.float64)

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Create processor function for resampling operation"""
        logger.debug(f"Applying resampling to array with shape: {x.shape}")
        up, down = _resampling_ratio(self.sampling_rate, self.target_sr)
        target_len = self.calculate_output_shape(x.shape)[-1]
        poly_len = _ceil_resampled_length(x.shape[-1], Fraction(up, down))
        if poly_len == target_len:
            result: NDArrayReal = scipy.signal.resample_poly(x, up, down, axis=-1)
        else:
            result = scipy.signal.resample(x, target_len, axis=-1)
        logger.debug(f"Resampling applied, returning result with shape: {result.shape}")
        return result

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        """Return resampling output dtype metadata."""
        return self._output_dtype(input_dtype)

Attributes

name = 'resampling' class-attribute instance-attribute

target_sr property

Target sampling rate captured at operation construction time.

Functions

__init__(sampling_rate, target_sr)

Initialize a resampling operation.

Parameters:

Name	Type	Description	Default
sampling_rate	float	Source sampling rate in Hz.	required
target_sr	float	Target sampling rate in Hz.	required

Raises:

Type	Description
ValueError	If sampling_rate or target_sr is not positive.

Source code in wandas/processing/temporal.py

def __init__(self, sampling_rate: float, target_sr: float):
    """
    Initialize a resampling operation.

    Args:
        sampling_rate (float): Source sampling rate in Hz.
        target_sr (float): Target sampling rate in Hz.

    Raises:
        ValueError: If ``sampling_rate`` or ``target_sr`` is not positive.
    """
    validate_sampling_rate(sampling_rate, "source sampling rate")
    validate_sampling_rate(target_sr, "target sampling rate")
    super().__init__(sampling_rate, target_sr=target_sr)

get_metadata_updates()

Update sampling rate to target sampling rate.

Returns:

Name	Type	Description
dict	dict[str, Any]	Metadata updates with the new sampling rate.

Notes

Resampling always produces output at target_sr, regardless of the input sampling rate.

Source code in wandas/processing/temporal.py

def get_metadata_updates(self) -> dict[str, Any]:
    """
    Update sampling rate to target sampling rate.

    Returns:
        dict: Metadata updates with the new sampling rate.

    Notes:
        Resampling always produces output at ``target_sr``, regardless of the
        input sampling rate.
    """
    return {"sampling_rate": self.target_sr}

calculate_output_shape(input_shape)

Calculate the output data shape after the operation.

Parameters:

Name	Type	Description	Default
input_shape	tuple[int, ...]	Input data shape.	required

Returns:

Type	Description
tuple[int, ...]	tuple[int, ...]: Output data shape.

Source code in wandas/processing/temporal.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """
    Calculate the output data shape after the operation.

    Args:
        input_shape (tuple[int, ...]): Input data shape.

    Returns:
        tuple[int, ...]: Output data shape.
    """
    # Calculate length after resampling using exact decimal sampling-rate ratio.
    ratio = _resampling_fraction(self.sampling_rate, self.target_sr)
    n_samples = _ceil_resampled_length(input_shape[-1], ratio)
    return (*input_shape[:-1], n_samples)

calculate_output_dtype(input_dtype, *input_dtypes)

Return resampling output dtype metadata.

Source code in wandas/processing/temporal.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    """Return resampling output dtype metadata."""
    return self._output_dtype(input_dtype)

RmsTrend

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Windowed linear RMS or reference-relative RMS amplitude level.

The operation accepts arrays shaped (channels, samples) and returns (channels, frames) using centered, zero-padded windows. dB=False returns RMS in the input unit. dB=True returns 20 * log10(max(RMS / ref, 1e-12)), bounded below by -240 dB, with one scalar reference shared across channels or one reference per channel. Applying Aw changes the frequency weighting before RMS; it does not establish instrument conformance. The operation is lazy when used through a Frame and preserves the input dtype contract by returning floating data.

Source code in wandas/processing/temporal.py

class RmsTrend(AudioOperation[NDArrayReal, NDArrayReal]):
    """Windowed linear RMS or reference-relative RMS amplitude level.

    The operation accepts arrays shaped ``(channels, samples)`` and returns
    ``(channels, frames)`` using centered, zero-padded windows. ``dB=False``
    returns RMS in the input unit. ``dB=True`` returns
    ``20 * log10(max(RMS / ref, 1e-12))``, bounded below by -240 dB, with one
    scalar reference shared across channels or one reference per channel.
    Applying ``Aw`` changes the frequency weighting before RMS; it does not
    establish instrument conformance. The operation is lazy when used through
    a Frame and preserves the input dtype contract by returning floating data.
    """

    name = "rms_trend"
    _display = "RMS"
    _validate_reference_count = True
    _calibration_scale: tuple[float, ...]

    def __init__(
        self,
        sampling_rate: float,
        frame_length: int = 2048,
        hop_length: int = 512,
        ref: list[float] | float = 1.0,
        dB: bool = False,
        Aw: bool = False,
        *,
        _calibration_scale: list[float] | float | NDArrayReal = 1.0,
    ) -> None:
        """Initialize a centered windowed RMS operation.

        Args:
            sampling_rate: Input sampling rate in Hz.
            frame_length: Window length in samples. Defaults to 2048.
            hop_length: Distance between output frames in samples. Defaults to
                512. The output sampling rate is ``sampling_rate / hop_length``.
            ref: Positive finite amplitude reference, either one scalar or one
                value per channel. For Pa input, ``2e-5`` produces dB SPL.
            dB: If True, return reference-relative amplitude level instead of
                linear RMS amplitude.
            Aw: If True, apply the implemented digital A-weighting filter before
                RMS calculation.
            _calibration_scale: Positive internal amplitude scale supplied by
                calibrated Frame execution. It is not a public recipe
                parameter.

        Raises:
            ValueError: If the sampling or window parameters are invalid, or if
                a reference or calibration scale is not finite and positive.
        """
        ref_array = np.array(ref if isinstance(ref, list) else [ref], dtype=float)
        if ref_array.size == 0 or np.any(~np.isfinite(ref_array)) or np.any(ref_array <= 0):
            raise ValueError(
                "Invalid RMS level reference\n"
                f"  Got: {ref_array.tolist()}\n"
                "  Expected: Positive finite reference values\n"
                "Reference-relative dB output requires a positive finite amplitude reference."
            )
        calibration_scale = _validated_calibration_scale(
            _calibration_scale,
            operation_label="RMS level",
        )
        super().__init__(
            sampling_rate,
            frame_length=frame_length,
            hop_length=hop_length,
            dB=dB,
            Aw=Aw,
            ref=ref_array,
        )
        object.__setattr__(self, "_calibration_scale", calibration_scale)

    @property
    def frame_length(self) -> int:
        """Frame length captured at operation construction time."""
        return self._config_value("frame_length")

    @property
    def hop_length(self) -> int:
        """Hop length captured at operation construction time."""
        return self._config_value("hop_length")

    @property
    def dB(self) -> bool:  # noqa: N802
        """Whether output is converted to decibels."""
        return self._config_value("dB")

    @property
    def Aw(self) -> bool:  # noqa: N802
        """Whether A-weighting is applied before RMS calculation."""
        return self._config_value("Aw")

    @property
    def ref(self) -> NDArrayReal:
        """Reference values captured at operation construction time."""
        return self._config_value("ref")

    def _reference_values(self, n_channels: int) -> NDArrayReal:
        """Return one validated reference value for each channel."""
        ref_config = self._config["ref"]
        if ref_config.size == 1:
            ref = np.repeat(ref_config, n_channels)
        elif ref_config.size == n_channels:
            ref = ref_config
        else:
            raise ValueError(
                "Reference count mismatch\n"
                f"  Got: {ref_config.size} reference values for {n_channels} channels\n"
                "  Expected: One shared reference or one reference per channel\n"
                "Provide ref as a scalar or a list matching the number of channels."
            )
        return np.asarray(ref, dtype=np.float64)

    def _calibration_scale_values(self, n_channels: int) -> NDArrayReal:
        """Return one validated internal amplitude scale for each channel."""
        return _calibration_scale_values(self._calibration_scale, n_channels)

    def get_metadata_updates(self) -> dict[str, Any]:
        """Return metadata updates for the frame-rate change.

        Returns:
            A mapping containing ``sampling_rate`` set to
                ``sampling_rate / hop_length``. The returned value describes the
                window centers, not the original sample rate.
        """
        new_sr = self.sampling_rate / self.hop_length
        return {"sampling_rate": new_sr}

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        """Calculate the centered, zero-padded output shape.

        Args:
            input_shape: Input shape whose last dimension contains samples;
                the usual Frame shape is ``(channels, samples)``.

        Returns:
            The input leading dimensions followed by the number of centered
                windows, so a two-dimensional input returns ``(channels, frames)``.

        Raises:
            ValueError: If dB output uses a reference or calibration scale that
                cannot be broadcast to the input channel count.
        """
        if self.dB and self._validate_reference_count:
            self._reference_values(input_shape[0])
            self._calibration_scale_values(input_shape[0])
        n_frames = _centered_frame_count(
            input_shape[-1],
            self.frame_length,
            self.hop_length,
        )
        return (*input_shape[:-1], n_frames)

    _keeps_input_precision = True

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Create processor function for RMS calculation"""
        logger.debug(f"Applying RMS to array with shape: {x.shape}")

        weighted_log_amplitude: NDArrayReal | None = None
        if self.Aw:
            # Apply A-weighting
            weighting_input = x
            if self.dB:
                weighting_input = np.asarray(x, dtype=np.float64)
                first_unsafe_samples = _level_filter_first_unsafe_samples(weighting_input)
                if np.any(first_unsafe_samples >= 0):
                    weighted_log_amplitude = _frequency_weight_log_amplitude(
                        weighting_input,
                        self.sampling_rate,
                        curve="A",
                        first_unsafe_samples=first_unsafe_samples,
                    )
            if weighted_log_amplitude is None:
                _x = A_weight(weighting_input, self.sampling_rate)
                if isinstance(_x, np.ndarray):
                    x = _x
                elif isinstance(_x, tuple):
                    x = _x[0]
                else:
                    raise ValueError("A_weighting returned an unexpected type.")

        if self.dB:
            references = self._reference_values(x.shape[0])
            calibration_scale = self._calibration_scale_values(x.shape[0])
            if weighted_log_amplitude is None:
                log_rms = _frame_log_rms(
                    x,
                    frame_length=self.frame_length,
                    hop_length=self.hop_length,
                )
            else:
                log_rms = _frame_log_rms_from_log_amplitude(
                    weighted_log_amplitude,
                    frame_length=self.frame_length,
                    hop_length=self.hop_length,
                )
            with np.errstate(divide="ignore", invalid="ignore"):
                np.add(
                    log_rms,
                    np.log(calibration_scale)[..., np.newaxis],
                    out=log_rms,
                )
                np.subtract(
                    log_rms,
                    np.log(references)[..., np.newaxis],
                    out=log_rms,
                )
            result = _bounded_db_from_log_ratio(
                log_rms,
                scale=20.0,
                ratio_floor=DB_FLOOR,
            )
        else:
            # Preserve the released linear RMS path exactly. Numerical scaling
            # belongs only to the versioned dB contract.
            result = _frame_rms(
                x,
                frame_length=self.frame_length,
                hop_length=self.hop_length,
            )
        logger.debug(f"RMS applied, returning result with shape: {result.shape}")
        return result

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        """Return RMS trend output dtype metadata."""
        return precision_real_dtype(input_dtype)

Attributes

name = 'rms_trend' class-attribute instance-attribute

frame_length property

Frame length captured at operation construction time.

hop_length property

Hop length captured at operation construction time.

dB property

Whether output is converted to decibels.

Aw property

Whether A-weighting is applied before RMS calculation.

ref property

Reference values captured at operation construction time.

Functions

__init__(sampling_rate, frame_length=2048, hop_length=512, ref=1.0, dB=False, Aw=False, *, _calibration_scale=1.0)

Initialize a centered windowed RMS operation.

Parameters:

Name	Type	Description	Default
sampling_rate	float	Input sampling rate in Hz.	required
frame_length	int	Window length in samples. Defaults to 2048.	2048
hop_length	int	Distance between output frames in samples. Defaults to 512. The output sampling rate is sampling_rate / hop_length.	512
ref	list[float] | float	Positive finite amplitude reference, either one scalar or one value per channel. For Pa input, 2e-5 produces dB SPL.	1.0
dB	bool	If True, return reference-relative amplitude level instead of linear RMS amplitude.	False
Aw	bool	If True, apply the implemented digital A-weighting filter before RMS calculation.	False
_calibration_scale	list[float] | float | NDArrayReal	Positive internal amplitude scale supplied by calibrated Frame execution. It is not a public recipe parameter.	1.0

Raises:

Type	Description
ValueError	If the sampling or window parameters are invalid, or if a reference or calibration scale is not finite and positive.

Source code in wandas/processing/temporal.py

def __init__(
    self,
    sampling_rate: float,
    frame_length: int = 2048,
    hop_length: int = 512,
    ref: list[float] | float = 1.0,
    dB: bool = False,
    Aw: bool = False,
    *,
    _calibration_scale: list[float] | float | NDArrayReal = 1.0,
) -> None:
    """Initialize a centered windowed RMS operation.

    Args:
        sampling_rate: Input sampling rate in Hz.
        frame_length: Window length in samples. Defaults to 2048.
        hop_length: Distance between output frames in samples. Defaults to
            512. The output sampling rate is ``sampling_rate / hop_length``.
        ref: Positive finite amplitude reference, either one scalar or one
            value per channel. For Pa input, ``2e-5`` produces dB SPL.
        dB: If True, return reference-relative amplitude level instead of
            linear RMS amplitude.
        Aw: If True, apply the implemented digital A-weighting filter before
            RMS calculation.
        _calibration_scale: Positive internal amplitude scale supplied by
            calibrated Frame execution. It is not a public recipe
            parameter.

    Raises:
        ValueError: If the sampling or window parameters are invalid, or if
            a reference or calibration scale is not finite and positive.
    """
    ref_array = np.array(ref if isinstance(ref, list) else [ref], dtype=float)
    if ref_array.size == 0 or np.any(~np.isfinite(ref_array)) or np.any(ref_array <= 0):
        raise ValueError(
            "Invalid RMS level reference\n"
            f"  Got: {ref_array.tolist()}\n"
            "  Expected: Positive finite reference values\n"
            "Reference-relative dB output requires a positive finite amplitude reference."
        )
    calibration_scale = _validated_calibration_scale(
        _calibration_scale,
        operation_label="RMS level",
    )
    super().__init__(
        sampling_rate,
        frame_length=frame_length,
        hop_length=hop_length,
        dB=dB,
        Aw=Aw,
        ref=ref_array,
    )
    object.__setattr__(self, "_calibration_scale", calibration_scale)

get_metadata_updates()

Return metadata updates for the frame-rate change.

Returns:

Type	Description
dict[str, Any]	A mapping containing sampling_rate set to sampling_rate / hop_length. The returned value describes the window centers, not the original sample rate.

Source code in wandas/processing/temporal.py

def get_metadata_updates(self) -> dict[str, Any]:
    """Return metadata updates for the frame-rate change.

    Returns:
        A mapping containing ``sampling_rate`` set to
            ``sampling_rate / hop_length``. The returned value describes the
            window centers, not the original sample rate.
    """
    new_sr = self.sampling_rate / self.hop_length
    return {"sampling_rate": new_sr}

calculate_output_shape(input_shape)

Calculate the centered, zero-padded output shape.

Parameters:

Name	Type	Description	Default
input_shape	tuple[int, ...]	Input shape whose last dimension contains samples; the usual Frame shape is (channels, samples).	required

Returns:

Type	Description
tuple[int, ...]	The input leading dimensions followed by the number of centered windows, so a two-dimensional input returns (channels, frames).

Raises:

Type	Description
ValueError	If dB output uses a reference or calibration scale that cannot be broadcast to the input channel count.

Source code in wandas/processing/temporal.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """Calculate the centered, zero-padded output shape.

    Args:
        input_shape: Input shape whose last dimension contains samples;
            the usual Frame shape is ``(channels, samples)``.

    Returns:
        The input leading dimensions followed by the number of centered
            windows, so a two-dimensional input returns ``(channels, frames)``.

    Raises:
        ValueError: If dB output uses a reference or calibration scale that
            cannot be broadcast to the input channel count.
    """
    if self.dB and self._validate_reference_count:
        self._reference_values(input_shape[0])
        self._calibration_scale_values(input_shape[0])
    n_frames = _centered_frame_count(
        input_shape[-1],
        self.frame_length,
        self.hop_length,
    )
    return (*input_shape[:-1], n_frames)

calculate_output_dtype(input_dtype, *input_dtypes)

Return RMS trend output dtype metadata.

Source code in wandas/processing/temporal.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    """Return RMS trend output dtype metadata."""
    return precision_real_dtype(input_dtype)

SoundLevel

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Frequency- and exponentially time-weighted RMS or level.

The operation applies A, C, or flat Z frequency weighting, smooths squared samples with a 125 ms (Fast) or 1 s (Slow) first-order exponential filter, and returns either the square root (linear RMS) or 10 * log10(max(smoothed_power / ref**2, 1e-20)), bounded below by -200 dB. The result is dB SPL only for pressure in Pa with ref=2e-5. The implementation is not a claim of complete IEC/JIS sound-level-meter conformance. Input and output arrays have the same (channels, samples) shape; Frame execution remains lazy and preserves the input sampling rate.

Source code in wandas/processing/temporal.py

class SoundLevel(AudioOperation[NDArrayReal, NDArrayReal]):
    """Frequency- and exponentially time-weighted RMS or level.

    The operation applies A, C, or flat Z frequency weighting, smooths squared
    samples with a 125 ms (Fast) or 1 s (Slow) first-order exponential filter,
    and returns either the square root (linear RMS) or
    ``10 * log10(max(smoothed_power / ref**2, 1e-20))``, bounded below by
    -200 dB. The result is dB SPL only for pressure in Pa with ``ref=2e-5``.
    The implementation is not a claim of complete IEC/JIS sound-level-meter
    conformance. Input and output arrays have the same ``(channels, samples)``
    shape; Frame execution remains lazy and preserves the input sampling rate.
    """

    name = "sound_level"
    _calibration_scale: tuple[float, ...]

    def __init__(
        self,
        sampling_rate: float,
        ref: list[float] | float | NDArrayReal = 1.0,
        freq_weighting: str | None = "Z",
        time_weighting: str = "Fast",
        dB: bool = False,
        *,
        _calibration_scale: list[float] | float | NDArrayReal = 1.0,
    ) -> None:
        """Initialize a frequency- and time-weighted level operation.

        Args:
            sampling_rate: Input sampling rate in Hz.
            ref: Positive finite amplitude reference, either one scalar or one
                value per channel. For Pa input, ``2e-5`` produces dB SPL.
            freq_weighting: Implemented frequency curve: ``"A"``, ``"C"``, or
                flat ``"Z"``. ``None`` is treated as ``"Z"``.
            time_weighting: Exponential time constant: ``"Fast"`` (125 ms) or
                ``"Slow"`` (1 s). The short forms ``"F"`` and ``"S"`` are
                accepted too.
            dB: If True, return ``10 * log10`` of smoothed power relative to
                ``ref**2``; otherwise return linear weighted RMS in the input
                unit.
            _calibration_scale: Positive internal amplitude scale supplied by
                calibrated Frame execution. It is not a public recipe
                parameter.

        Raises:
            ValueError: If the sampling rate, reference, calibration scale,
                frequency curve, or time weighting is invalid.
        """
        validate_sampling_rate(sampling_rate)
        ref_array = np.atleast_1d(np.array(ref, dtype=float, copy=True))
        if ref_array.size == 0 or np.any(~np.isfinite(ref_array)) or np.any(ref_array <= 0):
            raise ValueError(
                "Invalid sound level reference\n"
                f"  Got: {ref_array.tolist()}\n"
                "  Expected: Positive finite reference values\n"
                "Reference-relative dB output requires a positive finite amplitude reference."
            )
        calibration_scale = _validated_calibration_scale(
            _calibration_scale,
            operation_label="sound level",
        )
        normalized_freq_weighting = self._normalize_freq_weighting(freq_weighting)
        normalized_time_weighting = self._normalize_time_weighting(time_weighting)
        super().__init__(
            sampling_rate,
            ref=ref_array,
            freq_weighting=normalized_freq_weighting,
            time_weighting=normalized_time_weighting,
            dB=dB,
        )
        object.__setattr__(self, "_calibration_scale", calibration_scale)

    @staticmethod
    def _normalize_freq_weighting(freq_weighting: str | None) -> str:
        normalized = "Z" if freq_weighting is None else str(freq_weighting).upper()
        if normalized not in {"A", "C", "Z"}:
            raise ValueError(
                "Invalid frequency weighting\n"
                f"  Got: {freq_weighting!r}\n"
                "  Expected: 'A', 'C', or 'Z'\n"
                "Choose one of the implemented frequency-weighting curves."
            )
        return normalized

    @staticmethod
    def _normalize_time_weighting(time_weighting: str) -> str:
        normalized = str(time_weighting).strip().upper()
        if normalized in {"F", "FAST"}:
            return "Fast"
        if normalized in {"S", "SLOW"}:
            return "Slow"
        raise ValueError(
            "Invalid time weighting\n"
            f"  Got: {time_weighting!r}\n"
            "  Expected: 'Fast' or 'Slow'\n"
            "Choose one of the implemented exponential time constants."
        )

    @property
    def ref(self) -> NDArrayReal:
        """Reference values captured at operation construction time."""
        return self._config_value("ref")

    @property
    def freq_weighting(self) -> str:
        """Frequency weighting captured at operation construction time."""
        return self._config_value("freq_weighting")

    @property
    def time_weighting(self) -> str:
        """Time weighting captured at operation construction time."""
        return self._config_value("time_weighting")

    @property
    def dB(self) -> bool:  # noqa: N802
        """Whether output is converted to decibels."""
        return self._config_value("dB")

    @property
    def time_constant(self) -> float:
        """Return the RC time constant in seconds."""
        return 0.125 if self.time_weighting == "Fast" else 1.0

    @staticmethod
    def _output_dtype(
        input_dtype: np.dtype[Any],
    ) -> np.dtype[np.float32] | np.dtype[np.float64]:
        """Return the floating output dtype for the given input dtype."""
        if np.dtype(input_dtype) == np.dtype(np.float32):
            return np.dtype(np.float32)
        return np.dtype(np.float64)

    def get_display_name(self) -> str:
        """Get display name for the operation for use in channel labels."""
        freq_weighting = self.freq_weighting
        time_weighting = self.time_weighting
        if self.dB:
            return f"L{freq_weighting}{time_weighting[0]}"
        return f"{freq_weighting}{time_weighting[0]}RMS"

    def _reference_values(self, n_channels: int) -> NDArrayReal:
        """Return one validated reference value for each channel."""
        ref_config = self._config["ref"]
        if ref_config.size == 1:
            ref = np.repeat(ref_config, n_channels)
        elif ref_config.size == n_channels:
            ref = ref_config
        else:
            raise ValueError(
                "Reference count mismatch\n"
                f"  Got: {ref_config.size} reference values for {n_channels} channels\n"
                "  Expected: One shared reference or one reference per channel\n"
                "Provide ref as a scalar or a list matching the number of channels."
            )
        return np.asarray(ref, dtype=np.float64)

    def _calibration_scale_values(self, n_channels: int) -> NDArrayReal:
        """Return one validated internal amplitude scale for each channel."""
        return _calibration_scale_values(self._calibration_scale, n_channels)

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        """Validate channel-wise configuration and preserve input shape.

        Args:
            input_shape: Input shape, normally ``(channels, samples)``.

        Returns:
            The unchanged input shape. ``sound_level`` is sample-wise and keeps
                the input sampling rate.

        Raises:
            ValueError: If a per-channel reference or calibration scale cannot
                be broadcast to the input channel count.
        """
        if self.dB:
            self._reference_values(input_shape[0])
            self._calibration_scale_values(input_shape[0])
        return input_shape

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Create processor function for sound level calculation."""
        logger.debug(
            "Applying sound level to array with shape %s using %s/%s weighting",
            x.shape,
            self.freq_weighting,
            self.time_weighting,
        )
        output_dtype = self._output_dtype(x.dtype)
        weighted_input = x if x.dtype == np.float64 else np.asarray(x, dtype=np.float64)
        weighted = weighted_input
        freq_weighting = self.freq_weighting
        weighted_log_amplitude: NDArrayReal | None = None
        if freq_weighting == "Z":
            weighted = weighted_input
        else:
            if self.dB:
                first_unsafe_samples = _level_filter_first_unsafe_samples(weighted_input)
                if np.any(first_unsafe_samples >= 0):
                    weighted_log_amplitude = _frequency_weight_log_amplitude(
                        weighted_input,
                        self.sampling_rate,
                        curve=freq_weighting,
                        first_unsafe_samples=first_unsafe_samples,
                    )
                else:
                    weighted = frequency_weight(weighted_input, self.sampling_rate, curve=freq_weighting)
            else:
                weighted = frequency_weight(weighted_input, self.sampling_rate, curve=freq_weighting)
        alpha = np.asarray(np.exp(-1.0 / (self.sampling_rate * self.time_constant)), dtype=np.float64).item()
        if self.dB:
            references = self._reference_values(weighted_input.shape[0])
            calibration_scale = self._calibration_scale_values(weighted_input.shape[0])
            if weighted_log_amplitude is not None:
                log_smoothed_power = _exponential_power_from_log_amplitude(weighted_log_amplitude, alpha)
            elif _reference_floor_requires_log_power(
                references,
                calibration_scale,
            ) or _requires_scaled_square(
                weighted,
                minimum_power_scale=1.0 - alpha,
            ):
                log_smoothed_power = _exponential_power_log(weighted, alpha)
            else:
                squared = np.square(weighted)
                log_smoothed_power = scipy.signal.lfilter([1.0 - alpha], [1.0, -alpha], squared, axis=-1)
                del squared
                with np.errstate(divide="ignore", invalid="ignore"):
                    np.log(log_smoothed_power, out=log_smoothed_power)
            with np.errstate(divide="ignore", invalid="ignore"):
                np.add(
                    log_smoothed_power,
                    2.0 * np.log(calibration_scale[:, np.newaxis]),
                    out=log_smoothed_power,
                )
                np.subtract(
                    log_smoothed_power,
                    2.0 * np.log(references[:, np.newaxis]),
                    out=log_smoothed_power,
                )
            result = _bounded_db_from_log_ratio(
                log_smoothed_power,
                scale=10.0,
                ratio_floor=MIN_SOUND_LEVEL_POWER_RATIO,
            )
        else:
            # Preserve the released linear RMS path exactly. Numerical scaling
            # belongs only to the versioned dB contract.
            squared = np.square(weighted)
            smoothed = scipy.signal.lfilter([1.0 - alpha], [1.0, -alpha], squared, axis=-1)
            result = np.sqrt(smoothed)
        logger.debug(f"Sound level applied, returning result with shape: {result.shape}")
        return np.asarray(result, dtype=output_dtype)

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        """Return sound level output dtype metadata."""
        return self._output_dtype(input_dtype)

Attributes

name = 'sound_level' class-attribute instance-attribute

ref property

Reference values captured at operation construction time.

freq_weighting property

Frequency weighting captured at operation construction time.

time_weighting property

Time weighting captured at operation construction time.

dB property

Whether output is converted to decibels.

time_constant property

Return the RC time constant in seconds.

Functions

__init__(sampling_rate, ref=1.0, freq_weighting='Z', time_weighting='Fast', dB=False, *, _calibration_scale=1.0)

Initialize a frequency- and time-weighted level operation.

Parameters:

Name	Type	Description	Default
sampling_rate	float	Input sampling rate in Hz.	required
ref	list[float] | float | NDArrayReal	Positive finite amplitude reference, either one scalar or one value per channel. For Pa input, 2e-5 produces dB SPL.	1.0
freq_weighting	str | None	Implemented frequency curve: "A", "C", or flat "Z". None is treated as "Z".	'Z'
time_weighting	str	Exponential time constant: "Fast" (125 ms) or "Slow" (1 s). The short forms "F" and "S" are accepted too.	'Fast'
dB	bool	If True, return 10 * log10 of smoothed power relative to ref**2; otherwise return linear weighted RMS in the input unit.	False
_calibration_scale	list[float] | float | NDArrayReal	Positive internal amplitude scale supplied by calibrated Frame execution. It is not a public recipe parameter.	1.0

Raises:

Type	Description
ValueError	If the sampling rate, reference, calibration scale, frequency curve, or time weighting is invalid.

Source code in wandas/processing/temporal.py

def __init__(
    self,
    sampling_rate: float,
    ref: list[float] | float | NDArrayReal = 1.0,
    freq_weighting: str | None = "Z",
    time_weighting: str = "Fast",
    dB: bool = False,
    *,
    _calibration_scale: list[float] | float | NDArrayReal = 1.0,
) -> None:
    """Initialize a frequency- and time-weighted level operation.

    Args:
        sampling_rate: Input sampling rate in Hz.
        ref: Positive finite amplitude reference, either one scalar or one
            value per channel. For Pa input, ``2e-5`` produces dB SPL.
        freq_weighting: Implemented frequency curve: ``"A"``, ``"C"``, or
            flat ``"Z"``. ``None`` is treated as ``"Z"``.
        time_weighting: Exponential time constant: ``"Fast"`` (125 ms) or
            ``"Slow"`` (1 s). The short forms ``"F"`` and ``"S"`` are
            accepted too.
        dB: If True, return ``10 * log10`` of smoothed power relative to
            ``ref**2``; otherwise return linear weighted RMS in the input
            unit.
        _calibration_scale: Positive internal amplitude scale supplied by
            calibrated Frame execution. It is not a public recipe
            parameter.

    Raises:
        ValueError: If the sampling rate, reference, calibration scale,
            frequency curve, or time weighting is invalid.
    """
    validate_sampling_rate(sampling_rate)
    ref_array = np.atleast_1d(np.array(ref, dtype=float, copy=True))
    if ref_array.size == 0 or np.any(~np.isfinite(ref_array)) or np.any(ref_array <= 0):
        raise ValueError(
            "Invalid sound level reference\n"
            f"  Got: {ref_array.tolist()}\n"
            "  Expected: Positive finite reference values\n"
            "Reference-relative dB output requires a positive finite amplitude reference."
        )
    calibration_scale = _validated_calibration_scale(
        _calibration_scale,
        operation_label="sound level",
    )
    normalized_freq_weighting = self._normalize_freq_weighting(freq_weighting)
    normalized_time_weighting = self._normalize_time_weighting(time_weighting)
    super().__init__(
        sampling_rate,
        ref=ref_array,
        freq_weighting=normalized_freq_weighting,
        time_weighting=normalized_time_weighting,
        dB=dB,
    )
    object.__setattr__(self, "_calibration_scale", calibration_scale)

get_display_name()

Get display name for the operation for use in channel labels.

Source code in wandas/processing/temporal.py

def get_display_name(self) -> str:
    """Get display name for the operation for use in channel labels."""
    freq_weighting = self.freq_weighting
    time_weighting = self.time_weighting
    if self.dB:
        return f"L{freq_weighting}{time_weighting[0]}"
    return f"{freq_weighting}{time_weighting[0]}RMS"

calculate_output_shape(input_shape)

Validate channel-wise configuration and preserve input shape.

Parameters:

Name	Type	Description	Default
input_shape	tuple[int, ...]	Input shape, normally (channels, samples).	required

Returns:

Type	Description
tuple[int, ...]	The unchanged input shape. sound_level is sample-wise and keeps the input sampling rate.

Raises:

Type	Description
ValueError	If a per-channel reference or calibration scale cannot be broadcast to the input channel count.

Source code in wandas/processing/temporal.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """Validate channel-wise configuration and preserve input shape.

    Args:
        input_shape: Input shape, normally ``(channels, samples)``.

    Returns:
        The unchanged input shape. ``sound_level`` is sample-wise and keeps
            the input sampling rate.

    Raises:
        ValueError: If a per-channel reference or calibration scale cannot
            be broadcast to the input channel count.
    """
    if self.dB:
        self._reference_values(input_shape[0])
        self._calibration_scale_values(input_shape[0])
    return input_shape

calculate_output_dtype(input_dtype, *input_dtypes)

Return sound level output dtype metadata.

Source code in wandas/processing/temporal.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    """Return sound level output dtype metadata."""
    return self._output_dtype(input_dtype)

Trim

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Deprecated array-level trimming operation.

Use :meth:wandas.frames.channel.ChannelFrame.trim for structural time-range selection.

Source code in wandas/processing/temporal.py

class Trim(AudioOperation[NDArrayReal, NDArrayReal]):
    """Deprecated array-level trimming operation.

    Use :meth:`wandas.frames.channel.ChannelFrame.trim` for structural
    time-range selection.
    """

    name = "trim"
    _display = "trim"

    def __init__(
        self,
        sampling_rate: float,
        start: float,
        end: float,
    ) -> None:
        warnings.warn(
            "wandas.processing.Trim is deprecated; use Frame.trim() for structural time-range selection",
            DeprecationWarning,
            stacklevel=2,
        )
        super().__init__(sampling_rate, start=start, end=end)

    @property
    def start(self) -> float:
        """Start time captured at operation construction time."""
        return self._config_value("start")

    @property
    def end(self) -> float:
        """End time captured at operation construction time."""
        return self._config_value("end")

    @property
    def start_sample(self) -> int:
        """Start sample index derived from the captured start time."""
        return int(self.start * self.sampling_rate)

    @property
    def end_sample(self) -> int:
        """End sample index derived from the captured end time."""
        return int(self.end * self.sampling_rate)

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        """Return the legacy array-slice output shape."""
        start_sample, end_sample, _ = slice(self.start_sample, self.end_sample).indices(input_shape[-1])
        n_samples = max(0, end_sample - start_sample)
        return (*input_shape[:-1], n_samples)

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Apply the legacy array-level slice."""
        return x[..., self.start_sample : self.end_sample]

Attributes

name = 'trim' class-attribute instance-attribute

start property

Start time captured at operation construction time.

end property

End time captured at operation construction time.

start_sample property

Start sample index derived from the captured start time.

end_sample property

End sample index derived from the captured end time.

Functions

__init__(sampling_rate, start, end)

Source code in wandas/processing/temporal.py

def __init__(
    self,
    sampling_rate: float,
    start: float,
    end: float,
) -> None:
    warnings.warn(
        "wandas.processing.Trim is deprecated; use Frame.trim() for structural time-range selection",
        DeprecationWarning,
        stacklevel=2,
    )
    super().__init__(sampling_rate, start=start, end=end)

calculate_output_shape(input_shape)

Return the legacy array-slice output shape.

Source code in wandas/processing/temporal.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """Return the legacy array-slice output shape."""
    start_sample, end_sample, _ = slice(self.start_sample, self.end_sample).indices(input_shape[-1])
    n_samples = max(0, end_sample - start_sample)
    return (*input_shape[:-1], n_samples)

Functions

create_operation(name, sampling_rate, **params)

Create operation instance from name and parameters

Source code in wandas/processing/base.py

def create_operation(name: str, sampling_rate: float, **params: Any) -> AudioOperation[Any, Any]:
    """Create operation instance from name and parameters"""
    operation_class = get_operation(name)
    return operation_class(sampling_rate, **params)

get_operation(name)

Resolve and return a registered AudioOperation class by name.

Source code in wandas/processing/base.py

def get_operation(name: str) -> type[AudioOperation[Any, Any]]:
    """Resolve and return a registered ``AudioOperation`` class by name."""
    operation_name = _validate_nonblank_string(name, field_name="Operation name")

    with _OPERATION_LOCK:
        cached = _OPERATION_CACHE.get(operation_name)
        if cached is not None:
            return cached
        provider = _OPERATION_PROVIDERS.get(operation_name)

    if provider is None:
        raise ValueError(
            f"Unknown operation type: {operation_name!r}. Register it with "
            "register_operation() or register_lazy_operation()."
        )

    if isinstance(provider, _EagerOperationProvider):
        operation_class = provider.operation_class
    else:
        module = importlib.import_module(provider.module_name)
        try:
            candidate = getattr(module, provider.attribute_name)
        except AttributeError as exc:
            raise AttributeError(
                f"Could not resolve Operation {operation_name!r}: module "
                f"{provider.module_name!r} has no attribute "
                f"{provider.attribute_name!r}; expected that attribute to expose "
                f"a concrete {AudioOperation.__name__} subclass."
            ) from exc
        try:
            operation_class = _validate_operation_class(operation_name, candidate)
        except (TypeError, ValueError) as exc:
            raise type(exc)(
                f"Invalid lazy Operation provider for {operation_name!r}: "
                f"module {provider.module_name!r}, attribute "
                f"{provider.attribute_name!r} provided {candidate!r}; {exc}"
            ) from exc

    with _OPERATION_LOCK:
        cached = _OPERATION_CACHE.get(operation_name)
        if cached is not None:
            if cached is operation_class:
                return cached
        else:
            _OPERATION_CACHE[operation_name] = operation_class
            return operation_class
    raise ValueError(
        f"Operation name {operation_name!r} was resolved to conflicting "
        f"class objects: cache contains {cached!r}, while provider "
        f"resolved {operation_class!r}."
    )

register_lazy_operation(name, module_name, *, attribute_name)

Register a lazy operation provider using an explicit module attribute.

Registration stores module_name and the required keyword-only attribute_name without importing the module. The referenced class must expose a non-blank name equal to name when it is resolved. Only the exact same provider declaration may be registered more than once.

Source code in wandas/processing/base.py

def register_lazy_operation(
    name: str,
    module_name: str,
    *,
    attribute_name: str,
) -> None:
    """Register a lazy operation provider using an explicit module attribute.

    Registration stores ``module_name`` and the required keyword-only
    ``attribute_name`` without importing the module. The referenced class must
    expose a non-blank ``name`` equal to *name* when it is resolved. Only the
    exact same provider declaration may be registered more than once.
    """
    operation_name = _validate_nonblank_string(name, field_name="Operation name")
    provider_module = _validate_nonblank_string(module_name, field_name="module name")
    provider_attribute = _validate_nonblank_string(attribute_name, field_name="attribute name")
    provider = _LazyOperationProvider(provider_module, provider_attribute)

    with _OPERATION_LOCK:
        existing = _OPERATION_PROVIDERS.get(operation_name)
        if existing is None:
            _OPERATION_PROVIDERS[operation_name] = provider
            return
        if (
            isinstance(existing, _LazyOperationProvider)
            and existing.module_name == provider.module_name
            and existing.attribute_name == provider.attribute_name
        ):
            return
    _raise_provider_conflict(
        operation_name,
        existing,
        f"lazy module {provider.module_name!r} attribute {provider.attribute_name!r}",
    )

register_operation(operation_class)

Register a concrete eager AudioOperation class.

The class's name is its single provider key. Re-registering the exact same class object is idempotent; another class or a lazy provider owning the same name is rejected.

Source code in wandas/processing/base.py

def register_operation(
    operation_class: type[AudioOperation[Any, Any]],
) -> None:
    """Register a concrete eager ``AudioOperation`` class.

    The class's ``name`` is its single provider key. Re-registering the exact
    same class object is idempotent; another class or a lazy provider owning the
    same name is rejected.
    """
    declared_name = getattr(operation_class, "name", None) if inspect.isclass(operation_class) else None
    validated_class = _validate_operation_class(declared_name, operation_class)
    operation_name = validated_class.name
    provider = _EagerOperationProvider(validated_class)

    with _OPERATION_LOCK:
        existing = _OPERATION_PROVIDERS.get(operation_name)
        if existing is None:
            _OPERATION_PROVIDERS[operation_name] = provider
            _OPERATION_CACHE[operation_name] = validated_class
            return
        if isinstance(existing, _EagerOperationProvider) and existing.operation_class is validated_class:
            return
    _raise_provider_conflict(
        operation_name,
        existing,
        f"eager class {validated_class!r}",
    )

apply_channel_factors(data, factors)

Multiply channel-first data by one factor per channel without computing it.

Source code in wandas/processing/calibration.py

def apply_channel_factors(data: DaArray, factors: Sequence[float]) -> DaArray:
    """Multiply channel-first data by one factor per channel without computing it."""
    if not isinstance(data, DaArray):
        raise TypeError("Calibration data must be a Dask array")
    if data.ndim < 1:
        raise ValueError("Calibration data must have a channel axis")
    values = tuple(factors)
    if len(values) != int(data.shape[0]):
        raise ValueError(
            "Calibration factor length mismatch\n"
            f"  Got: {len(values)} factors\n"
            f"  Expected: {data.shape[0]} factors for the channel axis\n"
            "Align calibration metadata with the current channel order."
        )
    if any(
        isinstance(value, bool)
        or not isinstance(value, numbers.Real)
        or not math.isfinite(float(value))
        or float(value) <= 0
        for value in values
    ):
        raise ValueError(
            "Invalid calibration factors\n"
            f"  Got: {values!r}\n"
            "  Expected: one positive finite number per channel\n"
            "Validate each ChannelCalibration before applying it."
        )
    broadcast_shape = (len(values),) + (1,) * (data.ndim - 1)
    return data * np.asarray(values, dtype=float).reshape(broadcast_shape)

__getattr__(name)

Source code in wandas/processing/__init__.py

def __getattr__(name: str) -> Any:
    lazy_operation = _LAZY_OPERATION_CLASSES.get(name)
    if lazy_operation is not None:
        operation_name, _, _ = lazy_operation
        operation_class = get_operation(operation_name)
        globals()[name] = operation_class
        return operation_class
    raise AttributeError(f"module 'wandas.processing' has no attribute {name!r}")

wandas.processing.temporal

Attributes

logger = logging.getLogger(__name__) module-attribute

MIN_SOUND_LEVEL_POWER_RATIO = 1e-20 module-attribute

MAX_RESAMPLING_FACTOR = 1000000 module-attribute

Classes

ReSampling

Bases: ChannelIndependentAudioOperation[NDArrayReal, NDArrayReal]

Resampling operation

Source code in wandas/processing/temporal.py

class ReSampling(ChannelIndependentAudioOperation[NDArrayReal, NDArrayReal]):
    """Resampling operation"""

    name = "resampling"
    _display = "rs"

    def __init__(self, sampling_rate: float, target_sr: float):
        """
        Initialize a resampling operation.

        Args:
            sampling_rate (float): Source sampling rate in Hz.
            target_sr (float): Target sampling rate in Hz.

        Raises:
            ValueError: If ``sampling_rate`` or ``target_sr`` is not positive.
        """
        validate_sampling_rate(sampling_rate, "source sampling rate")
        validate_sampling_rate(target_sr, "target sampling rate")
        super().__init__(sampling_rate, target_sr=target_sr)

    @property
    def target_sr(self) -> float:
        """Target sampling rate captured at operation construction time."""
        return self._config_value("target_sr")

    def get_metadata_updates(self) -> dict[str, Any]:
        """
        Update sampling rate to target sampling rate.

        Returns:
            dict: Metadata updates with the new sampling rate.

        Notes:
            Resampling always produces output at ``target_sr``, regardless of the
            input sampling rate.
        """
        return {"sampling_rate": self.target_sr}

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        """
        Calculate the output data shape after the operation.

        Args:
            input_shape (tuple[int, ...]): Input data shape.

        Returns:
            tuple[int, ...]: Output data shape.
        """
        # Calculate length after resampling using exact decimal sampling-rate ratio.
        ratio = _resampling_fraction(self.sampling_rate, self.target_sr)
        n_samples = _ceil_resampled_length(input_shape[-1], ratio)
        return (*input_shape[:-1], n_samples)

    @staticmethod
    def _output_dtype(input_dtype: np.dtype[Any]) -> np.dtype[Any]:
        dtype = np.dtype(input_dtype)
        if dtype.kind == "f":
            return dtype
        return np.dtype(np.float64)

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Create processor function for resampling operation"""
        logger.debug(f"Applying resampling to array with shape: {x.shape}")
        up, down = _resampling_ratio(self.sampling_rate, self.target_sr)
        target_len = self.calculate_output_shape(x.shape)[-1]
        poly_len = _ceil_resampled_length(x.shape[-1], Fraction(up, down))
        if poly_len == target_len:
            result: NDArrayReal = scipy.signal.resample_poly(x, up, down, axis=-1)
        else:
            result = scipy.signal.resample(x, target_len, axis=-1)
        logger.debug(f"Resampling applied, returning result with shape: {result.shape}")
        return result

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        """Return resampling output dtype metadata."""
        return self._output_dtype(input_dtype)

Attributes

name = 'resampling' class-attribute instance-attribute

target_sr property

Target sampling rate captured at operation construction time.

Functions

__init__(sampling_rate, target_sr)

Initialize a resampling operation.

Parameters:

Name	Type	Description	Default
sampling_rate	float	Source sampling rate in Hz.	required
target_sr	float	Target sampling rate in Hz.	required

Raises:

Type	Description
ValueError	If sampling_rate or target_sr is not positive.

Source code in wandas/processing/temporal.py

def __init__(self, sampling_rate: float, target_sr: float):
    """
    Initialize a resampling operation.

    Args:
        sampling_rate (float): Source sampling rate in Hz.
        target_sr (float): Target sampling rate in Hz.

    Raises:
        ValueError: If ``sampling_rate`` or ``target_sr`` is not positive.
    """
    validate_sampling_rate(sampling_rate, "source sampling rate")
    validate_sampling_rate(target_sr, "target sampling rate")
    super().__init__(sampling_rate, target_sr=target_sr)

get_metadata_updates()

Update sampling rate to target sampling rate.

Returns:

Name	Type	Description
dict	dict[str, Any]	Metadata updates with the new sampling rate.

Notes

Resampling always produces output at target_sr, regardless of the input sampling rate.

Source code in wandas/processing/temporal.py

def get_metadata_updates(self) -> dict[str, Any]:
    """
    Update sampling rate to target sampling rate.

    Returns:
        dict: Metadata updates with the new sampling rate.

    Notes:
        Resampling always produces output at ``target_sr``, regardless of the
        input sampling rate.
    """
    return {"sampling_rate": self.target_sr}

calculate_output_shape(input_shape)

Calculate the output data shape after the operation.

Parameters:

Name	Type	Description	Default
input_shape	tuple[int, ...]	Input data shape.	required

Returns:

Type	Description
tuple[int, ...]	tuple[int, ...]: Output data shape.

Source code in wandas/processing/temporal.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """
    Calculate the output data shape after the operation.

    Args:
        input_shape (tuple[int, ...]): Input data shape.

    Returns:
        tuple[int, ...]: Output data shape.
    """
    # Calculate length after resampling using exact decimal sampling-rate ratio.
    ratio = _resampling_fraction(self.sampling_rate, self.target_sr)
    n_samples = _ceil_resampled_length(input_shape[-1], ratio)
    return (*input_shape[:-1], n_samples)

calculate_output_dtype(input_dtype, *input_dtypes)

Return resampling output dtype metadata.

Source code in wandas/processing/temporal.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    """Return resampling output dtype metadata."""
    return self._output_dtype(input_dtype)

Trim

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Deprecated array-level trimming operation.

Use :meth:wandas.frames.channel.ChannelFrame.trim for structural time-range selection.

Source code in wandas/processing/temporal.py

class Trim(AudioOperation[NDArrayReal, NDArrayReal]):
    """Deprecated array-level trimming operation.

    Use :meth:`wandas.frames.channel.ChannelFrame.trim` for structural
    time-range selection.
    """

    name = "trim"
    _display = "trim"

    def __init__(
        self,
        sampling_rate: float,
        start: float,
        end: float,
    ) -> None:
        warnings.warn(
            "wandas.processing.Trim is deprecated; use Frame.trim() for structural time-range selection",
            DeprecationWarning,
            stacklevel=2,
        )
        super().__init__(sampling_rate, start=start, end=end)

    @property
    def start(self) -> float:
        """Start time captured at operation construction time."""
        return self._config_value("start")

    @property
    def end(self) -> float:
        """End time captured at operation construction time."""
        return self._config_value("end")

    @property
    def start_sample(self) -> int:
        """Start sample index derived from the captured start time."""
        return int(self.start * self.sampling_rate)

    @property
    def end_sample(self) -> int:
        """End sample index derived from the captured end time."""
        return int(self.end * self.sampling_rate)

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        """Return the legacy array-slice output shape."""
        start_sample, end_sample, _ = slice(self.start_sample, self.end_sample).indices(input_shape[-1])
        n_samples = max(0, end_sample - start_sample)
        return (*input_shape[:-1], n_samples)

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Apply the legacy array-level slice."""
        return x[..., self.start_sample : self.end_sample]

Attributes

name = 'trim' class-attribute instance-attribute

start property

Start time captured at operation construction time.

end property

End time captured at operation construction time.

start_sample property

Start sample index derived from the captured start time.

end_sample property

End sample index derived from the captured end time.

Functions

__init__(sampling_rate, start, end)

Source code in wandas/processing/temporal.py

def __init__(
    self,
    sampling_rate: float,
    start: float,
    end: float,
) -> None:
    warnings.warn(
        "wandas.processing.Trim is deprecated; use Frame.trim() for structural time-range selection",
        DeprecationWarning,
        stacklevel=2,
    )
    super().__init__(sampling_rate, start=start, end=end)

calculate_output_shape(input_shape)

Return the legacy array-slice output shape.

Source code in wandas/processing/temporal.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """Return the legacy array-slice output shape."""
    start_sample, end_sample, _ = slice(self.start_sample, self.end_sample).indices(input_shape[-1])
    n_samples = max(0, end_sample - start_sample)
    return (*input_shape[:-1], n_samples)

FixLength

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Operation to adjust signal length to a specified length.

Source code in wandas/processing/temporal.py

class FixLength(AudioOperation[NDArrayReal, NDArrayReal]):
    """Operation to adjust signal length to a specified length."""

    name = "fix_length"
    _display = "fix"

    def __init__(
        self,
        sampling_rate: float,
        length: int | None = None,
        duration: float | None = None,
    ):
        """Initialize an operation that pads or truncates a signal.

        Args:
            sampling_rate: Input sampling rate in Hz.
            length: Target number of samples. Provide either ``length`` or
                ``duration``.
            duration: Target duration in seconds. It is converted to samples
                using ``sampling_rate``.

        Raises:
            ValueError: If neither ``length`` nor ``duration`` is provided.
        """
        if length is None:
            if duration is None:
                raise ValueError("Either length or duration must be provided.")
            length = int(duration * sampling_rate)
        super().__init__(sampling_rate, target_length=length)

    @property
    def target_length(self) -> int:
        """Target length captured at operation construction time."""
        return self._config_value("target_length")

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        """Return the input shape with its sample axis set to target length.

        Args:
            input_shape: Input array shape with samples on the last axis.

        Returns:
            Shape with the same leading dimensions and ``target_length`` as
                the final dimension.
        """
        return (*input_shape[:-1], self.target_length)

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Create processor function for padding operation"""
        logger.debug(f"Applying padding to array with shape: {x.shape}")
        # Apply padding
        pad_width = self.target_length - x.shape[-1]
        if pad_width > 0:
            result = np.pad(x, ((0, 0), (0, pad_width)), mode="constant")
        else:
            result = x[..., : self.target_length]
        logger.debug(f"Padding applied, returning result with shape: {result.shape}")
        return result

Attributes

name = 'fix_length' class-attribute instance-attribute

target_length property

Target length captured at operation construction time.

Functions

__init__(sampling_rate, length=None, duration=None)

Initialize an operation that pads or truncates a signal.

Parameters:

Name	Type	Description	Default
sampling_rate	float	Input sampling rate in Hz.	required
length	int | None	Target number of samples. Provide either length or duration.	None
duration	float | None	Target duration in seconds. It is converted to samples using sampling_rate.	None

Raises:

Type	Description
ValueError	If neither length nor duration is provided.

Source code in wandas/processing/temporal.py

def __init__(
    self,
    sampling_rate: float,
    length: int | None = None,
    duration: float | None = None,
):
    """Initialize an operation that pads or truncates a signal.

    Args:
        sampling_rate: Input sampling rate in Hz.
        length: Target number of samples. Provide either ``length`` or
            ``duration``.
        duration: Target duration in seconds. It is converted to samples
            using ``sampling_rate``.

    Raises:
        ValueError: If neither ``length`` nor ``duration`` is provided.
    """
    if length is None:
        if duration is None:
            raise ValueError("Either length or duration must be provided.")
        length = int(duration * sampling_rate)
    super().__init__(sampling_rate, target_length=length)

calculate_output_shape(input_shape)

Return the input shape with its sample axis set to target length.

Parameters:

Name	Type	Description	Default
input_shape	tuple[int, ...]	Input array shape with samples on the last axis.	required

Returns:

Type	Description
tuple[int, ...]	Shape with the same leading dimensions and target_length as the final dimension.

Source code in wandas/processing/temporal.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """Return the input shape with its sample axis set to target length.

    Args:
        input_shape: Input array shape with samples on the last axis.

    Returns:
        Shape with the same leading dimensions and ``target_length`` as
            the final dimension.
    """
    return (*input_shape[:-1], self.target_length)

RmsTrend

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Windowed linear RMS or reference-relative RMS amplitude level.

The operation accepts arrays shaped (channels, samples) and returns (channels, frames) using centered, zero-padded windows. dB=False returns RMS in the input unit. dB=True returns 20 * log10(max(RMS / ref, 1e-12)), bounded below by -240 dB, with one scalar reference shared across channels or one reference per channel. Applying Aw changes the frequency weighting before RMS; it does not establish instrument conformance. The operation is lazy when used through a Frame and preserves the input dtype contract by returning floating data.

Source code in wandas/processing/temporal.py

class RmsTrend(AudioOperation[NDArrayReal, NDArrayReal]):
    """Windowed linear RMS or reference-relative RMS amplitude level.

    The operation accepts arrays shaped ``(channels, samples)`` and returns
    ``(channels, frames)`` using centered, zero-padded windows. ``dB=False``
    returns RMS in the input unit. ``dB=True`` returns
    ``20 * log10(max(RMS / ref, 1e-12))``, bounded below by -240 dB, with one
    scalar reference shared across channels or one reference per channel.
    Applying ``Aw`` changes the frequency weighting before RMS; it does not
    establish instrument conformance. The operation is lazy when used through
    a Frame and preserves the input dtype contract by returning floating data.
    """

    name = "rms_trend"
    _display = "RMS"
    _validate_reference_count = True
    _calibration_scale: tuple[float, ...]

    def __init__(
        self,
        sampling_rate: float,
        frame_length: int = 2048,
        hop_length: int = 512,
        ref: list[float] | float = 1.0,
        dB: bool = False,
        Aw: bool = False,
        *,
        _calibration_scale: list[float] | float | NDArrayReal = 1.0,
    ) -> None:
        """Initialize a centered windowed RMS operation.

        Args:
            sampling_rate: Input sampling rate in Hz.
            frame_length: Window length in samples. Defaults to 2048.
            hop_length: Distance between output frames in samples. Defaults to
                512. The output sampling rate is ``sampling_rate / hop_length``.
            ref: Positive finite amplitude reference, either one scalar or one
                value per channel. For Pa input, ``2e-5`` produces dB SPL.
            dB: If True, return reference-relative amplitude level instead of
                linear RMS amplitude.
            Aw: If True, apply the implemented digital A-weighting filter before
                RMS calculation.
            _calibration_scale: Positive internal amplitude scale supplied by
                calibrated Frame execution. It is not a public recipe
                parameter.

        Raises:
            ValueError: If the sampling or window parameters are invalid, or if
                a reference or calibration scale is not finite and positive.
        """
        ref_array = np.array(ref if isinstance(ref, list) else [ref], dtype=float)
        if ref_array.size == 0 or np.any(~np.isfinite(ref_array)) or np.any(ref_array <= 0):
            raise ValueError(
                "Invalid RMS level reference\n"
                f"  Got: {ref_array.tolist()}\n"
                "  Expected: Positive finite reference values\n"
                "Reference-relative dB output requires a positive finite amplitude reference."
            )
        calibration_scale = _validated_calibration_scale(
            _calibration_scale,
            operation_label="RMS level",
        )
        super().__init__(
            sampling_rate,
            frame_length=frame_length,
            hop_length=hop_length,
            dB=dB,
            Aw=Aw,
            ref=ref_array,
        )
        object.__setattr__(self, "_calibration_scale", calibration_scale)

    @property
    def frame_length(self) -> int:
        """Frame length captured at operation construction time."""
        return self._config_value("frame_length")

    @property
    def hop_length(self) -> int:
        """Hop length captured at operation construction time."""
        return self._config_value("hop_length")

    @property
    def dB(self) -> bool:  # noqa: N802
        """Whether output is converted to decibels."""
        return self._config_value("dB")

    @property
    def Aw(self) -> bool:  # noqa: N802
        """Whether A-weighting is applied before RMS calculation."""
        return self._config_value("Aw")

    @property
    def ref(self) -> NDArrayReal:
        """Reference values captured at operation construction time."""
        return self._config_value("ref")

    def _reference_values(self, n_channels: int) -> NDArrayReal:
        """Return one validated reference value for each channel."""
        ref_config = self._config["ref"]
        if ref_config.size == 1:
            ref = np.repeat(ref_config, n_channels)
        elif ref_config.size == n_channels:
            ref = ref_config
        else:
            raise ValueError(
                "Reference count mismatch\n"
                f"  Got: {ref_config.size} reference values for {n_channels} channels\n"
                "  Expected: One shared reference or one reference per channel\n"
                "Provide ref as a scalar or a list matching the number of channels."
            )
        return np.asarray(ref, dtype=np.float64)

    def _calibration_scale_values(self, n_channels: int) -> NDArrayReal:
        """Return one validated internal amplitude scale for each channel."""
        return _calibration_scale_values(self._calibration_scale, n_channels)

    def get_metadata_updates(self) -> dict[str, Any]:
        """Return metadata updates for the frame-rate change.

        Returns:
            A mapping containing ``sampling_rate`` set to
                ``sampling_rate / hop_length``. The returned value describes the
                window centers, not the original sample rate.
        """
        new_sr = self.sampling_rate / self.hop_length
        return {"sampling_rate": new_sr}

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        """Calculate the centered, zero-padded output shape.

        Args:
            input_shape: Input shape whose last dimension contains samples;
                the usual Frame shape is ``(channels, samples)``.

        Returns:
            The input leading dimensions followed by the number of centered
                windows, so a two-dimensional input returns ``(channels, frames)``.

        Raises:
            ValueError: If dB output uses a reference or calibration scale that
                cannot be broadcast to the input channel count.
        """
        if self.dB and self._validate_reference_count:
            self._reference_values(input_shape[0])
            self._calibration_scale_values(input_shape[0])
        n_frames = _centered_frame_count(
            input_shape[-1],
            self.frame_length,
            self.hop_length,
        )
        return (*input_shape[:-1], n_frames)

    _keeps_input_precision = True

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Create processor function for RMS calculation"""
        logger.debug(f"Applying RMS to array with shape: {x.shape}")

        weighted_log_amplitude: NDArrayReal | None = None
        if self.Aw:
            # Apply A-weighting
            weighting_input = x
            if self.dB:
                weighting_input = np.asarray(x, dtype=np.float64)
                first_unsafe_samples = _level_filter_first_unsafe_samples(weighting_input)
                if np.any(first_unsafe_samples >= 0):
                    weighted_log_amplitude = _frequency_weight_log_amplitude(
                        weighting_input,
                        self.sampling_rate,
                        curve="A",
                        first_unsafe_samples=first_unsafe_samples,
                    )
            if weighted_log_amplitude is None:
                _x = A_weight(weighting_input, self.sampling_rate)
                if isinstance(_x, np.ndarray):
                    x = _x
                elif isinstance(_x, tuple):
                    x = _x[0]
                else:
                    raise ValueError("A_weighting returned an unexpected type.")

        if self.dB:
            references = self._reference_values(x.shape[0])
            calibration_scale = self._calibration_scale_values(x.shape[0])
            if weighted_log_amplitude is None:
                log_rms = _frame_log_rms(
                    x,
                    frame_length=self.frame_length,
                    hop_length=self.hop_length,
                )
            else:
                log_rms = _frame_log_rms_from_log_amplitude(
                    weighted_log_amplitude,
                    frame_length=self.frame_length,
                    hop_length=self.hop_length,
                )
            with np.errstate(divide="ignore", invalid="ignore"):
                np.add(
                    log_rms,
                    np.log(calibration_scale)[..., np.newaxis],
                    out=log_rms,
                )
                np.subtract(
                    log_rms,
                    np.log(references)[..., np.newaxis],
                    out=log_rms,
                )
            result = _bounded_db_from_log_ratio(
                log_rms,
                scale=20.0,
                ratio_floor=DB_FLOOR,
            )
        else:
            # Preserve the released linear RMS path exactly. Numerical scaling
            # belongs only to the versioned dB contract.
            result = _frame_rms(
                x,
                frame_length=self.frame_length,
                hop_length=self.hop_length,
            )
        logger.debug(f"RMS applied, returning result with shape: {result.shape}")
        return result

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        """Return RMS trend output dtype metadata."""
        return precision_real_dtype(input_dtype)

Attributes

name = 'rms_trend' class-attribute instance-attribute

frame_length property

Frame length captured at operation construction time.

hop_length property

Hop length captured at operation construction time.

dB property

Whether output is converted to decibels.

Aw property

Whether A-weighting is applied before RMS calculation.

ref property

Reference values captured at operation construction time.

Functions

__init__(sampling_rate, frame_length=2048, hop_length=512, ref=1.0, dB=False, Aw=False, *, _calibration_scale=1.0)

Initialize a centered windowed RMS operation.

Parameters:

Name	Type	Description	Default
sampling_rate	float	Input sampling rate in Hz.	required
frame_length	int	Window length in samples. Defaults to 2048.	2048
hop_length	int	Distance between output frames in samples. Defaults to 512. The output sampling rate is sampling_rate / hop_length.	512
ref	list[float] | float	Positive finite amplitude reference, either one scalar or one value per channel. For Pa input, 2e-5 produces dB SPL.	1.0
dB	bool	If True, return reference-relative amplitude level instead of linear RMS amplitude.	False
Aw	bool	If True, apply the implemented digital A-weighting filter before RMS calculation.	False
_calibration_scale	list[float] | float | NDArrayReal	Positive internal amplitude scale supplied by calibrated Frame execution. It is not a public recipe parameter.	1.0

Raises:

Type	Description
ValueError	If the sampling or window parameters are invalid, or if a reference or calibration scale is not finite and positive.

Source code in wandas/processing/temporal.py

def __init__(
    self,
    sampling_rate: float,
    frame_length: int = 2048,
    hop_length: int = 512,
    ref: list[float] | float = 1.0,
    dB: bool = False,
    Aw: bool = False,
    *,
    _calibration_scale: list[float] | float | NDArrayReal = 1.0,
) -> None:
    """Initialize a centered windowed RMS operation.

    Args:
        sampling_rate: Input sampling rate in Hz.
        frame_length: Window length in samples. Defaults to 2048.
        hop_length: Distance between output frames in samples. Defaults to
            512. The output sampling rate is ``sampling_rate / hop_length``.
        ref: Positive finite amplitude reference, either one scalar or one
            value per channel. For Pa input, ``2e-5`` produces dB SPL.
        dB: If True, return reference-relative amplitude level instead of
            linear RMS amplitude.
        Aw: If True, apply the implemented digital A-weighting filter before
            RMS calculation.
        _calibration_scale: Positive internal amplitude scale supplied by
            calibrated Frame execution. It is not a public recipe
            parameter.

    Raises:
        ValueError: If the sampling or window parameters are invalid, or if
            a reference or calibration scale is not finite and positive.
    """
    ref_array = np.array(ref if isinstance(ref, list) else [ref], dtype=float)
    if ref_array.size == 0 or np.any(~np.isfinite(ref_array)) or np.any(ref_array <= 0):
        raise ValueError(
            "Invalid RMS level reference\n"
            f"  Got: {ref_array.tolist()}\n"
            "  Expected: Positive finite reference values\n"
            "Reference-relative dB output requires a positive finite amplitude reference."
        )
    calibration_scale = _validated_calibration_scale(
        _calibration_scale,
        operation_label="RMS level",
    )
    super().__init__(
        sampling_rate,
        frame_length=frame_length,
        hop_length=hop_length,
        dB=dB,
        Aw=Aw,
        ref=ref_array,
    )
    object.__setattr__(self, "_calibration_scale", calibration_scale)

get_metadata_updates()

Return metadata updates for the frame-rate change.

Returns:

Type	Description
dict[str, Any]	A mapping containing sampling_rate set to sampling_rate / hop_length. The returned value describes the window centers, not the original sample rate.

Source code in wandas/processing/temporal.py

def get_metadata_updates(self) -> dict[str, Any]:
    """Return metadata updates for the frame-rate change.

    Returns:
        A mapping containing ``sampling_rate`` set to
            ``sampling_rate / hop_length``. The returned value describes the
            window centers, not the original sample rate.
    """
    new_sr = self.sampling_rate / self.hop_length
    return {"sampling_rate": new_sr}

calculate_output_shape(input_shape)

Calculate the centered, zero-padded output shape.

Parameters:

Name	Type	Description	Default
input_shape	tuple[int, ...]	Input shape whose last dimension contains samples; the usual Frame shape is (channels, samples).	required

Returns:

Type	Description
tuple[int, ...]	The input leading dimensions followed by the number of centered windows, so a two-dimensional input returns (channels, frames).

Raises:

Type	Description
ValueError	If dB output uses a reference or calibration scale that cannot be broadcast to the input channel count.

Source code in wandas/processing/temporal.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """Calculate the centered, zero-padded output shape.

    Args:
        input_shape: Input shape whose last dimension contains samples;
            the usual Frame shape is ``(channels, samples)``.

    Returns:
        The input leading dimensions followed by the number of centered
            windows, so a two-dimensional input returns ``(channels, frames)``.

    Raises:
        ValueError: If dB output uses a reference or calibration scale that
            cannot be broadcast to the input channel count.
    """
    if self.dB and self._validate_reference_count:
        self._reference_values(input_shape[0])
        self._calibration_scale_values(input_shape[0])
    n_frames = _centered_frame_count(
        input_shape[-1],
        self.frame_length,
        self.hop_length,
    )
    return (*input_shape[:-1], n_frames)

calculate_output_dtype(input_dtype, *input_dtypes)

Return RMS trend output dtype metadata.

Source code in wandas/processing/temporal.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    """Return RMS trend output dtype metadata."""
    return precision_real_dtype(input_dtype)

SoundLevel

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Frequency- and exponentially time-weighted RMS or level.

The operation applies A, C, or flat Z frequency weighting, smooths squared samples with a 125 ms (Fast) or 1 s (Slow) first-order exponential filter, and returns either the square root (linear RMS) or 10 * log10(max(smoothed_power / ref**2, 1e-20)), bounded below by -200 dB. The result is dB SPL only for pressure in Pa with ref=2e-5. The implementation is not a claim of complete IEC/JIS sound-level-meter conformance. Input and output arrays have the same (channels, samples) shape; Frame execution remains lazy and preserves the input sampling rate.

Source code in wandas/processing/temporal.py

class SoundLevel(AudioOperation[NDArrayReal, NDArrayReal]):
    """Frequency- and exponentially time-weighted RMS or level.

    The operation applies A, C, or flat Z frequency weighting, smooths squared
    samples with a 125 ms (Fast) or 1 s (Slow) first-order exponential filter,
    and returns either the square root (linear RMS) or
    ``10 * log10(max(smoothed_power / ref**2, 1e-20))``, bounded below by
    -200 dB. The result is dB SPL only for pressure in Pa with ``ref=2e-5``.
    The implementation is not a claim of complete IEC/JIS sound-level-meter
    conformance. Input and output arrays have the same ``(channels, samples)``
    shape; Frame execution remains lazy and preserves the input sampling rate.
    """

    name = "sound_level"
    _calibration_scale: tuple[float, ...]

    def __init__(
        self,
        sampling_rate: float,
        ref: list[float] | float | NDArrayReal = 1.0,
        freq_weighting: str | None = "Z",
        time_weighting: str = "Fast",
        dB: bool = False,
        *,
        _calibration_scale: list[float] | float | NDArrayReal = 1.0,
    ) -> None:
        """Initialize a frequency- and time-weighted level operation.

        Args:
            sampling_rate: Input sampling rate in Hz.
            ref: Positive finite amplitude reference, either one scalar or one
                value per channel. For Pa input, ``2e-5`` produces dB SPL.
            freq_weighting: Implemented frequency curve: ``"A"``, ``"C"``, or
                flat ``"Z"``. ``None`` is treated as ``"Z"``.
            time_weighting: Exponential time constant: ``"Fast"`` (125 ms) or
                ``"Slow"`` (1 s). The short forms ``"F"`` and ``"S"`` are
                accepted too.
            dB: If True, return ``10 * log10`` of smoothed power relative to
                ``ref**2``; otherwise return linear weighted RMS in the input
                unit.
            _calibration_scale: Positive internal amplitude scale supplied by
                calibrated Frame execution. It is not a public recipe
                parameter.

        Raises:
            ValueError: If the sampling rate, reference, calibration scale,
                frequency curve, or time weighting is invalid.
        """
        validate_sampling_rate(sampling_rate)
        ref_array = np.atleast_1d(np.array(ref, dtype=float, copy=True))
        if ref_array.size == 0 or np.any(~np.isfinite(ref_array)) or np.any(ref_array <= 0):
            raise ValueError(
                "Invalid sound level reference\n"
                f"  Got: {ref_array.tolist()}\n"
                "  Expected: Positive finite reference values\n"
                "Reference-relative dB output requires a positive finite amplitude reference."
            )
        calibration_scale = _validated_calibration_scale(
            _calibration_scale,
            operation_label="sound level",
        )
        normalized_freq_weighting = self._normalize_freq_weighting(freq_weighting)
        normalized_time_weighting = self._normalize_time_weighting(time_weighting)
        super().__init__(
            sampling_rate,
            ref=ref_array,
            freq_weighting=normalized_freq_weighting,
            time_weighting=normalized_time_weighting,
            dB=dB,
        )
        object.__setattr__(self, "_calibration_scale", calibration_scale)

    @staticmethod
    def _normalize_freq_weighting(freq_weighting: str | None) -> str:
        normalized = "Z" if freq_weighting is None else str(freq_weighting).upper()
        if normalized not in {"A", "C", "Z"}:
            raise ValueError(
                "Invalid frequency weighting\n"
                f"  Got: {freq_weighting!r}\n"
                "  Expected: 'A', 'C', or 'Z'\n"
                "Choose one of the implemented frequency-weighting curves."
            )
        return normalized

    @staticmethod
    def _normalize_time_weighting(time_weighting: str) -> str:
        normalized = str(time_weighting).strip().upper()
        if normalized in {"F", "FAST"}:
            return "Fast"
        if normalized in {"S", "SLOW"}:
            return "Slow"
        raise ValueError(
            "Invalid time weighting\n"
            f"  Got: {time_weighting!r}\n"
            "  Expected: 'Fast' or 'Slow'\n"
            "Choose one of the implemented exponential time constants."
        )

    @property
    def ref(self) -> NDArrayReal:
        """Reference values captured at operation construction time."""
        return self._config_value("ref")

    @property
    def freq_weighting(self) -> str:
        """Frequency weighting captured at operation construction time."""
        return self._config_value("freq_weighting")

    @property
    def time_weighting(self) -> str:
        """Time weighting captured at operation construction time."""
        return self._config_value("time_weighting")

    @property
    def dB(self) -> bool:  # noqa: N802
        """Whether output is converted to decibels."""
        return self._config_value("dB")

    @property
    def time_constant(self) -> float:
        """Return the RC time constant in seconds."""
        return 0.125 if self.time_weighting == "Fast" else 1.0

    @staticmethod
    def _output_dtype(
        input_dtype: np.dtype[Any],
    ) -> np.dtype[np.float32] | np.dtype[np.float64]:
        """Return the floating output dtype for the given input dtype."""
        if np.dtype(input_dtype) == np.dtype(np.float32):
            return np.dtype(np.float32)
        return np.dtype(np.float64)

    def get_display_name(self) -> str:
        """Get display name for the operation for use in channel labels."""
        freq_weighting = self.freq_weighting
        time_weighting = self.time_weighting
        if self.dB:
            return f"L{freq_weighting}{time_weighting[0]}"
        return f"{freq_weighting}{time_weighting[0]}RMS"

    def _reference_values(self, n_channels: int) -> NDArrayReal:
        """Return one validated reference value for each channel."""
        ref_config = self._config["ref"]
        if ref_config.size == 1:
            ref = np.repeat(ref_config, n_channels)
        elif ref_config.size == n_channels:
            ref = ref_config
        else:
            raise ValueError(
                "Reference count mismatch\n"
                f"  Got: {ref_config.size} reference values for {n_channels} channels\n"
                "  Expected: One shared reference or one reference per channel\n"
                "Provide ref as a scalar or a list matching the number of channels."
            )
        return np.asarray(ref, dtype=np.float64)

    def _calibration_scale_values(self, n_channels: int) -> NDArrayReal:
        """Return one validated internal amplitude scale for each channel."""
        return _calibration_scale_values(self._calibration_scale, n_channels)

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        """Validate channel-wise configuration and preserve input shape.

        Args:
            input_shape: Input shape, normally ``(channels, samples)``.

        Returns:
            The unchanged input shape. ``sound_level`` is sample-wise and keeps
                the input sampling rate.

        Raises:
            ValueError: If a per-channel reference or calibration scale cannot
                be broadcast to the input channel count.
        """
        if self.dB:
            self._reference_values(input_shape[0])
            self._calibration_scale_values(input_shape[0])
        return input_shape

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Create processor function for sound level calculation."""
        logger.debug(
            "Applying sound level to array with shape %s using %s/%s weighting",
            x.shape,
            self.freq_weighting,
            self.time_weighting,
        )
        output_dtype = self._output_dtype(x.dtype)
        weighted_input = x if x.dtype == np.float64 else np.asarray(x, dtype=np.float64)
        weighted = weighted_input
        freq_weighting = self.freq_weighting
        weighted_log_amplitude: NDArrayReal | None = None
        if freq_weighting == "Z":
            weighted = weighted_input
        else:
            if self.dB:
                first_unsafe_samples = _level_filter_first_unsafe_samples(weighted_input)
                if np.any(first_unsafe_samples >= 0):
                    weighted_log_amplitude = _frequency_weight_log_amplitude(
                        weighted_input,
                        self.sampling_rate,
                        curve=freq_weighting,
                        first_unsafe_samples=first_unsafe_samples,
                    )
                else:
                    weighted = frequency_weight(weighted_input, self.sampling_rate, curve=freq_weighting)
            else:
                weighted = frequency_weight(weighted_input, self.sampling_rate, curve=freq_weighting)
        alpha = np.asarray(np.exp(-1.0 / (self.sampling_rate * self.time_constant)), dtype=np.float64).item()
        if self.dB:
            references = self._reference_values(weighted_input.shape[0])
            calibration_scale = self._calibration_scale_values(weighted_input.shape[0])
            if weighted_log_amplitude is not None:
                log_smoothed_power = _exponential_power_from_log_amplitude(weighted_log_amplitude, alpha)
            elif _reference_floor_requires_log_power(
                references,
                calibration_scale,
            ) or _requires_scaled_square(
                weighted,
                minimum_power_scale=1.0 - alpha,
            ):
                log_smoothed_power = _exponential_power_log(weighted, alpha)
            else:
                squared = np.square(weighted)
                log_smoothed_power = scipy.signal.lfilter([1.0 - alpha], [1.0, -alpha], squared, axis=-1)
                del squared
                with np.errstate(divide="ignore", invalid="ignore"):
                    np.log(log_smoothed_power, out=log_smoothed_power)
            with np.errstate(divide="ignore", invalid="ignore"):
                np.add(
                    log_smoothed_power,
                    2.0 * np.log(calibration_scale[:, np.newaxis]),
                    out=log_smoothed_power,
                )
                np.subtract(
                    log_smoothed_power,
                    2.0 * np.log(references[:, np.newaxis]),
                    out=log_smoothed_power,
                )
            result = _bounded_db_from_log_ratio(
                log_smoothed_power,
                scale=10.0,
                ratio_floor=MIN_SOUND_LEVEL_POWER_RATIO,
            )
        else:
            # Preserve the released linear RMS path exactly. Numerical scaling
            # belongs only to the versioned dB contract.
            squared = np.square(weighted)
            smoothed = scipy.signal.lfilter([1.0 - alpha], [1.0, -alpha], squared, axis=-1)
            result = np.sqrt(smoothed)
        logger.debug(f"Sound level applied, returning result with shape: {result.shape}")
        return np.asarray(result, dtype=output_dtype)

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        """Return sound level output dtype metadata."""
        return self._output_dtype(input_dtype)

Attributes

name = 'sound_level' class-attribute instance-attribute

ref property

Reference values captured at operation construction time.

freq_weighting property

Frequency weighting captured at operation construction time.

time_weighting property

Time weighting captured at operation construction time.

dB property

Whether output is converted to decibels.

time_constant property

Return the RC time constant in seconds.

Functions

__init__(sampling_rate, ref=1.0, freq_weighting='Z', time_weighting='Fast', dB=False, *, _calibration_scale=1.0)

Initialize a frequency- and time-weighted level operation.

Parameters:

Name	Type	Description	Default
sampling_rate	float	Input sampling rate in Hz.	required
ref	list[float] | float | NDArrayReal	Positive finite amplitude reference, either one scalar or one value per channel. For Pa input, 2e-5 produces dB SPL.	1.0
freq_weighting	str | None	Implemented frequency curve: "A", "C", or flat "Z". None is treated as "Z".	'Z'
time_weighting	str	Exponential time constant: "Fast" (125 ms) or "Slow" (1 s). The short forms "F" and "S" are accepted too.	'Fast'
dB	bool	If True, return 10 * log10 of smoothed power relative to ref**2; otherwise return linear weighted RMS in the input unit.	False
_calibration_scale	list[float] | float | NDArrayReal	Positive internal amplitude scale supplied by calibrated Frame execution. It is not a public recipe parameter.	1.0

Raises:

Type	Description
ValueError	If the sampling rate, reference, calibration scale, frequency curve, or time weighting is invalid.

Source code in wandas/processing/temporal.py

def __init__(
    self,
    sampling_rate: float,
    ref: list[float] | float | NDArrayReal = 1.0,
    freq_weighting: str | None = "Z",
    time_weighting: str = "Fast",
    dB: bool = False,
    *,
    _calibration_scale: list[float] | float | NDArrayReal = 1.0,
) -> None:
    """Initialize a frequency- and time-weighted level operation.

    Args:
        sampling_rate: Input sampling rate in Hz.
        ref: Positive finite amplitude reference, either one scalar or one
            value per channel. For Pa input, ``2e-5`` produces dB SPL.
        freq_weighting: Implemented frequency curve: ``"A"``, ``"C"``, or
            flat ``"Z"``. ``None`` is treated as ``"Z"``.
        time_weighting: Exponential time constant: ``"Fast"`` (125 ms) or
            ``"Slow"`` (1 s). The short forms ``"F"`` and ``"S"`` are
            accepted too.
        dB: If True, return ``10 * log10`` of smoothed power relative to
            ``ref**2``; otherwise return linear weighted RMS in the input
            unit.
        _calibration_scale: Positive internal amplitude scale supplied by
            calibrated Frame execution. It is not a public recipe
            parameter.

    Raises:
        ValueError: If the sampling rate, reference, calibration scale,
            frequency curve, or time weighting is invalid.
    """
    validate_sampling_rate(sampling_rate)
    ref_array = np.atleast_1d(np.array(ref, dtype=float, copy=True))
    if ref_array.size == 0 or np.any(~np.isfinite(ref_array)) or np.any(ref_array <= 0):
        raise ValueError(
            "Invalid sound level reference\n"
            f"  Got: {ref_array.tolist()}\n"
            "  Expected: Positive finite reference values\n"
            "Reference-relative dB output requires a positive finite amplitude reference."
        )
    calibration_scale = _validated_calibration_scale(
        _calibration_scale,
        operation_label="sound level",
    )
    normalized_freq_weighting = self._normalize_freq_weighting(freq_weighting)
    normalized_time_weighting = self._normalize_time_weighting(time_weighting)
    super().__init__(
        sampling_rate,
        ref=ref_array,
        freq_weighting=normalized_freq_weighting,
        time_weighting=normalized_time_weighting,
        dB=dB,
    )
    object.__setattr__(self, "_calibration_scale", calibration_scale)

get_display_name()

Get display name for the operation for use in channel labels.

Source code in wandas/processing/temporal.py

def get_display_name(self) -> str:
    """Get display name for the operation for use in channel labels."""
    freq_weighting = self.freq_weighting
    time_weighting = self.time_weighting
    if self.dB:
        return f"L{freq_weighting}{time_weighting[0]}"
    return f"{freq_weighting}{time_weighting[0]}RMS"

calculate_output_shape(input_shape)

Validate channel-wise configuration and preserve input shape.

Parameters:

Name	Type	Description	Default
input_shape	tuple[int, ...]	Input shape, normally (channels, samples).	required

Returns:

Type	Description
tuple[int, ...]	The unchanged input shape. sound_level is sample-wise and keeps the input sampling rate.

Raises:

Type	Description
ValueError	If a per-channel reference or calibration scale cannot be broadcast to the input channel count.

Source code in wandas/processing/temporal.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """Validate channel-wise configuration and preserve input shape.

    Args:
        input_shape: Input shape, normally ``(channels, samples)``.

    Returns:
        The unchanged input shape. ``sound_level`` is sample-wise and keeps
            the input sampling rate.

    Raises:
        ValueError: If a per-channel reference or calibration scale cannot
            be broadcast to the input channel count.
    """
    if self.dB:
        self._reference_values(input_shape[0])
        self._calibration_scale_values(input_shape[0])
    return input_shape

calculate_output_dtype(input_dtype, *input_dtypes)

Return sound level output dtype metadata.

Source code in wandas/processing/temporal.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    """Return sound level output dtype metadata."""
    return self._output_dtype(input_dtype)

Functions

wandas.processing.spectral

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

FFT

Bases: AudioOperation[NDArrayReal, NDArrayComplex]

One-sided, coherent-gain-normalized peak-amplitude FFT.

The input is truncated to n_fft; the selected window spans the remaining min(len(x), n_fft) samples and is applied before zero-padding to n_fft. The coherent gain is taken from that same window, so a short input keeps its amplitude. DC and Nyquist bins retain their real-FFT scaling; every other positive-frequency bin is doubled. The complex result therefore has the same physical unit as the input, and an on-bin sinusoid's magnitude is its peak amplitude.

Source code in wandas/processing/spectral.py

class FFT(AudioOperation[NDArrayReal, NDArrayComplex]):
    """One-sided, coherent-gain-normalized peak-amplitude FFT.

    The input is truncated to ``n_fft``; the selected window spans the
    remaining ``min(len(x), n_fft)`` samples and is applied before zero-padding
    to ``n_fft``. The coherent gain is taken from that same window, so a short
    input keeps its amplitude. DC and Nyquist bins retain their real-FFT
    scaling; every other positive-frequency bin is doubled. The complex result
    therefore has the same physical unit as the input, and an on-bin sinusoid's
    magnitude is its peak amplitude.
    """

    name = "fft"
    _display = "FFT"

    def __init__(self, sampling_rate: float, n_fft: int | None = None, window: str = "hann"):
        """
        Initialize FFT operation

        Args:
            sampling_rate: float. Sampling rate (Hz)
            n_fft: int, optional. FFT size, default is None (determined by input size)
            window: str, optional. Window function type, default is 'hann'

        Raises:
            ValueError: If n_fft is not a positive integer
        """
        # Validate n_fft parameter
        if n_fft is not None and n_fft <= 0:
            raise ValueError(
                f"Invalid FFT size\n"
                f"  Got: {n_fft}\n"
                f"  Expected: Positive integer > 0\n"
                f"FFT size must be a positive integer.\n"
                f"Common values: 512, 1024, 2048, 4096,\n"
                f"8192 (powers of 2 are most efficient)"
            )

        super().__init__(sampling_rate, n_fft=n_fft, window=window)

    @property
    def n_fft(self) -> int | None:
        """FFT size captured at operation construction time."""
        return self._config_value("n_fft")

    @property
    def window(self) -> str:
        """Window name captured at operation construction time."""
        return self._config_value("window")

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        """
        Calculate output data shape after the operation.

        Args:
            input_shape: tuple. Input data shape (channels, samples).

        Returns:
            tuple: Output data shape (channels, freqs).
        """
        n_fft = self.n_fft
        n_freqs = n_fft // 2 + 1 if n_fft else input_shape[-1] // 2 + 1
        return (*input_shape[:-1], n_freqs)

    _keeps_input_precision = True

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        return precision_complex_dtype(input_dtype)

    def _process(self, x: NDArrayReal) -> NDArrayComplex:
        """Window the available samples, then zero-pad to ``n_fft``."""
        fft_size = int(x.shape[-1]) if self.n_fft is None else self.n_fft
        x = x[..., :fft_size]
        win = scipy.signal.get_window(self.window, x.shape[-1])
        result: NDArrayComplex = np.fft.rfft(x * win, n=fft_size, axis=-1)
        return _normalize_rfft_amplitude(
            result,
            n_fft=fft_size,
            window_gain=float(np.sum(win)),
        )

Attributes

name = 'fft' class-attribute instance-attribute

n_fft property

FFT size captured at operation construction time.

window property

Window name captured at operation construction time.

Functions

__init__(sampling_rate, n_fft=None, window='hann')

Initialize FFT operation

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate (Hz)	required
n_fft	int | None	int, optional. FFT size, default is None (determined by input size)	None
window	str	str, optional. Window function type, default is 'hann'	'hann'

Raises:

Type	Description
ValueError	If n_fft is not a positive integer

Source code in wandas/processing/spectral.py

def __init__(self, sampling_rate: float, n_fft: int | None = None, window: str = "hann"):
    """
    Initialize FFT operation

    Args:
        sampling_rate: float. Sampling rate (Hz)
        n_fft: int, optional. FFT size, default is None (determined by input size)
        window: str, optional. Window function type, default is 'hann'

    Raises:
        ValueError: If n_fft is not a positive integer
    """
    # Validate n_fft parameter
    if n_fft is not None and n_fft <= 0:
        raise ValueError(
            f"Invalid FFT size\n"
            f"  Got: {n_fft}\n"
            f"  Expected: Positive integer > 0\n"
            f"FFT size must be a positive integer.\n"
            f"Common values: 512, 1024, 2048, 4096,\n"
            f"8192 (powers of 2 are most efficient)"
        )

    super().__init__(sampling_rate, n_fft=n_fft, window=window)

calculate_output_shape(input_shape)

Calculate output data shape after the operation.

Parameters:

Name	Type	Description	Default
input_shape	tuple[int, ...]	tuple. Input data shape (channels, samples).	required

Returns:

Name	Type	Description
tuple	tuple[int, ...]	Output data shape (channels, freqs).

Source code in wandas/processing/spectral.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """
    Calculate output data shape after the operation.

    Args:
        input_shape: tuple. Input data shape (channels, samples).

    Returns:
        tuple: Output data shape (channels, freqs).
    """
    n_fft = self.n_fft
    n_freqs = n_fft // 2 + 1 if n_fft else input_shape[-1] // 2 + 1
    return (*input_shape[:-1], n_freqs)

calculate_output_dtype(input_dtype, *input_dtypes)

Source code in wandas/processing/spectral.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    return precision_complex_dtype(input_dtype)

IFFT

Bases: AudioOperation[NDArrayComplex, NDArrayReal]

Inverse of Wandas' one-sided peak-amplitude FFT normalization.

For a spectrum produced by :class:FFT with matching n_fft, window and win_length, the result is the analysed samples multiplied by that analysis window and zero-padded to n_fft. A boxcar window therefore reconstructs the prepared input exactly; tapered windows intentionally reconstruct the windowed waveform rather than guessing samples discarded by the analysis window.

Source code in wandas/processing/spectral.py

class IFFT(AudioOperation[NDArrayComplex, NDArrayReal]):
    """Inverse of Wandas' one-sided peak-amplitude FFT normalization.

    For a spectrum produced by :class:`FFT` with matching ``n_fft``,
    ``window`` and ``win_length``, the result is the analysed samples
    multiplied by that analysis window and zero-padded to ``n_fft``. A boxcar
    window therefore reconstructs the prepared input exactly; tapered windows
    intentionally reconstruct the windowed waveform rather than guessing
    samples discarded by the analysis window.
    """

    name = "ifft"
    _display = "iFFT"

    def __init__(
        self,
        sampling_rate: float,
        n_fft: int | None = None,
        window: str = "hann",
        win_length: int | None = None,
    ):
        """
        Initialize IFFT operation

        Args:
            sampling_rate: float. Sampling rate (Hz)
            n_fft: Optional[int], optional. IFFT size, default is None (determined based on input size)
            window: str, optional. Window function type, default is 'hann'
            win_length: Optional[int], optional. Length of the analysis window whose
                coherent gain is undone, default is None (the IFFT size)
        """
        config: dict[str, Any] = {"n_fft": n_fft, "window": window}
        # Omitted when unset so existing IFFT configurations stay identical.
        if win_length is not None:
            if (
                isinstance(win_length, bool)
                or not isinstance(win_length, int | np.integer)
                or win_length < 1
                or (n_fft is not None and win_length > n_fft)
            ):
                upper_bound = "" if n_fft is None else f" no larger than n_fft ({n_fft})"
                raise ValueError(
                    "Invalid win_length for IFFT\n"
                    f"  Got: {win_length!r}\n"
                    f"  Expected: a positive integer{upper_bound}\n"
                    "Pass the length of the analysis window applied before zero-padding."
                )
            config["win_length"] = int(win_length)
        super().__init__(sampling_rate, **config)

    @property
    def win_length(self) -> int | None:
        """Analysis window length captured at operation construction time."""
        return self._config.get("win_length")

    @property
    def n_fft(self) -> int | None:
        """IFFT size captured at operation construction time."""
        return self._config_value("n_fft")

    @property
    def window(self) -> str:
        """Window name captured at operation construction time."""
        return self._config_value("window")

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        """
        Calculate output data shape after operation

        Args:
            input_shape: tuple. Input data shape (channels, freqs)

        Returns:
            tuple: Output data shape (channels, samples)
        """
        n_fft = self.n_fft
        n_samples = 2 * (input_shape[-1] - 1) if n_fft is None else n_fft
        return (*input_shape[:-1], n_samples)

    _keeps_input_precision = True

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        return precision_real_dtype(input_dtype)

    def _process(self, x: NDArrayComplex) -> NDArrayReal:
        """Invert Wandas peak-amplitude scaling to a windowed waveform."""
        logger.debug(f"Applying IFFT to array with shape: {x.shape}")

        fft_size = 2 * (int(x.shape[-1]) - 1) if self.n_fft is None else self.n_fft
        if self.win_length is not None and self.win_length > fft_size:
            raise ValueError(
                "Invalid win_length for IFFT\n"
                f"  Got: {self.win_length}\n"
                f"  Expected: at most the IFFT size ({fft_size}) inferred from the input\n"
                "Pass n_fft, or a win_length that fits the spectrum."
            )
        win = scipy.signal.get_window(self.window, fft_size if self.win_length is None else self.win_length)
        _x = _denormalize_rfft_amplitude(
            x,
            n_fft=fft_size,
            window_gain=float(np.sum(win)),
        )

        result: NDArrayReal = np.fft.irfft(_x, n=fft_size, axis=-1)

        logger.debug(f"IFFT applied, returning result with shape: {result.shape}")
        return result

Attributes

name = 'ifft' class-attribute instance-attribute

win_length property

Analysis window length captured at operation construction time.

n_fft property

IFFT size captured at operation construction time.

window property

Window name captured at operation construction time.

Functions

__init__(sampling_rate, n_fft=None, window='hann', win_length=None)

Initialize IFFT operation

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate (Hz)	required
n_fft	int | None	Optional[int], optional. IFFT size, default is None (determined based on input size)	None
window	str	str, optional. Window function type, default is 'hann'	'hann'
win_length	int | None	Optional[int], optional. Length of the analysis window whose coherent gain is undone, default is None (the IFFT size)	None

Source code in wandas/processing/spectral.py

def __init__(
    self,
    sampling_rate: float,
    n_fft: int | None = None,
    window: str = "hann",
    win_length: int | None = None,
):
    """
    Initialize IFFT operation

    Args:
        sampling_rate: float. Sampling rate (Hz)
        n_fft: Optional[int], optional. IFFT size, default is None (determined based on input size)
        window: str, optional. Window function type, default is 'hann'
        win_length: Optional[int], optional. Length of the analysis window whose
            coherent gain is undone, default is None (the IFFT size)
    """
    config: dict[str, Any] = {"n_fft": n_fft, "window": window}
    # Omitted when unset so existing IFFT configurations stay identical.
    if win_length is not None:
        if (
            isinstance(win_length, bool)
            or not isinstance(win_length, int | np.integer)
            or win_length < 1
            or (n_fft is not None and win_length > n_fft)
        ):
            upper_bound = "" if n_fft is None else f" no larger than n_fft ({n_fft})"
            raise ValueError(
                "Invalid win_length for IFFT\n"
                f"  Got: {win_length!r}\n"
                f"  Expected: a positive integer{upper_bound}\n"
                "Pass the length of the analysis window applied before zero-padding."
            )
        config["win_length"] = int(win_length)
    super().__init__(sampling_rate, **config)

calculate_output_shape(input_shape)

Calculate output data shape after operation

Parameters:

Name	Type	Description	Default
input_shape	tuple[int, ...]	tuple. Input data shape (channels, freqs)	required

Returns:

Name	Type	Description
tuple	tuple[int, ...]	Output data shape (channels, samples)

Source code in wandas/processing/spectral.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """
    Calculate output data shape after operation

    Args:
        input_shape: tuple. Input data shape (channels, freqs)

    Returns:
        tuple: Output data shape (channels, samples)
    """
    n_fft = self.n_fft
    n_samples = 2 * (input_shape[-1] - 1) if n_fft is None else n_fft
    return (*input_shape[:-1], n_samples)

calculate_output_dtype(input_dtype, *input_dtypes)

Source code in wandas/processing/spectral.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    return precision_real_dtype(input_dtype)

STFT

Bases: AudioOperation[NDArrayReal, NDArrayComplex]

One-sided peak-amplitude Short-Time Fourier Transform.

Each frame uses SciPy's coherent-gain magnitude scaling, with non-DC and non-Nyquist positive-frequency bins doubled. Values retain the input physical unit.

Source code in wandas/processing/spectral.py

class STFT(AudioOperation[NDArrayReal, NDArrayComplex]):
    """One-sided peak-amplitude Short-Time Fourier Transform.

    Each frame uses SciPy's coherent-gain magnitude scaling, with non-DC and
    non-Nyquist positive-frequency bins doubled. Values retain the input
    physical unit.
    """

    name = "stft"
    _display = "STFT"

    def __init__(
        self,
        sampling_rate: float,
        n_fft: int = 2048,
        hop_length: int | None = None,
        win_length: int | None = None,
        window: str = "hann",
        *,
        allow_sparse: bool = False,
        detrend: str | None = None,
    ):
        """
        Initialize STFT operation

        Args:
            sampling_rate: float. Sampling rate (Hz)
            n_fft: int. FFT size, default is 2048
            hop_length: int, optional. Number of samples between frames. Default is win_length // 4
            win_length: int, optional. Window length. Default is n_fft
            window: str. Window type, default is 'hann'
            allow_sparse: Permit gaps between windows when hop_length exceeds win_length.
            detrend: None, ``"constant"`` or ``"linear"``. Trend removed from each frame's
                samples before windowing, as in SciPy ``ShortTimeFFT.stft_detrend``.

        Raises:
            ValueError: If n_fft is not positive, win_length > n_fft, or hop_length is invalid
        """
        validate_stft_recipe_params({"detrend": detrend})
        # Validate and compute parameters
        actual_win_length, actual_hop_length = _validate_spectral_params(
            n_fft, win_length, hop_length, "STFT", allow_sparse=allow_sparse
        )

        self._SFT = scipy.signal.ShortTimeFFT(
            win=scipy.signal.get_window(window, actual_win_length),
            hop=actual_hop_length,
            fs=sampling_rate,
            mfft=n_fft,
            scale_to="magnitude",
        )
        config: dict[str, Any] = {
            "n_fft": n_fft,
            "win_length": actual_win_length,
            "hop_length": actual_hop_length,
            "window": window,
            "allow_sparse": allow_sparse,
        }
        # Omitted when unset so existing STFT configurations stay identical.
        if detrend is not None:
            config["detrend"] = detrend
        super().__init__(sampling_rate, **config)

    @property
    def n_fft(self) -> int:
        """FFT size captured at operation construction time."""
        return self._config_value("n_fft")

    @property
    def detrend(self) -> str | None:
        """Per-frame trend removal captured at operation construction time."""
        return self._config.get("detrend")

    @property
    def win_length(self) -> int:
        """Window length captured at operation construction time."""
        return self._config_value("win_length")

    @property
    def hop_length(self) -> int:
        """Hop length captured at operation construction time."""
        return self._config_value("hop_length")

    @property
    def window(self) -> str:
        """Window name captured at operation construction time."""
        return self._config_value("window")

    @property
    def frame_time_origin(self) -> float:
        """First actual frame center relative to the input start, in seconds.

        SciPy's padding-dependent minimum frame index is determined without
        computing signal samples. This derived domain state is not a Recipe
        parameter or a second copy of the window/hop configuration.
        """
        return float(self._SFT.p_min * self._SFT.delta_t)

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        """
        Calculate output data shape after operation

        Args:
            input_shape: tuple. Input data shape

        Returns:
            tuple: Output data shape
        """
        n_channels = input_shape[0]
        n_samples = input_shape[-1]
        n_f = len(self._SFT.f)
        n_t = len(self._SFT.t(n_samples))
        return (n_channels, n_f, n_t)

    _keeps_input_precision = True

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        return precision_complex_dtype(input_dtype)

    def _process(self, x: NDArrayReal) -> NDArrayComplex:
        """Apply SciPy STFT processing to multiple channels at once"""
        logger.debug(f"Applying SciPy STFT to array with shape: {x.shape}")

        # Convert 1D input to 2D
        if x.ndim == 1:
            x = x.reshape(1, -1)

        # Apply STFT to all channels at once
        result: NDArrayComplex = (
            self._SFT.stft(x)
            if self.detrend is None
            else self._SFT.stft_detrend(x, cast(Literal["constant", "linear"], self.detrend))
        )
        result = _normalize_rfft_amplitude(
            result,
            n_fft=self.n_fft,
            window_gain=1.0,
            axis=-2,
        )
        logger.debug(f"SciPy STFT applied, returning result with shape: {result.shape}")
        return result

Attributes

name = 'stft' class-attribute instance-attribute

n_fft property

FFT size captured at operation construction time.

detrend property

Per-frame trend removal captured at operation construction time.

win_length property

Window length captured at operation construction time.

hop_length property

Hop length captured at operation construction time.

window property

Window name captured at operation construction time.

frame_time_origin property

First actual frame center relative to the input start, in seconds.

SciPy's padding-dependent minimum frame index is determined without computing signal samples. This derived domain state is not a Recipe parameter or a second copy of the window/hop configuration.

Functions

__init__(sampling_rate, n_fft=2048, hop_length=None, win_length=None, window='hann', *, allow_sparse=False, detrend=None)

Initialize STFT operation

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate (Hz)	required
n_fft	int	int. FFT size, default is 2048	2048
hop_length	int | None	int, optional. Number of samples between frames. Default is win_length // 4	None
win_length	int | None	int, optional. Window length. Default is n_fft	None
window	str	str. Window type, default is 'hann'	'hann'
allow_sparse	bool	Permit gaps between windows when hop_length exceeds win_length.	False
detrend	str | None	None, "constant" or "linear". Trend removed from each frame's samples before windowing, as in SciPy ShortTimeFFT.stft_detrend.	None

Raises:

Type	Description
ValueError	If n_fft is not positive, win_length > n_fft, or hop_length is invalid

Source code in wandas/processing/spectral.py

def __init__(
    self,
    sampling_rate: float,
    n_fft: int = 2048,
    hop_length: int | None = None,
    win_length: int | None = None,
    window: str = "hann",
    *,
    allow_sparse: bool = False,
    detrend: str | None = None,
):
    """
    Initialize STFT operation

    Args:
        sampling_rate: float. Sampling rate (Hz)
        n_fft: int. FFT size, default is 2048
        hop_length: int, optional. Number of samples between frames. Default is win_length // 4
        win_length: int, optional. Window length. Default is n_fft
        window: str. Window type, default is 'hann'
        allow_sparse: Permit gaps between windows when hop_length exceeds win_length.
        detrend: None, ``"constant"`` or ``"linear"``. Trend removed from each frame's
            samples before windowing, as in SciPy ``ShortTimeFFT.stft_detrend``.

    Raises:
        ValueError: If n_fft is not positive, win_length > n_fft, or hop_length is invalid
    """
    validate_stft_recipe_params({"detrend": detrend})
    # Validate and compute parameters
    actual_win_length, actual_hop_length = _validate_spectral_params(
        n_fft, win_length, hop_length, "STFT", allow_sparse=allow_sparse
    )

    self._SFT = scipy.signal.ShortTimeFFT(
        win=scipy.signal.get_window(window, actual_win_length),
        hop=actual_hop_length,
        fs=sampling_rate,
        mfft=n_fft,
        scale_to="magnitude",
    )
    config: dict[str, Any] = {
        "n_fft": n_fft,
        "win_length": actual_win_length,
        "hop_length": actual_hop_length,
        "window": window,
        "allow_sparse": allow_sparse,
    }
    # Omitted when unset so existing STFT configurations stay identical.
    if detrend is not None:
        config["detrend"] = detrend
    super().__init__(sampling_rate, **config)

calculate_output_shape(input_shape)

Calculate output data shape after operation

Parameters:

Name	Type	Description	Default
input_shape	tuple[int, ...]	tuple. Input data shape	required

Returns:

Name	Type	Description
tuple	tuple[int, ...]	Output data shape

Source code in wandas/processing/spectral.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """
    Calculate output data shape after operation

    Args:
        input_shape: tuple. Input data shape

    Returns:
        tuple: Output data shape
    """
    n_channels = input_shape[0]
    n_samples = input_shape[-1]
    n_f = len(self._SFT.f)
    n_t = len(self._SFT.t(n_samples))
    return (n_channels, n_f, n_t)

calculate_output_dtype(input_dtype, *input_dtypes)

Source code in wandas/processing/spectral.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    return precision_complex_dtype(input_dtype)

ISTFT

Bases: AudioOperation[NDArrayComplex, NDArrayReal]

Inverse Short-Time Fourier Transform operation

Source code in wandas/processing/spectral.py

class ISTFT(AudioOperation[NDArrayComplex, NDArrayReal]):
    """Inverse Short-Time Fourier Transform operation"""

    name = "istft"
    _display = "iSTFT"

    def __init__(
        self,
        sampling_rate: float,
        n_fft: int = 2048,
        hop_length: int | None = None,
        win_length: int | None = None,
        window: str = "hann",
        length: int | None = None,
    ):
        """
        Initialize ISTFT operation

        Args:
            sampling_rate: float. Sampling rate (Hz)
            n_fft: int. FFT size, default is 2048
            hop_length: int, optional. Number of samples between frames. Default is win_length // 4
            win_length: int, optional. Window length. Default is n_fft
            window: str. Window type, default is 'hann'
            length: int, optional. Length of output signal. Default is None (determined from input)

        Raises:
            ValueError: If n_fft is not positive, win_length > n_fft, or hop_length is invalid
        """
        # Validate and compute parameters
        actual_win_length, actual_hop_length = _validate_spectral_params(n_fft, win_length, hop_length, "ISTFT")

        # Instantiate ShortTimeFFT for ISTFT calculation
        self._SFT = scipy.signal.ShortTimeFFT(
            win=scipy.signal.get_window(window, actual_win_length),
            hop=actual_hop_length,
            fs=sampling_rate,
            mfft=n_fft,
            scale_to="magnitude",  # Consistent scaling with STFT
        )

        super().__init__(
            sampling_rate,
            n_fft=n_fft,
            win_length=actual_win_length,
            hop_length=actual_hop_length,
            window=window,
            length=length,
        )

    @property
    def n_fft(self) -> int:
        """FFT size captured at operation construction time."""
        return self._config_value("n_fft")

    @property
    def win_length(self) -> int:
        """Window length captured at operation construction time."""
        return self._config_value("win_length")

    @property
    def hop_length(self) -> int:
        """Hop length captured at operation construction time."""
        return self._config_value("hop_length")

    @property
    def window(self) -> str:
        """Window name captured at operation construction time."""
        return self._config_value("window")

    @property
    def length(self) -> int | None:
        """Output length captured at operation construction time."""
        return self._config_value("length")

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        """
        Calculate output data shape after ISTFT operation.

        Uses the SciPy ShortTimeFFT calculation formula to compute the expected
        output length based on the input spectrogram dimensions and output range
        parameters (k0, k1).

        Args:
            input_shape: tuple. Input spectrogram shape (channels, n_freqs, n_frames)
                where n_freqs = n_fft // 2 + 1 and n_frames is the number of time frames.

        Returns:
            tuple: Output shape (channels, output_samples) where output_samples is the
                reconstructed signal length determined by the output range [k0, k1).

        Notes:
            The calculation follows SciPy's ShortTimeFFT.istft() implementation.
            When k1 is None (default), the maximum reconstructible signal length is
            computed as:

            .. math::

            q_{max} = n_{frames} + p_{min}

            k_{max} = (q_{max} - 1) \\cdot hop + m_{num} - m_{num\\_mid}

            The output length is then:

            .. math::

            output\\_samples = k_1 - k_0

            where k0 defaults to 0 and k1 defaults to k_max.

            Parameters that affect the calculation:
            - n_frames: number of time frames in the STFT
            - p_min: minimum frame index (ShortTimeFFT property)
            - hop: hop length (samples between frames)
            - m_num: window length
            - m_num_mid: window midpoint position
            - length: optional length override (if set, limits output)

        References:
            - SciPy ShortTimeFFT.istft:
          https://docs.scipy.org/doc/scipy/reference/generated/scipy.signal.ShortTimeFFT.istft.html
            - SciPy Source: https://github.com/scipy/scipy/blob/main/scipy/signal/_short_time_fft.py
        """
        n_channels = input_shape[0]
        n_frames = input_shape[-1]  # time_frames

        # Follow SciPy ShortTimeFFT formula
        # See: https://github.com/scipy/scipy/blob/main/scipy/signal/_short_time_fft.py
        q_max = n_frames + self._SFT.p_min
        k_max = (q_max - 1) * self._SFT.hop + self._SFT.m_num - self._SFT.m_num_mid

        # Default parameters: k0=0, k1=None (which becomes k_max)
        # The output length is k1 - k0 = k_max - 0 = k_max
        k0 = 0
        k1 = k_max

        # If length is specified, it acts as an override to limit the output
        length = self.length
        if length is not None:
            k1 = min(length, k1)

        output_samples = k1 - k0

        return (n_channels, output_samples)

    _keeps_input_precision = True

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        return precision_real_dtype(input_dtype)

    def _process(self, x: NDArrayComplex) -> NDArrayReal:
        """
        Apply SciPy ISTFT processing to multiple channels at once using ShortTimeFFT"""
        logger.debug(f"Applying SciPy ISTFT (ShortTimeFFT) to array with shape: {x.shape}")

        # Convert 2D input to 3D (assume single channel)
        if x.ndim == 2:
            x = x.reshape(1, *x.shape)

        # Adjust scaling back if STFT applied factor of 2
        _x = _denormalize_rfft_amplitude(
            x,
            n_fft=self.n_fft,
            window_gain=1.0,
            axis=-2,
        )

        # Apply ISTFT using the ShortTimeFFT instance
        result: NDArrayReal = self._SFT.istft(_x)

        # Trim to desired length if specified
        length = self.length
        if length is not None:
            result = result[..., :length]

        logger.debug(f"ShortTimeFFT applied, returning result with shape: {result.shape}")
        return result

    def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
        """Execute ISTFT on Frame-internal channel-first spectrogram data."""
        self._validate_process_inputs(data, *inputs, ndim=3)
        return super().process(data, *inputs)

Attributes

name = 'istft' class-attribute instance-attribute

n_fft property

FFT size captured at operation construction time.

win_length property

Window length captured at operation construction time.

hop_length property

Hop length captured at operation construction time.

window property

Window name captured at operation construction time.

length property

Output length captured at operation construction time.

Functions

__init__(sampling_rate, n_fft=2048, hop_length=None, win_length=None, window='hann', length=None)

Initialize ISTFT operation

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate (Hz)	required
n_fft	int	int. FFT size, default is 2048	2048
hop_length	int | None	int, optional. Number of samples between frames. Default is win_length // 4	None
win_length	int | None	int, optional. Window length. Default is n_fft	None
window	str	str. Window type, default is 'hann'	'hann'
length	int | None	int, optional. Length of output signal. Default is None (determined from input)	None

Raises:

Type	Description
ValueError	If n_fft is not positive, win_length > n_fft, or hop_length is invalid

Source code in wandas/processing/spectral.py

def __init__(
    self,
    sampling_rate: float,
    n_fft: int = 2048,
    hop_length: int | None = None,
    win_length: int | None = None,
    window: str = "hann",
    length: int | None = None,
):
    """
    Initialize ISTFT operation

    Args:
        sampling_rate: float. Sampling rate (Hz)
        n_fft: int. FFT size, default is 2048
        hop_length: int, optional. Number of samples between frames. Default is win_length // 4
        win_length: int, optional. Window length. Default is n_fft
        window: str. Window type, default is 'hann'
        length: int, optional. Length of output signal. Default is None (determined from input)

    Raises:
        ValueError: If n_fft is not positive, win_length > n_fft, or hop_length is invalid
    """
    # Validate and compute parameters
    actual_win_length, actual_hop_length = _validate_spectral_params(n_fft, win_length, hop_length, "ISTFT")

    # Instantiate ShortTimeFFT for ISTFT calculation
    self._SFT = scipy.signal.ShortTimeFFT(
        win=scipy.signal.get_window(window, actual_win_length),
        hop=actual_hop_length,
        fs=sampling_rate,
        mfft=n_fft,
        scale_to="magnitude",  # Consistent scaling with STFT
    )

    super().__init__(
        sampling_rate,
        n_fft=n_fft,
        win_length=actual_win_length,
        hop_length=actual_hop_length,
        window=window,
        length=length,
    )

calculate_output_shape(input_shape)

Calculate output data shape after ISTFT operation.

Uses the SciPy ShortTimeFFT calculation formula to compute the expected output length based on the input spectrogram dimensions and output range parameters (k0, k1).

Parameters:

Name	Type	Description	Default
input_shape	tuple[int, ...]	tuple. Input spectrogram shape (channels, n_freqs, n_frames) where n_freqs = n_fft // 2 + 1 and n_frames is the number of time frames.	required

Returns:

Name	Type	Description
tuple	tuple[int, ...]	Output shape (channels, output_samples) where output_samples is the reconstructed signal length determined by the output range [k0, k1).

Notes

The calculation follows SciPy's ShortTimeFFT.istft() implementation. When k1 is None (default), the maximum reconstructible signal length is computed as:

.. math::

q_{max} = n_{frames} + p_{min}

k_{max} = (q_{max} - 1) \cdot hop + m_{num} - m_{num_mid}

The output length is then:

.. math::

output_samples = k_1 - k_0

where k0 defaults to 0 and k1 defaults to k_max.

Parameters that affect the calculation: - n_frames: number of time frames in the STFT - p_min: minimum frame index (ShortTimeFFT property) - hop: hop length (samples between frames) - m_num: window length - m_num_mid: window midpoint position - length: optional length override (if set, limits output)

References

• SciPy ShortTimeFFT.istft:

https://docs.scipy.org/doc/scipy/reference/generated/scipy.signal.ShortTimeFFT.istft.html - SciPy Source: https://github.com/scipy/scipy/blob/main/scipy/signal/_short_time_fft.py

Source code in wandas/processing/spectral.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """
    Calculate output data shape after ISTFT operation.

    Uses the SciPy ShortTimeFFT calculation formula to compute the expected
    output length based on the input spectrogram dimensions and output range
    parameters (k0, k1).

    Args:
        input_shape: tuple. Input spectrogram shape (channels, n_freqs, n_frames)
            where n_freqs = n_fft // 2 + 1 and n_frames is the number of time frames.

    Returns:
        tuple: Output shape (channels, output_samples) where output_samples is the
            reconstructed signal length determined by the output range [k0, k1).

    Notes:
        The calculation follows SciPy's ShortTimeFFT.istft() implementation.
        When k1 is None (default), the maximum reconstructible signal length is
        computed as:

        .. math::

        q_{max} = n_{frames} + p_{min}

        k_{max} = (q_{max} - 1) \\cdot hop + m_{num} - m_{num\\_mid}

        The output length is then:

        .. math::

        output\\_samples = k_1 - k_0

        where k0 defaults to 0 and k1 defaults to k_max.

        Parameters that affect the calculation:
        - n_frames: number of time frames in the STFT
        - p_min: minimum frame index (ShortTimeFFT property)
        - hop: hop length (samples between frames)
        - m_num: window length
        - m_num_mid: window midpoint position
        - length: optional length override (if set, limits output)

    References:
        - SciPy ShortTimeFFT.istft:
      https://docs.scipy.org/doc/scipy/reference/generated/scipy.signal.ShortTimeFFT.istft.html
        - SciPy Source: https://github.com/scipy/scipy/blob/main/scipy/signal/_short_time_fft.py
    """
    n_channels = input_shape[0]
    n_frames = input_shape[-1]  # time_frames

    # Follow SciPy ShortTimeFFT formula
    # See: https://github.com/scipy/scipy/blob/main/scipy/signal/_short_time_fft.py
    q_max = n_frames + self._SFT.p_min
    k_max = (q_max - 1) * self._SFT.hop + self._SFT.m_num - self._SFT.m_num_mid

    # Default parameters: k0=0, k1=None (which becomes k_max)
    # The output length is k1 - k0 = k_max - 0 = k_max
    k0 = 0
    k1 = k_max

    # If length is specified, it acts as an override to limit the output
    length = self.length
    if length is not None:
        k1 = min(length, k1)

    output_samples = k1 - k0

    return (n_channels, output_samples)

calculate_output_dtype(input_dtype, *input_dtypes)

Source code in wandas/processing/spectral.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    return precision_real_dtype(input_dtype)

process(data, *inputs)

Execute ISTFT on Frame-internal channel-first spectrogram data.

Source code in wandas/processing/spectral.py

def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
    """Execute ISTFT on Frame-internal channel-first spectrogram data."""
    self._validate_process_inputs(data, *inputs, ndim=3)
    return super().process(data, *inputs)

Welch

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Welch-averaged one-sided peak-amplitude spectrum.

Segment power spectra are averaged with scaling="spectrum" and then converted to peak amplitude. Values retain the input physical unit; they are neither power spectral density nor expressed per hertz. For an on-bin sine wave with peak amplitude A, the corresponding bin is approximately A.

Internally, this uses scipy.signal.welch with scaling="spectrum" and converts the power spectrum to amplitude spectrum:

• DC component (f=0): A = sqrt(P)
• positive non-Nyquist components: A = sqrt(2*P)

Source code in wandas/processing/spectral.py

class Welch(AudioOperation[NDArrayReal, NDArrayReal]):
    """Welch-averaged one-sided peak-amplitude spectrum.

    Segment power spectra are averaged with ``scaling="spectrum"`` and then
    converted to peak amplitude. Values retain the input physical unit; they
    are neither power spectral density nor expressed per hertz. For an on-bin
    sine wave with peak amplitude ``A``, the corresponding bin is approximately
    ``A``.

    Internally, this uses ``scipy.signal.welch`` with ``scaling="spectrum"``
    and converts the power spectrum to amplitude spectrum:

    - DC component (f=0): A = sqrt(P)
    - positive non-Nyquist components: A = sqrt(2*P)
    """

    name = "welch"
    _display = "Welch"

    def __init__(
        self,
        sampling_rate: float,
        n_fft: int = 2048,
        hop_length: int | None = None,
        win_length: int | None = None,
        window: str = "hann",
        average: str = "mean",
        detrend: str = "constant",
    ):
        """
        Initialize Welch operation

        Args:
            sampling_rate: float. Sampling rate (Hz)
            n_fft: int, optional. FFT size, default is 2048
            hop_length: int, optional. Number of samples between frames. Default is win_length // 4
            win_length: int, optional. Window length. Default is n_fft
            window: str, optional. Window function type, default is 'hann'
            average: str, optional. Averaging method, default is 'mean'
            detrend: str, optional. Detrend method, default is 'constant'

        Raises:
            ValueError: If n_fft, win_length, or hop_length are invalid
        """
        # Validate and compute parameters
        actual_win_length, actual_hop_length = _validate_spectral_params(n_fft, win_length, hop_length, "Welch method")

        super().__init__(
            sampling_rate,
            n_fft=n_fft,
            win_length=actual_win_length,
            hop_length=actual_hop_length,
            window=window,
            average=average,
            detrend=detrend,
        )

    @property
    def n_fft(self) -> int:
        """FFT size captured at operation construction time."""
        return self._config_value("n_fft")

    @property
    def win_length(self) -> int:
        """Window length captured at operation construction time."""
        return self._config_value("win_length")

    @property
    def hop_length(self) -> int:
        """Hop length captured at operation construction time."""
        return self._config_value("hop_length")

    @property
    def window(self) -> str:
        """Window name captured at operation construction time."""
        return self._config_value("window")

    @property
    def average(self) -> str:
        """Averaging method captured at operation construction time."""
        return self._config_value("average")

    @property
    def detrend(self) -> str:
        """Detrend method captured at operation construction time."""
        return self._config_value("detrend")

    @property
    def noverlap(self) -> int:
        """Overlap captured at operation construction time."""
        return self.win_length - self.hop_length

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        """
        Calculate output data shape after operation

        Args:
            input_shape: tuple. Input data shape (channels, samples)

        Returns:
            tuple: Output data shape (channels, freqs)
        """
        n_freqs = self.n_fft // 2 + 1
        return (*input_shape[:-1], n_freqs)

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        return _spectral_real_dtype(input_dtype)

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Return a Welch-averaged one-sided peak-amplitude spectrum.

        Converts ``scipy.signal.welch(..., scaling="spectrum")`` power to
        peak amplitude for consistency with FFT/STFT.
        """
        from scipy import signal as ss

        if not isinstance(x, np.ndarray):
            raise ValueError("Welch operation requires a numpy ndarray, but received a non-ndarray.")

        def welch_power(samples: NDArrayReal) -> NDArrayReal:
            _, power = ss.welch(
                samples,
                nperseg=self.win_length,
                noverlap=self.noverlap,
                nfft=self.n_fft,
                window=self.window,
                average=self.average,
                detrend=self.detrend,
                scaling="spectrum",
            )
            return power

        # SciPy holds every segment and its FFT at once, about ten times the input for a
        # long recording. For the mean, call it on runs of whole segments (overlapping by
        # ``noverlap`` so the segments are the same) and weight each run by its segment
        # count. The median of the runs is not the median of all segments, so it keeps one
        # call.
        n_samples = x.shape[-1]
        n_segments = (n_samples - self.win_length) // self.hop_length + 1 if n_samples >= self.win_length else 0
        lanes = max(1, int(np.prod(x.shape[:-1])))
        segments_per_run = max(1, _WELCH_RUN_ELEMENTS // (lanes * self.n_fft))
        if self.average != "mean" or n_segments <= segments_per_run:
            result = welch_power(x)
        else:
            weighted_sum: NDArrayReal | None = None
            for first in range(0, n_segments, segments_per_run):
                count = min(segments_per_run, n_segments - first)
                start = first * self.hop_length
                stop = start + (count - 1) * self.hop_length + self.win_length
                power = welch_power(x[..., start:stop])
                # Scale by a scalar of the spectrum's dtype: NumPy 1.x promotes float32 by a
                # Python int above 65535 to float64.
                run = power * power.dtype.type(count)
                weighted_sum = run if weighted_sum is None else weighted_sum + run
            assert weighted_sum is not None
            result = weighted_sum / weighted_sum.dtype.type(n_segments)

        # Convert power spectrum to amplitude spectrum for consistency with FFT/STFT.
        # scipy.signal.welch with scaling='spectrum' returns a one-sided power spectrum
        # where for a sine wave with amplitude A:
        #   - DC component (f=0): P = A^2 (no factor of 2 since DC is not mirrored)
        #   - AC components (f>0): P = A^2/2 (half power due to one-sided spectrum)
        # To recover amplitude A:
        #   - DC: A = sqrt(P)
        #   - AC: A = sqrt(2*P) = sqrt(2) * sqrt(P)
        result = np.sqrt(result)
        result[_rfft_positive_frequency_bins(result.ndim, n_fft=self.n_fft, axis=-1)] *= np.sqrt(2)

        return result

Attributes

name = 'welch' class-attribute instance-attribute

n_fft property

FFT size captured at operation construction time.

win_length property

Window length captured at operation construction time.

hop_length property

Hop length captured at operation construction time.

window property

Window name captured at operation construction time.

average property

Averaging method captured at operation construction time.

detrend property

Detrend method captured at operation construction time.

noverlap property

Overlap captured at operation construction time.

Functions

__init__(sampling_rate, n_fft=2048, hop_length=None, win_length=None, window='hann', average='mean', detrend='constant')

Initialize Welch operation

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate (Hz)	required
n_fft	int	int, optional. FFT size, default is 2048	2048
hop_length	int | None	int, optional. Number of samples between frames. Default is win_length // 4	None
win_length	int | None	int, optional. Window length. Default is n_fft	None
window	str	str, optional. Window function type, default is 'hann'	'hann'
average	str	str, optional. Averaging method, default is 'mean'	'mean'
detrend	str	str, optional. Detrend method, default is 'constant'	'constant'

Raises:

Type	Description
ValueError	If n_fft, win_length, or hop_length are invalid

Source code in wandas/processing/spectral.py

def __init__(
    self,
    sampling_rate: float,
    n_fft: int = 2048,
    hop_length: int | None = None,
    win_length: int | None = None,
    window: str = "hann",
    average: str = "mean",
    detrend: str = "constant",
):
    """
    Initialize Welch operation

    Args:
        sampling_rate: float. Sampling rate (Hz)
        n_fft: int, optional. FFT size, default is 2048
        hop_length: int, optional. Number of samples between frames. Default is win_length // 4
        win_length: int, optional. Window length. Default is n_fft
        window: str, optional. Window function type, default is 'hann'
        average: str, optional. Averaging method, default is 'mean'
        detrend: str, optional. Detrend method, default is 'constant'

    Raises:
        ValueError: If n_fft, win_length, or hop_length are invalid
    """
    # Validate and compute parameters
    actual_win_length, actual_hop_length = _validate_spectral_params(n_fft, win_length, hop_length, "Welch method")

    super().__init__(
        sampling_rate,
        n_fft=n_fft,
        win_length=actual_win_length,
        hop_length=actual_hop_length,
        window=window,
        average=average,
        detrend=detrend,
    )

calculate_output_shape(input_shape)

Calculate output data shape after operation

Parameters:

Name	Type	Description	Default
input_shape	tuple[int, ...]	tuple. Input data shape (channels, samples)	required

Returns:

Name	Type	Description
tuple	tuple[int, ...]	Output data shape (channels, freqs)

Source code in wandas/processing/spectral.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """
    Calculate output data shape after operation

    Args:
        input_shape: tuple. Input data shape (channels, samples)

    Returns:
        tuple: Output data shape (channels, freqs)
    """
    n_freqs = self.n_fft // 2 + 1
    return (*input_shape[:-1], n_freqs)

calculate_output_dtype(input_dtype, *input_dtypes)

Source code in wandas/processing/spectral.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    return _spectral_real_dtype(input_dtype)

NOctSpectrum

Bases: _NOctBase, ChannelIndependentAudioOperation[NDArrayReal, NDArrayReal]

N-octave spectrum operation

Source code in wandas/processing/spectral.py

class NOctSpectrum(_NOctBase, ChannelIndependentAudioOperation[NDArrayReal, NDArrayReal]):
    """N-octave spectrum operation"""

    name = "noct_spectrum"
    _display = "Oct"

    _keeps_input_precision = True

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        """Keep the input precision; MoSQITo computes in float64."""
        return precision_real_dtype(input_dtype)

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Create processor function for octave spectrum"""
        logger.debug(f"Applying NoctSpectrum to array with shape: {x.shape}")
        spec, f_nom = noct_spectrum(
            sig=x.T,
            fs=self.sampling_rate,
            fmin=self.fmin,
            fmax=self.fmax,
            n=self.n,
            G=self.G,
            fr=self.fr,
        )
        # MoSQITo returns one row per exact band, and its nominal labels can be
        # shorter, so the band count comes from the exact center frequencies.
        f_exact, _ = _center_freq(fmin=self.fmin, fmax=self.fmax, n=self.n, G=self.G, fr=self.fr)
        n_bands = noct_band_labels(f_exact, f_nom).size
        spec = np.asarray(spec).reshape(n_bands, x.shape[0]).T
        logger.debug(f"NoctSpectrum applied, returning result with shape: {spec.shape}")
        return np.array(spec)

Attributes

name = 'noct_spectrum' class-attribute instance-attribute

Functions

calculate_output_dtype(input_dtype, *input_dtypes)

Keep the input precision; MoSQITo computes in float64.

Source code in wandas/processing/spectral.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    """Keep the input precision; MoSQITo computes in float64."""
    return precision_real_dtype(input_dtype)

NOctSynthesis

Bases: _NOctBase

N-octave synthesis operation using an explicit original FFT size.

n_fft is required because a one-sided spectrum's bin count cannot distinguish an odd FFT size from the adjacent even size. The value is captured in the operation configuration and is used to construct the canonical real-FFT frequency grid.

The input is Wandas' one-sided peak-amplitude spectrum. MoSQITo sums the squared bins of an RMS spectrum in each band, so every bin except DC and Nyquist is divided by sqrt(2), and every bin by the square root of the analysis window's equivalent noise bandwidth in bins, n_fft * sum(w**2) / sum(w)**2. w is the win_length-sample window that produced the spectrum.

Source code in wandas/processing/spectral.py

class NOctSynthesis(_NOctBase):
    """N-octave synthesis operation using an explicit original FFT size.

    ``n_fft`` is required because a one-sided spectrum's bin count cannot
    distinguish an odd FFT size from the adjacent even size. The value is
    captured in the operation configuration and is used to construct the
    canonical real-FFT frequency grid.

    The input is Wandas' one-sided peak-amplitude spectrum. MoSQITo sums the
    squared bins of an RMS spectrum in each band, so every bin except DC and
    Nyquist is divided by ``sqrt(2)``, and every bin by the square root of the
    analysis window's equivalent noise bandwidth in bins,
    ``n_fft * sum(w**2) / sum(w)**2``. ``w`` is the ``win_length``-sample
    ``window`` that produced the spectrum.
    """

    name = "noct_synthesis"
    _display = "Octs"

    def __init__(
        self,
        sampling_rate: float,
        fmin: float,
        fmax: float,
        n: int = 3,
        G: int = 10,
        fr: int = 1000,
        *,
        n_fft: int,
        window: str = "hann",
        win_length: int | None = None,
    ) -> None:
        """Initialize N-octave synthesis with the source spectrum's FFT size.

        Args:
            sampling_rate: Sampling rate in Hz. The public synthesis Frame
                method requires 48000 Hz.
            fmin: Lower frequency bound in Hz.
            fmax: Upper frequency bound in Hz.
            n: Number of bands per octave.
            G: Exact center-frequency ratio convention, either 2 or 10.
            fr: Reference frequency in Hz.
            n_fft: Positive integer FFT size that produced the complete
                one-sided input spectrum.
            window: Analysis window that produced the input spectrum.
            win_length: Length of that window in samples. Default is ``n_fft``.

        Raises:
            TypeError: If ``n_fft`` is not an integer or ``G`` is not an
                integer ratio convention.
            ValueError: If ``n_fft`` is not positive, ``G`` is not 2 or 10, or
                ``win_length`` is not an integer with ``1 <= win_length <= n_fft``.
        """
        AudioOperation.__init__(
            self,
            sampling_rate,
            fmin=fmin,
            fmax=fmax,
            n=n,
            G=G,
            fr=fr,
            n_fft=n_fft,
            window=window,
            win_length=n_fft if win_length is None else win_length,
        )

    @property
    def n_fft(self) -> int:
        """Return the source FFT size captured by this operation."""
        return int(self._config_value("n_fft"))

    @property
    def window(self) -> str:
        """Analysis window of the source spectrum."""
        return self._config_value("window")

    @property
    def win_length(self) -> int:
        """Analysis window length of the source spectrum, in samples."""
        return int(self._config_value("win_length"))

    def validate_params(self) -> None:
        """Validate common N-octave parameters and the explicit FFT size."""
        super().validate_params()
        value = self._config_value("n_fft")
        if isinstance(value, bool) or not isinstance(value, numbers.Integral):
            raise TypeError(
                "Invalid n_fft for NOctSynthesis\n"
                f"  Got: {value!r} ({type(value).__name__})\n"
                "  Expected: a positive integer\n"
                "Pass the positive integer n_fft stored by SpectralFrame."
            )
        normalized = int(value)
        if normalized <= 0:
            raise ValueError(
                "Invalid n_fft for NOctSynthesis\n"
                f"  Got: {normalized}\n"
                "  Expected: a positive integer\n"
                "Pass the positive integer n_fft stored by SpectralFrame."
            )
        win_length = self._config_value("win_length")
        if (
            isinstance(win_length, bool)
            or not isinstance(win_length, numbers.Integral)
            or not 1 <= int(win_length) <= normalized
        ):
            raise ValueError(
                "Invalid win_length for NOctSynthesis\n"
                f"  Got: {win_length!r}\n"
                f"  Expected: an integer with 1 <= win_length <= n_fft ({normalized})\n"
                "Pass the win_length stored by SpectralFrame."
            )

    def _validate_process_shape(self, data: Any, *inputs: Any) -> None:
        """Reject spectra whose bin count disagrees with the explicit FFT size."""
        del inputs
        expected_bins = self.n_fft // 2 + 1
        actual_bins = data.shape[-1]
        if actual_bins != expected_bins:
            raise ValueError(
                "Invalid frequency bin count for NOctSynthesis\n"
                f"  Got: {actual_bins} bins\n"
                f"  Expected: {expected_bins} bins for n_fft={self.n_fft}\n"
                "Pass the complete one-sided spectrum matching the explicit n_fft."
            )

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        """Count the bands MoSQITo synthesizes.

        MoSQITo drops the top bands whose upper edge ``fc * 2**(1 / (2 * n))``
        exceeds half its sampling rate, ``2 * max(freqs)``. It skips the drop
        when the only such band is the first one.
        """
        shape = super().calculate_output_shape(input_shape)
        f_exact, _ = _center_freq(fmin=self.fmin, fmax=self.fmax, n=self.n, G=self.G, fr=self.fr)
        nyquist = np.fft.rfftfreq(self.n_fft, d=1 / self.sampling_rate).max()
        dropped = np.flatnonzero(np.asarray(f_exact) * 2 ** (1 / (2 * self.n)) > nyquist)
        if dropped.any():
            return (shape[0], shape[1] - dropped.size)
        return shape

    _keeps_input_precision = True

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        """Keep the input precision; MoSQITo computes in float64."""
        return precision_real_dtype(input_dtype)

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Convert the peak-amplitude bins to RMS band contributions, then synthesize."""
        logger.debug(f"Applying NoctSynthesis to array with shape: {x.shape}")
        window_values = scipy.signal.get_window(self.window, self.win_length)
        enbw_bins = self.n_fft * np.sum(window_values**2) / np.sum(window_values) ** 2
        rms = np.abs(x) / np.sqrt(enbw_bins)
        rms[_rfft_positive_frequency_bins(rms.ndim, n_fft=self.n_fft, axis=-1)] /= np.sqrt(2)
        result = self._synthesize(rms, n_fft=self.n_fft)
        logger.debug(f"NoctSynthesis applied, returning result with shape: {result.shape}")
        return np.array(result)

Attributes

name = 'noct_synthesis' class-attribute instance-attribute

n_fft property

Return the source FFT size captured by this operation.

window property

Analysis window of the source spectrum.

win_length property

Analysis window length of the source spectrum, in samples.

Functions

__init__(sampling_rate, fmin, fmax, n=3, G=10, fr=1000, *, n_fft, window='hann', win_length=None)

Initialize N-octave synthesis with the source spectrum's FFT size.

Parameters:

Name	Type	Description	Default
sampling_rate	float	Sampling rate in Hz. The public synthesis Frame method requires 48000 Hz.	required
fmin	float	Lower frequency bound in Hz.	required
fmax	float	Upper frequency bound in Hz.	required
n	int	Number of bands per octave.	3
G	int	Exact center-frequency ratio convention, either 2 or 10.	10
fr	int	Reference frequency in Hz.	1000
n_fft	int	Positive integer FFT size that produced the complete one-sided input spectrum.	required
window	str	Analysis window that produced the input spectrum.	'hann'
win_length	int | None	Length of that window in samples. Default is n_fft.	None

Raises:

Type	Description
TypeError	If n_fft is not an integer or G is not an integer ratio convention.
ValueError	If n_fft is not positive, G is not 2 or 10, or win_length is not an integer with 1 <= win_length <= n_fft.

Source code in wandas/processing/spectral.py

def __init__(
    self,
    sampling_rate: float,
    fmin: float,
    fmax: float,
    n: int = 3,
    G: int = 10,
    fr: int = 1000,
    *,
    n_fft: int,
    window: str = "hann",
    win_length: int | None = None,
) -> None:
    """Initialize N-octave synthesis with the source spectrum's FFT size.

    Args:
        sampling_rate: Sampling rate in Hz. The public synthesis Frame
            method requires 48000 Hz.
        fmin: Lower frequency bound in Hz.
        fmax: Upper frequency bound in Hz.
        n: Number of bands per octave.
        G: Exact center-frequency ratio convention, either 2 or 10.
        fr: Reference frequency in Hz.
        n_fft: Positive integer FFT size that produced the complete
            one-sided input spectrum.
        window: Analysis window that produced the input spectrum.
        win_length: Length of that window in samples. Default is ``n_fft``.

    Raises:
        TypeError: If ``n_fft`` is not an integer or ``G`` is not an
            integer ratio convention.
        ValueError: If ``n_fft`` is not positive, ``G`` is not 2 or 10, or
            ``win_length`` is not an integer with ``1 <= win_length <= n_fft``.
    """
    AudioOperation.__init__(
        self,
        sampling_rate,
        fmin=fmin,
        fmax=fmax,
        n=n,
        G=G,
        fr=fr,
        n_fft=n_fft,
        window=window,
        win_length=n_fft if win_length is None else win_length,
    )

validate_params()

Validate common N-octave parameters and the explicit FFT size.

Source code in wandas/processing/spectral.py

def validate_params(self) -> None:
    """Validate common N-octave parameters and the explicit FFT size."""
    super().validate_params()
    value = self._config_value("n_fft")
    if isinstance(value, bool) or not isinstance(value, numbers.Integral):
        raise TypeError(
            "Invalid n_fft for NOctSynthesis\n"
            f"  Got: {value!r} ({type(value).__name__})\n"
            "  Expected: a positive integer\n"
            "Pass the positive integer n_fft stored by SpectralFrame."
        )
    normalized = int(value)
    if normalized <= 0:
        raise ValueError(
            "Invalid n_fft for NOctSynthesis\n"
            f"  Got: {normalized}\n"
            "  Expected: a positive integer\n"
            "Pass the positive integer n_fft stored by SpectralFrame."
        )
    win_length = self._config_value("win_length")
    if (
        isinstance(win_length, bool)
        or not isinstance(win_length, numbers.Integral)
        or not 1 <= int(win_length) <= normalized
    ):
        raise ValueError(
            "Invalid win_length for NOctSynthesis\n"
            f"  Got: {win_length!r}\n"
            f"  Expected: an integer with 1 <= win_length <= n_fft ({normalized})\n"
            "Pass the win_length stored by SpectralFrame."
        )

calculate_output_shape(input_shape)

Count the bands MoSQITo synthesizes.

MoSQITo drops the top bands whose upper edge fc * 2**(1 / (2 * n)) exceeds half its sampling rate, 2 * max(freqs). It skips the drop when the only such band is the first one.

Source code in wandas/processing/spectral.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """Count the bands MoSQITo synthesizes.

    MoSQITo drops the top bands whose upper edge ``fc * 2**(1 / (2 * n))``
    exceeds half its sampling rate, ``2 * max(freqs)``. It skips the drop
    when the only such band is the first one.
    """
    shape = super().calculate_output_shape(input_shape)
    f_exact, _ = _center_freq(fmin=self.fmin, fmax=self.fmax, n=self.n, G=self.G, fr=self.fr)
    nyquist = np.fft.rfftfreq(self.n_fft, d=1 / self.sampling_rate).max()
    dropped = np.flatnonzero(np.asarray(f_exact) * 2 ** (1 / (2 * self.n)) > nyquist)
    if dropped.any():
        return (shape[0], shape[1] - dropped.size)
    return shape

calculate_output_dtype(input_dtype, *input_dtypes)

Keep the input precision; MoSQITo computes in float64.

Source code in wandas/processing/spectral.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    """Keep the input precision; MoSQITo computes in float64."""
    return precision_real_dtype(input_dtype)

Coherence

Bases: _CrossSpectralBase

Coherence estimation operation

Source code in wandas/processing/spectral.py

class Coherence(_CrossSpectralBase):
    """Coherence estimation operation"""

    name = "coherence"
    _method_label = "Coherence"
    _display = "Coh"

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Processor function for coherence estimation operation"""
        logger.debug(f"Applying coherence estimation to array with shape: {x.shape}")
        # Coherence is scaling-invariant; one CSD matrix supplies both the cross
        # and auto spectra instead of separate Welch passes per pair.
        cross = self._cross_spectral_matrix(x, scaling="density", average="mean")
        result: NDArrayReal = flatten_output_input_pairs(coherence_from_cross_spectra(cross))

        logger.debug(f"Coherence estimation applied, result shape: {result.shape}")
        return result

Attributes

name = 'coherence' class-attribute instance-attribute

CSD

Bases: _ScaledCrossSpectralBase

Cross-spectral density estimation operation

Source code in wandas/processing/spectral.py

class CSD(_ScaledCrossSpectralBase):
    """Cross-spectral density estimation operation"""

    name = "csd"
    _method_label = "CSD"
    _display = "CSD"

    def _process(self, x: NDArrayReal) -> NDArrayComplex:
        """Processor function for cross-spectral density estimation operation"""
        logger.debug(f"Applying CSD estimation to array with shape: {x.shape}")
        cross = self._cross_spectral_matrix(x, scaling=self.scaling, average=self.average)
        result: NDArrayComplex = flatten_output_input_pairs(cross)

        logger.debug(f"CSD estimation applied, result shape: {result.shape}")
        return result

Attributes

name = 'csd' class-attribute instance-attribute

TransferFunction

Bases: _ScaledCrossSpectralBase

Transfer function estimation operation

Source code in wandas/processing/spectral.py

class TransferFunction(_ScaledCrossSpectralBase):
    """Transfer function estimation operation"""

    name = "transfer_function"
    _method_label = "Transfer function"
    _display = "H"

    def _process(self, x: NDArrayReal) -> NDArrayComplex:
        """Processor function for transfer function estimation operation"""
        logger.debug(f"Applying transfer function estimation to array with shape: {x.shape}")
        # The CSD diagonal is the Welch auto-spectrum, so one matrix supplies
        # H[output, input] = P_out_in / P_in_in. Exact zero denominators remain
        # complex NaN; nonzero near-zero bins are untouched.
        cross = self._cross_spectral_matrix(x, scaling=self.scaling, average=self.average)
        h_f = transfer_function_ratio(cross, cross_spectral_auto_power(cross))
        result: NDArrayComplex = flatten_output_input_pairs(h_f)

        logger.debug(f"Transfer function estimation applied, result shape: {result.shape}")
        return result

Attributes

name = 'transfer_function' class-attribute instance-attribute

TransferFunctionFromCSD

Bases: _CrossSpectralDerivation

Canonical H_out_in = P_out_in / P_in_in from an existing CSD matrix.

Source code in wandas/processing/spectral.py

class TransferFunctionFromCSD(_CrossSpectralDerivation):
    """Canonical ``H_out_in = P_out_in / P_in_in`` from an existing CSD matrix."""

    name = "transfer_function_from_csd"
    _display = "H"

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        return np.result_type(input_dtype, np.complex64)

    def _process(self, x: NDArrayComplex) -> NDArrayComplex:
        cross = self._matrix(x)
        result: NDArrayComplex = flatten_output_input_pairs(
            transfer_function_ratio(cross, cross_spectral_auto_power(cross))
        )
        return result

Attributes

name = 'transfer_function_from_csd' class-attribute instance-attribute

Functions

calculate_output_dtype(input_dtype, *input_dtypes)

Source code in wandas/processing/spectral.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    return np.result_type(input_dtype, np.complex64)

CoherenceFromCSD

Bases: _CrossSpectralDerivation

Magnitude-squared coherence from an existing CSD matrix.

Source code in wandas/processing/spectral.py

class CoherenceFromCSD(_CrossSpectralDerivation):
    """Magnitude-squared coherence from an existing CSD matrix."""

    name = "coherence_from_csd"
    _display = "Coh"

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        return _spectral_real_dtype(np.finfo(input_dtype).dtype)

    def _process(self, x: NDArrayComplex) -> NDArrayReal:
        result: NDArrayReal = flatten_output_input_pairs(coherence_from_cross_spectra(self._matrix(x)))
        return result

Attributes

name = 'coherence_from_csd' class-attribute instance-attribute

Functions

calculate_output_dtype(input_dtype, *input_dtypes)

Source code in wandas/processing/spectral.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    return _spectral_real_dtype(np.finfo(input_dtype).dtype)

SpectrogramCSD

Bases: AudioOperation[NDArrayComplex, NDArrayComplex]

Every P_out_in averaged over the frames of an amplitude-normalized STFT.

The input is (channel, frequency, time) as produced by :class:STFT. Undoing its coherent-gain scaling gives SciPy's csd scaling for the same window and segments: scaling="spectrum" divides mean(conj(X_in) * X_out) by two in the doubled bins, and scaling="density" also multiplies by sum(w)**2 / (fs * sum(w**2)). Frames are averaged with the mean and are not detrended.

Source code in wandas/processing/spectral.py

class SpectrogramCSD(AudioOperation[NDArrayComplex, NDArrayComplex]):
    """Every ``P_out_in`` averaged over the frames of an amplitude-normalized STFT.

    The input is ``(channel, frequency, time)`` as produced by :class:`STFT`.
    Undoing its coherent-gain scaling gives SciPy's ``csd`` scaling for the same
    window and segments: ``scaling="spectrum"`` divides
    ``mean(conj(X_in) * X_out)`` by two in the doubled bins, and
    ``scaling="density"`` also multiplies by ``sum(w)**2 / (fs * sum(w**2))``.
    Frames are averaged with the mean and are not detrended.
    """

    name = "spectrogram_csd"
    _display = "CSD"

    def __init__(
        self,
        sampling_rate: float,
        n_fft: int = 2048,
        win_length: int | None = None,
        window: str = "hann",
        scaling: str = "spectrum",
    ):
        # The CSD averages existing frames, so any hop the STFT accepted is valid here.
        resolved_win_length = n_fft if win_length is None else win_length
        actual_win_length, _ = _validate_spectral_params(n_fft, win_length, resolved_win_length, "Spectrogram CSD")
        if scaling not in ("spectrum", "density"):
            raise ValueError(
                "Invalid scaling for spectrogram CSD\n"
                f"  Got: {scaling!r}\n"
                "  Expected: 'spectrum' or 'density'\n"
                "Use 'spectrum' for power per bin or 'density' for power per hertz."
            )
        super().__init__(
            sampling_rate,
            n_fft=n_fft,
            win_length=actual_win_length,
            window=window,
            scaling=scaling,
        )

    @property
    def n_fft(self) -> int:
        """FFT size captured at operation construction time."""
        return self._config_value("n_fft")

    @property
    def win_length(self) -> int:
        """Window length captured at operation construction time."""
        return self._config_value("win_length")

    @property
    def window(self) -> str:
        """Window name captured at operation construction time."""
        return self._config_value("window")

    @property
    def scaling(self) -> str:
        """CSD scaling captured at operation construction time."""
        return self._config_value("scaling")

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        expected_frequency_bins = self.n_fft // 2 + 1
        if len(input_shape) != 3 or int(input_shape[-2]) != expected_frequency_bins or input_shape[-1] == 0:
            raise ValueError(
                "Invalid spectrogram for CSD\n"
                f"  Got: shape {input_shape}\n"
                f"  Expected: (channel, {expected_frequency_bins}, time) with at least one time frame "
                f"for n_fft={self.n_fft}\n"
                "Select a non-empty frame range of the complete STFT before computing the CSD."
            )
        n_channels = input_shape[0]
        return (n_channels * n_channels, input_shape[1])

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        if not np.issubdtype(input_dtype, np.complexfloating):
            raise ValueError(
                "Invalid spectrogram for CSD\n"
                f"  Got: {input_dtype} values\n"
                "  Expected: complex STFT values with phase\n"
                "Compute the CSD from the STFT before taking its magnitude."
            )
        return np.result_type(input_dtype, np.complex64)

    def _process(self, x: NDArrayComplex) -> NDArrayComplex:
        cross = np.einsum("ift,oft->oif", np.conj(x), x) / x.shape[-1]
        cross = _denormalize_rfft_amplitude(cross, n_fft=self.n_fft, window_gain=1.0)
        if self.scaling == "density":
            win = scipy.signal.get_window(self.window, self.win_length)
            cross *= np.sum(win) ** 2 / (self.sampling_rate * np.sum(win**2))
        result: NDArrayComplex = flatten_output_input_pairs(cross)
        return result

Attributes

name = 'spectrogram_csd' class-attribute instance-attribute

n_fft property

FFT size captured at operation construction time.

win_length property

Window length captured at operation construction time.

window property

Window name captured at operation construction time.

scaling property

CSD scaling captured at operation construction time.

Functions

__init__(sampling_rate, n_fft=2048, win_length=None, window='hann', scaling='spectrum')

Source code in wandas/processing/spectral.py

def __init__(
    self,
    sampling_rate: float,
    n_fft: int = 2048,
    win_length: int | None = None,
    window: str = "hann",
    scaling: str = "spectrum",
):
    # The CSD averages existing frames, so any hop the STFT accepted is valid here.
    resolved_win_length = n_fft if win_length is None else win_length
    actual_win_length, _ = _validate_spectral_params(n_fft, win_length, resolved_win_length, "Spectrogram CSD")
    if scaling not in ("spectrum", "density"):
        raise ValueError(
            "Invalid scaling for spectrogram CSD\n"
            f"  Got: {scaling!r}\n"
            "  Expected: 'spectrum' or 'density'\n"
            "Use 'spectrum' for power per bin or 'density' for power per hertz."
        )
    super().__init__(
        sampling_rate,
        n_fft=n_fft,
        win_length=actual_win_length,
        window=window,
        scaling=scaling,
    )

calculate_output_shape(input_shape)

Source code in wandas/processing/spectral.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    expected_frequency_bins = self.n_fft // 2 + 1
    if len(input_shape) != 3 or int(input_shape[-2]) != expected_frequency_bins or input_shape[-1] == 0:
        raise ValueError(
            "Invalid spectrogram for CSD\n"
            f"  Got: shape {input_shape}\n"
            f"  Expected: (channel, {expected_frequency_bins}, time) with at least one time frame "
            f"for n_fft={self.n_fft}\n"
            "Select a non-empty frame range of the complete STFT before computing the CSD."
        )
    n_channels = input_shape[0]
    return (n_channels * n_channels, input_shape[1])

calculate_output_dtype(input_dtype, *input_dtypes)

Source code in wandas/processing/spectral.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    if not np.issubdtype(input_dtype, np.complexfloating):
        raise ValueError(
            "Invalid spectrogram for CSD\n"
            f"  Got: {input_dtype} values\n"
            "  Expected: complex STFT values with phase\n"
            "Compute the CSD from the STFT before taking its magnitude."
        )
    return np.result_type(input_dtype, np.complex64)

Functions

noct_spectrum(*args, **kwargs)

Source code in wandas/processing/spectral.py

def noct_spectrum(*args: Any, **kwargs: Any) -> Any:
    return require_mosqito_sound_level_meter("noct_spectrum").noct_spectrum(*args, **kwargs)

noct_synthesis(*args, **kwargs)

Source code in wandas/processing/spectral.py

def noct_synthesis(*args: Any, **kwargs: Any) -> Any:
    return require_mosqito_sound_level_meter("noct_synthesis").noct_synthesis(*args, **kwargs)

noct_band_labels(f_exact, f_nom)

Return one center-frequency label per band that MoSQITo computes.

MoSQITo computes a band for every exact center frequency between fmin and fmax, but its nominal labels cover only its fixed tables (25 Hz to 20 kHz for 1/3 octave, 31.5 Hz to 16 kHz for octave). A band with a nominal label keeps it; a band outside the table is labelled with its exact center frequency.

Parameters:

Name	Type	Description	Default
f_exact	Any	Exact center frequencies returned by MoSQITo.	required
f_nom	Any	Nominal center frequencies returned by MoSQITo, a contiguous run of the bands in f_exact.	required

Returns:

Type	Description
NDArrayReal	Labels with the same length as f_exact.

Source code in wandas/processing/spectral.py

def noct_band_labels(f_exact: Any, f_nom: Any) -> NDArrayReal:
    """Return one center-frequency label per band that MoSQITo computes.

    MoSQITo computes a band for every exact center frequency between ``fmin``
    and ``fmax``, but its nominal labels cover only its fixed tables (25 Hz to
    20 kHz for 1/3 octave, 31.5 Hz to 16 kHz for octave). A band with a
    nominal label keeps it; a band outside the table is labelled with its
    exact center frequency.

    Args:
        f_exact: Exact center frequencies returned by MoSQITo.
        f_nom: Nominal center frequencies returned by MoSQITo, a contiguous
            run of the bands in ``f_exact``.

    Returns:
        Labels with the same length as ``f_exact``.
    """
    nominal = np.asarray(f_nom)
    exact = np.asarray(f_exact, dtype=np.float64)
    if nominal.size == exact.size:
        return nominal
    labels = exact.copy()
    if nominal.size:
        # Nominal values are within about 1 % of their exact band, while
        # adjacent bands are at least a third of an octave apart.
        start = int(np.argmin(np.abs(np.log(exact / float(nominal[0])))))
        labels[start : start + nominal.size] = nominal
    return labels

validate_noct_recipe_params(params)

Validate portable N-octave parameters without importing MoSQITo.

Source code in wandas/processing/spectral.py

def validate_noct_recipe_params(params: Mapping[str, Any]) -> None:
    """Validate portable N-octave parameters without importing MoSQITo."""
    if "G" in params:
        _validate_noct_g(params["G"])

validate_stft_recipe_params(params)

Validate sparse opt-in before constructing a portable STFT graph.

Source code in wandas/processing/spectral.py

def validate_stft_recipe_params(params: Mapping[str, Any]) -> None:
    """Validate sparse opt-in before constructing a portable STFT graph."""
    unknown = set(params) - {"n_fft", "hop_length", "win_length", "window", "allow_sparse", "detrend"}
    if unknown:
        raise ValueError(f"Unsupported STFT Recipe parameters: {sorted(unknown)}")
    if type(params.get("allow_sparse", False)) is not bool:
        raise TypeError("allow_sparse must be a bool")
    detrend = params.get("detrend")
    if detrend is not None and detrend not in ("constant", "linear"):
        raise ValueError(
            "Invalid detrend for STFT\n"
            f"  Got: {detrend!r}\n"
            "  Expected: None, 'constant' or 'linear'\n"
            "Use 'constant' to remove each frame's mean or 'linear' to remove its linear trend."
        )

wandas.processing.cepstral

Real-cepstrum analysis, liftering, and spectral-envelope reconstruction.

Attributes

logger = logging.getLogger(__name__) module-attribute

DEFAULT_LOG_FLOOR = 1e-12 module-attribute

__all__ = ['DEFAULT_LOG_FLOOR', 'Cepstrum', 'Lifter', 'SpectralEnvelope', 'SpectrogramCepstrum'] module-attribute

Classes

Cepstrum

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Calculate a normalized real cepstrum.

The operation windows the first min(len(x), n_fft) samples of each channel, zero-pads them to n_fft, calculates the one-sided FFT using the same amplitude normalization as :class:wandas.processing.spectral.FFT, applies a positive floor, and returns irfft(log(magnitude)). Processing is lazy when called through :meth:AudioOperation.process.

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate in Hz.	required
n_fft	int | None	int, optional. FFT size. None uses the input sample count. A smaller value truncates the input and a larger value zero-pads it.	None
window	str	str, default="hann". SciPy window name applied before the FFT.	'hann'
floor	float	float, default=1e-12. Positive finite floor applied to normalized magnitudes before log.	DEFAULT_LOG_FLOOR

Raises:

Type	Description
TypeError	If n_fft is not an integer or window is not a non-empty string.
ValueError	If n_fft or floor is not positive and finite.

Source code in wandas/processing/cepstral.py

class Cepstrum(AudioOperation[NDArrayReal, NDArrayReal]):
    """Calculate a normalized real cepstrum.

    The operation windows the first ``min(len(x), n_fft)`` samples of each
    channel, zero-pads them to ``n_fft``, calculates the one-sided FFT using the
    same amplitude normalization as :class:`wandas.processing.spectral.FFT`,
    applies a positive floor, and returns ``irfft(log(magnitude))``. Processing
    is lazy when called through :meth:`AudioOperation.process`.

    Args:
        sampling_rate: float. Sampling rate in Hz.
        n_fft: int, optional. FFT size. ``None`` uses the input sample count. A smaller value truncates
            the input and a larger value zero-pads it.
        window: str, default="hann". SciPy window name applied before the FFT.
        floor: float, default=1e-12. Positive finite floor applied to normalized magnitudes before ``log``.

    Raises:
        TypeError: If ``n_fft`` is not an integer or ``window`` is not a non-empty string.
        ValueError: If ``n_fft`` or ``floor`` is not positive and finite.
    """

    name = "cepstrum"
    _display = "cepstrum"

    def __init__(
        self,
        sampling_rate: float,
        n_fft: int | None = None,
        window: str = "hann",
        floor: float = DEFAULT_LOG_FLOOR,
    ) -> None:
        if n_fft is not None and (isinstance(n_fft, bool) or not isinstance(n_fft, numbers.Integral)):
            raise TypeError(
                "Invalid FFT size for cepstrum\n"
                f"  Got: {type(n_fft).__name__}\n"
                "  Expected: a positive integer or None\n"
                "Pass an integer FFT size, or omit n_fft to use the input length."
            )
        normalized_n_fft = None if n_fft is None else int(n_fft)
        if normalized_n_fft is not None and normalized_n_fft <= 0:
            raise ValueError(
                "Invalid FFT size for cepstrum\n"
                f"  Got: {normalized_n_fft}\n"
                "  Expected: a positive integer\n"
                "Use n_fft=None to match the input length automatically."
            )
        if not isinstance(window, str) or not window:
            raise TypeError(
                "Invalid window for cepstrum\n"
                f"  Got: {window!r}\n"
                "  Expected: a non-empty SciPy window name\n"
                "Use a name such as 'hann' or 'boxcar'."
            )
        normalized_floor = _normalize_log_floor(floor, analysis_name="cepstrum")
        super().__init__(
            sampling_rate,
            n_fft=normalized_n_fft,
            window=window,
            floor=normalized_floor,
        )

    @property
    def n_fft(self) -> int | None:
        """Return the configured FFT size, or ``None`` for input length."""
        return self._config_value("n_fft")

    @property
    def window(self) -> str:
        """Return the configured analysis-window name."""
        return self._config_value("window")

    @property
    def floor(self) -> float:
        """Return the positive log-magnitude floor."""
        return self._config_value("floor")

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        """Return ``(..., n_fft)`` without evaluating input data."""
        n_fft = _resolve_fft_size(self.n_fft, int(input_shape[-1]))
        return (*input_shape[:-1], n_fft)

    _keeps_input_precision = True

    def calculate_output_dtype(
        self,
        input_dtype: np.dtype[Any],
        *input_dtypes: np.dtype[Any],
    ) -> np.dtype[Any]:
        """Return the real dtype matching the input precision."""
        return precision_real_dtype(input_dtype)

    def _process(self, data: NDArrayReal) -> NDArrayReal:
        """Calculate the eager real-cepstrum kernel for delayed execution."""
        if np.iscomplexobj(data):
            raise TypeError(
                "Cepstrum analysis requires real-valued input\n"
                f"  Got: {np.asarray(data).dtype}\n"
                "  Expected: real time-domain samples\n"
                "Use ChannelFrame time-domain data as the input."
            )
        n_fft = _resolve_fft_size(self.n_fft, int(data.shape[-1]))
        analysis = np.asarray(data[..., :n_fft], dtype=np.float64)
        window_values = scipy.signal.get_window(self.window, analysis.shape[-1])
        window_gain = float(np.sum(window_values))
        if not np.isfinite(window_gain) or window_gain == 0:
            raise ValueError(
                "Invalid window gain for cepstrum\n"
                f"  Window: {self.window!r}\n"
                f"  Gain: {window_gain}\n"
                "Use a window with a finite non-zero coherent gain."
            )
        spectrum = np.fft.rfft(analysis * window_values, n=n_fft, axis=-1)
        normalized_spectrum = _normalize_rfft_amplitude(
            spectrum,
            n_fft=n_fft,
            window_gain=window_gain,
        )
        magnitude = np.abs(normalized_spectrum)
        log_magnitude = np.log(np.maximum(magnitude, self.floor))
        return np.asarray(np.fft.irfft(log_magnitude, n=n_fft, axis=-1), dtype=np.float64)

Attributes

name = 'cepstrum' class-attribute instance-attribute

n_fft property

Return the configured FFT size, or None for input length.

window property

Return the configured analysis-window name.

floor property

Return the positive log-magnitude floor.

Functions

__init__(sampling_rate, n_fft=None, window='hann', floor=DEFAULT_LOG_FLOOR)

Source code in wandas/processing/cepstral.py

def __init__(
    self,
    sampling_rate: float,
    n_fft: int | None = None,
    window: str = "hann",
    floor: float = DEFAULT_LOG_FLOOR,
) -> None:
    if n_fft is not None and (isinstance(n_fft, bool) or not isinstance(n_fft, numbers.Integral)):
        raise TypeError(
            "Invalid FFT size for cepstrum\n"
            f"  Got: {type(n_fft).__name__}\n"
            "  Expected: a positive integer or None\n"
            "Pass an integer FFT size, or omit n_fft to use the input length."
        )
    normalized_n_fft = None if n_fft is None else int(n_fft)
    if normalized_n_fft is not None and normalized_n_fft <= 0:
        raise ValueError(
            "Invalid FFT size for cepstrum\n"
            f"  Got: {normalized_n_fft}\n"
            "  Expected: a positive integer\n"
            "Use n_fft=None to match the input length automatically."
        )
    if not isinstance(window, str) or not window:
        raise TypeError(
            "Invalid window for cepstrum\n"
            f"  Got: {window!r}\n"
            "  Expected: a non-empty SciPy window name\n"
            "Use a name such as 'hann' or 'boxcar'."
        )
    normalized_floor = _normalize_log_floor(floor, analysis_name="cepstrum")
    super().__init__(
        sampling_rate,
        n_fft=normalized_n_fft,
        window=window,
        floor=normalized_floor,
    )

calculate_output_shape(input_shape)

Return (..., n_fft) without evaluating input data.

Source code in wandas/processing/cepstral.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """Return ``(..., n_fft)`` without evaluating input data."""
    n_fft = _resolve_fft_size(self.n_fft, int(input_shape[-1]))
    return (*input_shape[:-1], n_fft)

calculate_output_dtype(input_dtype, *input_dtypes)

Return the real dtype matching the input precision.

Source code in wandas/processing/cepstral.py

def calculate_output_dtype(
    self,
    input_dtype: np.dtype[Any],
    *input_dtypes: np.dtype[Any],
) -> np.dtype[Any]:
    """Return the real dtype matching the input precision."""
    return precision_real_dtype(input_dtype)

SpectrogramCepstrum

Bases: AudioOperation[NDArrayComplex, NDArrayReal]

Calculate a real cepstrum independently at every STFT time frame.

Input data is a normalized one-sided spectrum shaped (channel, frequency, time). The operation discards phase, applies a positive log floor, and performs irfft along the frequency axis. The result is shaped (channel, quefrency, time).

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate in Hz.	required
n_fft	int	int. FFT size used to create the input spectrogram.	required
floor	float	float, default=1e-12. Positive finite floor applied to magnitude before log.	DEFAULT_LOG_FLOOR

Raises:

Type	Description
TypeError	If n_fft is not an integer or floor is not real.
ValueError	If n_fft or floor is not positive, or input shape disagrees with the FFT size.

Source code in wandas/processing/cepstral.py

class SpectrogramCepstrum(AudioOperation[NDArrayComplex, NDArrayReal]):
    """Calculate a real cepstrum independently at every STFT time frame.

    Input data is a normalized one-sided spectrum shaped
    ``(channel, frequency, time)``. The operation discards phase, applies a
    positive log floor, and performs ``irfft`` along the frequency axis. The
    result is shaped ``(channel, quefrency, time)``.

    Args:
        sampling_rate: float. Sampling rate in Hz.
        n_fft: int. FFT size used to create the input spectrogram.
        floor: float, default=1e-12. Positive finite floor applied to magnitude before ``log``.

    Raises:
        TypeError: If ``n_fft`` is not an integer or ``floor`` is not real.
        ValueError: If ``n_fft`` or ``floor`` is not positive, or input shape disagrees
            with the FFT size.
    """

    name = "spectrogram_cepstrum"
    _display = "cepstrum"

    def __init__(
        self,
        sampling_rate: float,
        n_fft: int,
        floor: float = DEFAULT_LOG_FLOOR,
    ) -> None:
        if isinstance(n_fft, bool) or not isinstance(n_fft, numbers.Integral):
            raise TypeError(
                "Invalid FFT size for spectrogram cepstrum\n"
                f"  Got: {type(n_fft).__name__}\n"
                "  Expected: a positive integer\n"
                "Pass the FFT size used to create the spectrogram."
            )
        normalized_n_fft = int(n_fft)
        if normalized_n_fft <= 0:
            raise ValueError(
                "Invalid FFT size for spectrogram cepstrum\n"
                f"  Got: {normalized_n_fft}\n"
                "  Expected: a positive integer\n"
                "Pass the FFT size used to create the spectrogram."
            )
        normalized_floor = _normalize_log_floor(
            floor,
            analysis_name="spectrogram cepstrum",
        )
        super().__init__(
            sampling_rate,
            n_fft=normalized_n_fft,
            floor=normalized_floor,
        )

    @property
    def n_fft(self) -> int:
        """Return the FFT size of the input spectrogram."""
        return self._config_value("n_fft")

    @property
    def floor(self) -> float:
        """Return the positive log-magnitude floor."""
        return self._config_value("floor")

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        """Replace the frequency axis with a complete quefrency axis."""
        expected_frequency_bins = self.n_fft // 2 + 1
        if len(input_shape) != 3 or int(input_shape[-2]) != expected_frequency_bins:
            raise ValueError(
                "Invalid spectrogram shape for cepstrum\n"
                f"  Got: {input_shape}\n"
                f"  Expected: (channels, {expected_frequency_bins}, time) for n_fft={self.n_fft}\n"
                "Use the n_fft stored by the source SpectrogramFrame."
            )
        return (int(input_shape[0]), self.n_fft, int(input_shape[-1]))

    _keeps_input_precision = True

    def calculate_output_dtype(
        self,
        input_dtype: np.dtype[Any],
        *input_dtypes: np.dtype[Any],
    ) -> np.dtype[Any]:
        """Return the real dtype matching the input precision."""
        return precision_real_dtype(input_dtype)

    def _process(self, data: NDArrayComplex) -> NDArrayReal:
        """Calculate the eager framewise real-cepstrum kernel."""
        self.calculate_output_shape(np.asarray(data).shape)
        magnitude = np.abs(np.asarray(data))
        log_magnitude = np.log(np.maximum(magnitude, self.floor))
        result = np.fft.irfft(log_magnitude, n=self.n_fft, axis=-2)
        return np.asarray(result, dtype=np.float64)

Attributes

name = 'spectrogram_cepstrum' class-attribute instance-attribute

n_fft property

Return the FFT size of the input spectrogram.

floor property

Return the positive log-magnitude floor.

Functions

__init__(sampling_rate, n_fft, floor=DEFAULT_LOG_FLOOR)

Source code in wandas/processing/cepstral.py

def __init__(
    self,
    sampling_rate: float,
    n_fft: int,
    floor: float = DEFAULT_LOG_FLOOR,
) -> None:
    if isinstance(n_fft, bool) or not isinstance(n_fft, numbers.Integral):
        raise TypeError(
            "Invalid FFT size for spectrogram cepstrum\n"
            f"  Got: {type(n_fft).__name__}\n"
            "  Expected: a positive integer\n"
            "Pass the FFT size used to create the spectrogram."
        )
    normalized_n_fft = int(n_fft)
    if normalized_n_fft <= 0:
        raise ValueError(
            "Invalid FFT size for spectrogram cepstrum\n"
            f"  Got: {normalized_n_fft}\n"
            "  Expected: a positive integer\n"
            "Pass the FFT size used to create the spectrogram."
        )
    normalized_floor = _normalize_log_floor(
        floor,
        analysis_name="spectrogram cepstrum",
    )
    super().__init__(
        sampling_rate,
        n_fft=normalized_n_fft,
        floor=normalized_floor,
    )

calculate_output_shape(input_shape)

Replace the frequency axis with a complete quefrency axis.

Source code in wandas/processing/cepstral.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """Replace the frequency axis with a complete quefrency axis."""
    expected_frequency_bins = self.n_fft // 2 + 1
    if len(input_shape) != 3 or int(input_shape[-2]) != expected_frequency_bins:
        raise ValueError(
            "Invalid spectrogram shape for cepstrum\n"
            f"  Got: {input_shape}\n"
            f"  Expected: (channels, {expected_frequency_bins}, time) for n_fft={self.n_fft}\n"
            "Use the n_fft stored by the source SpectrogramFrame."
        )
    return (int(input_shape[0]), self.n_fft, int(input_shape[-1]))

calculate_output_dtype(input_dtype, *input_dtypes)

Return the real dtype matching the input precision.

Source code in wandas/processing/cepstral.py

def calculate_output_dtype(
    self,
    input_dtype: np.dtype[Any],
    *input_dtypes: np.dtype[Any],
) -> np.dtype[Any]:
    """Return the real dtype matching the input precision."""
    return precision_real_dtype(input_dtype)

Lifter

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Keep low- or high-quefrency real-cepstrum coefficients.

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate in Hz; its reciprocal is the quefrency-bin spacing.	required
cutoff	float	float. Positive quefrency boundary in seconds. The represented bin and its circularly mirrored negative-quefrency bins are included in low mode.	required
mode	Literal['low', 'high']	{"low", "high"}, default="low". "low" keeps the smooth spectral-envelope region. "high" keeps the complementary fine structure.	'low'
axis	int	int, default=-1. Non-channel quefrency axis. CepstrogramFrame uses -2.	-1

Raises:

Type	Description
TypeError	If cutoff is not a real number or axis is not an integer.
ValueError	If the cutoff is non-positive, non-finite, smaller than one bin, or overlaps the mirrored half of the concrete cepstrum; or if mode is unknown or axis does not identify a non-channel input axis.

Source code in wandas/processing/cepstral.py

class Lifter(AudioOperation[NDArrayReal, NDArrayReal]):
    """Keep low- or high-quefrency real-cepstrum coefficients.

    Args:
        sampling_rate: float. Sampling rate in Hz; its reciprocal is the quefrency-bin spacing.
        cutoff: float. Positive quefrency boundary in seconds. The represented bin and its
            circularly mirrored negative-quefrency bins are included in low mode.
        mode: {"low", "high"}, default="low". ``"low"`` keeps the smooth spectral-envelope region. ``"high"`` keeps
            the complementary fine structure.
        axis: int, default=-1. Non-channel quefrency axis. ``CepstrogramFrame`` uses ``-2``.

    Raises:
        TypeError: If ``cutoff`` is not a real number or ``axis`` is not an integer.
        ValueError: If the cutoff is non-positive, non-finite, smaller than one bin, or
            overlaps the mirrored half of the concrete cepstrum; or if ``mode`` is
            unknown or ``axis`` does not identify a non-channel input axis.
    """

    name = "lifter"
    _display = "lifter"

    def __init__(
        self,
        sampling_rate: float,
        cutoff: float,
        mode: Literal["low", "high"] = "low",
        *,
        axis: int = -1,
    ) -> None:
        if isinstance(cutoff, bool) or not isinstance(cutoff, numbers.Real):
            raise TypeError(
                "Invalid lifter cutoff\n"
                f"  Got: {type(cutoff).__name__}\n"
                "  Expected: a positive finite duration in seconds\n"
                "Pass a real quefrency boundary such as 0.002."
            )
        normalized_cutoff = float(cutoff)
        if not np.isfinite(normalized_cutoff) or normalized_cutoff <= 0:
            raise ValueError(
                "Invalid lifter cutoff\n"
                f"  Got: {normalized_cutoff}\n"
                "  Expected: a positive finite duration in seconds\n"
                "Choose a small positive quefrency boundary."
            )
        if mode not in {"low", "high"}:
            raise ValueError(
                "Invalid lifter mode\n"
                f"  Got: {mode!r}\n"
                "  Expected: 'low' or 'high'\n"
                "Use 'low' for the envelope or 'high' for fine structure."
            )
        normalized_axis = _normalize_transform_axis(axis, operation_name="lifter")
        super().__init__(
            sampling_rate,
            cutoff=normalized_cutoff,
            mode=mode,
            axis=normalized_axis,
        )

    @property
    def cutoff(self) -> float:
        """Return the quefrency cutoff in seconds."""
        return self._config_value("cutoff")

    @property
    def mode(self) -> Literal["low", "high"]:
        """Return the selected low- or high-quefrency mode."""
        return self._config_value("mode")

    @property
    def axis(self) -> int:
        """Return the configured quefrency axis."""
        return self._config_value("axis")

    def calculate_output_dtype(
        self,
        input_dtype: np.dtype[Any],
        *input_dtypes: np.dtype[Any],
    ) -> np.dtype[Any]:
        """Preserve a real floating input dtype."""
        return np.dtype(np.result_type(input_dtype, np.float32))

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        """Validate the cutoff against the known cepstrum length."""
        axis = _resolve_transform_axis(self.axis, len(input_shape), operation_name="lifter")
        self._resolve_cutoff_bins(int(input_shape[axis]))
        return input_shape

    def _resolve_cutoff_bins(self, coefficient_count: int) -> int:
        """Return the represented cutoff after validating mirrored regions."""
        cutoff_bins = int(np.floor(self.cutoff * self.sampling_rate))
        if cutoff_bins < 1:
            raise ValueError(
                "Invalid lifter cutoff for this sampling rate\n"
                f"  Got: {self.cutoff} seconds\n"
                f"  Expected: at least one bin ({1 / self.sampling_rate:g} seconds)\n"
                "Increase cutoff so it reaches a represented quefrency bin."
            )
        if 2 * cutoff_bins >= coefficient_count:
            maximum_cutoff_bins = (coefficient_count - 1) // 2
            raise ValueError(
                "Invalid lifter cutoff for this cepstrum length\n"
                f"  Got: {self.cutoff} seconds ({cutoff_bins} bins)\n"
                f"  Expected: at most {maximum_cutoff_bins} non-overlapping bins\n"
                "Choose a smaller cutoff so mirrored regions do not overlap."
            )
        return cutoff_bins

    def _process(self, data: NDArrayReal) -> NDArrayReal:
        """Apply the eager symmetric lifter mask for delayed execution."""
        if np.iscomplexobj(data):
            raise TypeError("Lifter requires real-valued cepstral coefficients.")
        coefficients = np.asarray(data)
        axis = _resolve_transform_axis(self.axis, coefficients.ndim, operation_name="lifter")
        coefficient_count = int(coefficients.shape[axis])
        cutoff_bins = self._resolve_cutoff_bins(coefficient_count)
        keep = np.zeros(coefficient_count, dtype=bool)
        keep[: cutoff_bins + 1] = True
        keep[-cutoff_bins:] = True
        if self.mode == "high":
            keep = ~keep
        mask_shape = [1] * coefficients.ndim
        mask_shape[axis] = coefficient_count
        result = np.where(keep.reshape(mask_shape), coefficients, 0)
        return np.asarray(result, dtype=self.calculate_output_dtype(coefficients.dtype))

Attributes

name = 'lifter' class-attribute instance-attribute

cutoff property

Return the quefrency cutoff in seconds.

mode property

Return the selected low- or high-quefrency mode.

axis property

Return the configured quefrency axis.

Functions

__init__(sampling_rate, cutoff, mode='low', *, axis=-1)

Source code in wandas/processing/cepstral.py

def __init__(
    self,
    sampling_rate: float,
    cutoff: float,
    mode: Literal["low", "high"] = "low",
    *,
    axis: int = -1,
) -> None:
    if isinstance(cutoff, bool) or not isinstance(cutoff, numbers.Real):
        raise TypeError(
            "Invalid lifter cutoff\n"
            f"  Got: {type(cutoff).__name__}\n"
            "  Expected: a positive finite duration in seconds\n"
            "Pass a real quefrency boundary such as 0.002."
        )
    normalized_cutoff = float(cutoff)
    if not np.isfinite(normalized_cutoff) or normalized_cutoff <= 0:
        raise ValueError(
            "Invalid lifter cutoff\n"
            f"  Got: {normalized_cutoff}\n"
            "  Expected: a positive finite duration in seconds\n"
            "Choose a small positive quefrency boundary."
        )
    if mode not in {"low", "high"}:
        raise ValueError(
            "Invalid lifter mode\n"
            f"  Got: {mode!r}\n"
            "  Expected: 'low' or 'high'\n"
            "Use 'low' for the envelope or 'high' for fine structure."
        )
    normalized_axis = _normalize_transform_axis(axis, operation_name="lifter")
    super().__init__(
        sampling_rate,
        cutoff=normalized_cutoff,
        mode=mode,
        axis=normalized_axis,
    )

calculate_output_dtype(input_dtype, *input_dtypes)

Preserve a real floating input dtype.

Source code in wandas/processing/cepstral.py

def calculate_output_dtype(
    self,
    input_dtype: np.dtype[Any],
    *input_dtypes: np.dtype[Any],
) -> np.dtype[Any]:
    """Preserve a real floating input dtype."""
    return np.dtype(np.result_type(input_dtype, np.float32))

calculate_output_shape(input_shape)

Validate the cutoff against the known cepstrum length.

Source code in wandas/processing/cepstral.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """Validate the cutoff against the known cepstrum length."""
    axis = _resolve_transform_axis(self.axis, len(input_shape), operation_name="lifter")
    self._resolve_cutoff_bins(int(input_shape[axis]))
    return input_shape

SpectralEnvelope

Bases: AudioOperation[NDArrayReal, NDArrayComplex]

Reconstruct a normalized one-sided spectral envelope.

The input must be a complete, circularly symmetric real cepstrum. The operation calculates exp(real(rfft(cepstrum))) and returns complex data with zero phase so it can be represented by SpectralFrame. Processing is lazy through :meth:AudioOperation.process.

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate in Hz.	required
axis	int	int, default=-1. Non-channel quefrency axis. CepstrogramFrame uses -2.	-1

Raises:

Type	Description
TypeError	If axis is not an integer or concrete input is complex-valued.
ValueError	If axis is invalid or concrete coefficients are not circularly symmetric.

Source code in wandas/processing/cepstral.py

class SpectralEnvelope(AudioOperation[NDArrayReal, NDArrayComplex]):
    """Reconstruct a normalized one-sided spectral envelope.

    The input must be a complete, circularly symmetric real cepstrum. The
    operation calculates ``exp(real(rfft(cepstrum)))`` and returns complex data
    with zero phase so it can be represented by ``SpectralFrame``. Processing is
    lazy through :meth:`AudioOperation.process`.

    Args:
        sampling_rate: float. Sampling rate in Hz.
        axis: int, default=-1. Non-channel quefrency axis. ``CepstrogramFrame`` uses ``-2``.

    Raises:
        TypeError: If ``axis`` is not an integer or concrete input is complex-valued.
        ValueError: If ``axis`` is invalid or concrete coefficients are not circularly
            symmetric.
    """

    name = "spectral_envelope"
    _display = "spectral envelope"

    def __init__(self, sampling_rate: float, *, axis: int = -1) -> None:
        normalized_axis = _normalize_transform_axis(
            axis,
            operation_name="spectral envelope",
        )
        super().__init__(sampling_rate, axis=normalized_axis)

    @property
    def axis(self) -> int:
        """Return the configured quefrency axis."""
        return self._config_value("axis")

    def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
        """Replace the quefrency axis with its one-sided frequency axis."""
        axis = _resolve_transform_axis(
            self.axis,
            len(input_shape),
            operation_name="spectral envelope",
        )
        output_shape = list(input_shape)
        output_shape[axis] = int(input_shape[axis]) // 2 + 1
        return tuple(output_shape)

    _keeps_input_precision = True

    def calculate_output_dtype(
        self,
        input_dtype: np.dtype[Any],
        *input_dtypes: np.dtype[Any],
    ) -> np.dtype[Any]:
        """Return the complex dtype matching the input precision."""
        return precision_complex_dtype(input_dtype)

    def _process(self, data: NDArrayReal) -> NDArrayComplex:
        """Calculate the eager spectral-envelope kernel for delayed execution."""
        if np.iscomplexobj(data):
            raise TypeError("SpectralEnvelope requires real-valued cepstral coefficients.")
        coefficients = np.asarray(data, dtype=np.float64)
        axis = _resolve_transform_axis(
            self.axis,
            coefficients.ndim,
            operation_name="spectral envelope",
        )
        transformed = np.moveaxis(coefficients, axis, -1)
        tolerance = 64 * np.finfo(np.float64).eps
        if not np.allclose(
            transformed[..., 1:],
            transformed[..., :0:-1],
            rtol=tolerance,
            atol=tolerance,
        ):
            raise ValueError("SpectralEnvelope requires symmetric real cepstral coefficients.")
        log_envelope = np.fft.rfft(transformed, axis=-1)
        envelope = np.exp(np.real(log_envelope))
        restored_axis = np.moveaxis(envelope, -1, axis)
        return np.asarray(restored_axis, dtype=np.complex128)

Attributes

name = 'spectral_envelope' class-attribute instance-attribute

axis property

Return the configured quefrency axis.

Functions

__init__(sampling_rate, *, axis=-1)

Source code in wandas/processing/cepstral.py

def __init__(self, sampling_rate: float, *, axis: int = -1) -> None:
    normalized_axis = _normalize_transform_axis(
        axis,
        operation_name="spectral envelope",
    )
    super().__init__(sampling_rate, axis=normalized_axis)

calculate_output_shape(input_shape)

Replace the quefrency axis with its one-sided frequency axis.

Source code in wandas/processing/cepstral.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """Replace the quefrency axis with its one-sided frequency axis."""
    axis = _resolve_transform_axis(
        self.axis,
        len(input_shape),
        operation_name="spectral envelope",
    )
    output_shape = list(input_shape)
    output_shape[axis] = int(input_shape[axis]) // 2 + 1
    return tuple(output_shape)

calculate_output_dtype(input_dtype, *input_dtypes)

Return the complex dtype matching the input precision.

Source code in wandas/processing/cepstral.py

def calculate_output_dtype(
    self,
    input_dtype: np.dtype[Any],
    *input_dtypes: np.dtype[Any],
) -> np.dtype[Any]:
    """Return the complex dtype matching the input precision."""
    return precision_complex_dtype(input_dtype)

Functions

wandas.processing.conversion

Explicit numerical representation conversions.

Attributes

__all__ = ['Astype'] module-attribute

Classes

Astype

Bases: ChannelIndependentAudioOperation[Any, Any]

Convert a raw Frame tensor to a supported real or complex floating dtype.

The eager kernel is channel-independent, preserves shape, and never mutates its input. :meth:process builds a lazy Dask graph whose dtype metadata is the exact selected target before computation. Real or integer inputs can produce float32/float64; complex inputs can produce complex64/complex128.

Parameters:

Name	Type	Description	Default
sampling_rate	float	Sampling rate in Hz. It is preserved and does not affect the numerical cast.	required
dtype	DTypeLike	Supported target NumPy dtype or equivalent dtype-like value.	required

Raises:

Type	Description
TypeError	If dtype is not understood by NumPy.
ValueError	If the target is unsupported or :meth:process receives an input whose real/complex domain does not match it.

Source code in wandas/processing/conversion.py

class Astype(ChannelIndependentAudioOperation[Any, Any]):
    """Convert a raw Frame tensor to a supported real or complex floating dtype.

    The eager kernel is channel-independent, preserves shape, and never mutates
    its input. :meth:`process` builds a lazy Dask graph whose dtype metadata is
    the exact selected target before computation. Real or integer inputs can
    produce float32/float64; complex inputs can produce complex64/complex128.

    Args:
        sampling_rate: Sampling rate in Hz. It is preserved and does not affect
            the numerical cast.
        dtype: Supported target NumPy dtype or equivalent dtype-like value.

    Raises:
        TypeError: If *dtype* is not understood by NumPy.
        ValueError: If the target is unsupported or :meth:`process` receives an
            input whose real/complex domain does not match it.
    """

    name = "astype"
    _display = "astype"

    def __init__(self, sampling_rate: float, dtype: npt.DTypeLike) -> None:
        """Initialize a dtype conversion with a canonical target dtype."""
        super().__init__(sampling_rate, dtype=_normalize_target_dtype(dtype))

    @property
    def dtype(self) -> str:
        """Return the canonical target dtype name."""
        return self._config_value("dtype")

    def validate_params(self) -> None:
        """Reject unsupported target representations at construction time."""
        if self.dtype not in _SUPPORTED_TARGET_DTYPES:
            raise ValueError(
                "Unsupported dtype for astype\n"
                f"  Got: {self.dtype}\n"
                "  Expected: float32, float64, complex64, or complex128\n"
                "Choose a supported floating representation."
            )

    def calculate_output_dtype(
        self,
        input_dtype: np.dtype[Any],
        *input_dtypes: np.dtype[Any],
    ) -> np.dtype[Any]:
        """Return exact output metadata after validating the source domain."""
        del input_dtypes
        return np.dtype(_normalize_astype_dtype(input_dtype, self.dtype))

    def _build_execution_graph(
        self,
        data: DaArray,
        inputs: tuple[DaArray, ...],
        *,
        output_shape: tuple[int, ...],
        output_dtype: np.dtype[Any],
    ) -> DaArray:
        """Cast each existing Dask block without changing chunk boundaries."""
        del inputs, output_shape
        return data.astype(output_dtype)

    def _process(self, data: np.ndarray[Any, Any]) -> np.ndarray[Any, Any]:
        """Convert one eager channel-first tensor without mutating the input."""
        target = _normalize_astype_dtype(data.dtype, self.dtype)
        return data.astype(target, copy=False)

Attributes

name = 'astype' class-attribute instance-attribute

dtype property

Return the canonical target dtype name.

Functions

__init__(sampling_rate, dtype)

Initialize a dtype conversion with a canonical target dtype.

Source code in wandas/processing/conversion.py

def __init__(self, sampling_rate: float, dtype: npt.DTypeLike) -> None:
    """Initialize a dtype conversion with a canonical target dtype."""
    super().__init__(sampling_rate, dtype=_normalize_target_dtype(dtype))

validate_params()

Reject unsupported target representations at construction time.

Source code in wandas/processing/conversion.py

def validate_params(self) -> None:
    """Reject unsupported target representations at construction time."""
    if self.dtype not in _SUPPORTED_TARGET_DTYPES:
        raise ValueError(
            "Unsupported dtype for astype\n"
            f"  Got: {self.dtype}\n"
            "  Expected: float32, float64, complex64, or complex128\n"
            "Choose a supported floating representation."
        )

calculate_output_dtype(input_dtype, *input_dtypes)

Return exact output metadata after validating the source domain.

Source code in wandas/processing/conversion.py

def calculate_output_dtype(
    self,
    input_dtype: np.dtype[Any],
    *input_dtypes: np.dtype[Any],
) -> np.dtype[Any]:
    """Return exact output metadata after validating the source domain."""
    del input_dtypes
    return np.dtype(_normalize_astype_dtype(input_dtype, self.dtype))

Functions

integer_arithmetic_as_float64(*operands)

Return arithmetic operands cast to float64 when NumPy would compute in an integer dtype.

Array operands are cast lazily; scalars are returned unchanged. Operands whose promoted dtype is already floating or complex are returned as given, so integer arithmetic cannot wrap around and other dtypes are unaffected.

Source code in wandas/processing/conversion.py

def integer_arithmetic_as_float64(*operands: Any) -> tuple[Any, ...]:
    """Return arithmetic operands cast to float64 when NumPy would compute in an integer dtype.

    Array operands are cast lazily; scalars are returned unchanged. Operands
    whose promoted dtype is already floating or complex are returned as given,
    so integer arithmetic cannot wrap around and other dtypes are unaffected.
    """
    try:
        result_dtype = np.result_type(*(_dtype_or_scalar(operand) for operand in operands))
    except (TypeError, ValueError):
        return operands
    if not np.issubdtype(result_dtype, np.integer):
        return operands
    return tuple(
        operand.astype(np.float64) if isinstance(operand, DaArray | np.ndarray) else operand for operand in operands
    )

wandas.processing.stats

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

ABS

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Absolute value operation

Source code in wandas/processing/stats.py

class ABS(AudioOperation[NDArrayReal, NDArrayReal]):
    """Absolute value operation"""

    name = "abs"
    _display = "abs"

    def __init__(self, sampling_rate: float):
        """
        Initialize absolute value operation

        Args:
            sampling_rate: float. Sampling rate (Hz)
        """
        super().__init__(sampling_rate)

    def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
        (values,) = integer_arithmetic_as_float64(data)
        return da.abs(values)

Attributes

name = 'abs' class-attribute instance-attribute

Functions

__init__(sampling_rate)

Initialize absolute value operation

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate (Hz)	required

Source code in wandas/processing/stats.py

def __init__(self, sampling_rate: float):
    """
    Initialize absolute value operation

    Args:
        sampling_rate: float. Sampling rate (Hz)
    """
    super().__init__(sampling_rate)

process(data, *inputs)

Source code in wandas/processing/stats.py

def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
    (values,) = integer_arithmetic_as_float64(data)
    return da.abs(values)

Power

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Power operation

Source code in wandas/processing/stats.py

class Power(AudioOperation[NDArrayReal, NDArrayReal]):
    """Power operation"""

    name = "power"
    _display = "pow"

    def __init__(self, sampling_rate: float, exponent: float):
        """
        Initialize power operation

        Args:
            sampling_rate: float. Sampling rate (Hz)
            exponent: float. Power exponent
        """
        super().__init__(sampling_rate, exponent=exponent)

    @property
    def exponent(self) -> float:
        """Exponent captured at operation construction time."""
        return self._config_value("exponent")

    @property
    def exp(self) -> float:
        """Backward-compatible read-only alias for the captured exponent."""
        return self.exponent

    def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
        values, exponent = integer_arithmetic_as_float64(data, self.exponent)
        return da.power(values, exponent)

Attributes

name = 'power' class-attribute instance-attribute

exponent property

Exponent captured at operation construction time.

exp property

Backward-compatible read-only alias for the captured exponent.

Functions

__init__(sampling_rate, exponent)

Initialize power operation

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate (Hz)	required
exponent	float	float. Power exponent	required

Source code in wandas/processing/stats.py

def __init__(self, sampling_rate: float, exponent: float):
    """
    Initialize power operation

    Args:
        sampling_rate: float. Sampling rate (Hz)
        exponent: float. Power exponent
    """
    super().__init__(sampling_rate, exponent=exponent)

process(data, *inputs)

Source code in wandas/processing/stats.py

def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
    values, exponent = integer_arithmetic_as_float64(data, self.exponent)
    return da.power(values, exponent)

Sum

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Sum calculation

Source code in wandas/processing/stats.py

class Sum(AudioOperation[NDArrayReal, NDArrayReal]):
    """Sum calculation"""

    name = "sum"
    _display = "sum"

    def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
        return data.sum(axis=0, keepdims=True)

Attributes

name = 'sum' class-attribute instance-attribute

Functions

process(data, *inputs)

Source code in wandas/processing/stats.py

def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
    return data.sum(axis=0, keepdims=True)

Mean

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Mean calculation

Source code in wandas/processing/stats.py

class Mean(AudioOperation[NDArrayReal, NDArrayReal]):
    """Mean calculation"""

    name = "mean"
    _display = "mean"

    def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
        return data.mean(axis=0, keepdims=True)

Attributes

name = 'mean' class-attribute instance-attribute

Functions

process(data, *inputs)

Source code in wandas/processing/stats.py

def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
    return data.mean(axis=0, keepdims=True)

ChannelDifference

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Channel difference calculation operation

Source code in wandas/processing/stats.py

class ChannelDifference(AudioOperation[NDArrayReal, NDArrayReal]):
    """Channel difference calculation operation"""

    name = "channel_difference"
    _display = "diff"

    def __init__(self, sampling_rate: float, other_channel: int = 0):
        """
        Initialize channel difference calculation

        Args:
            sampling_rate: float. Sampling rate (Hz)
            other_channel: int. Channel to calculate difference with, default is 0
        """
        super().__init__(sampling_rate, other_channel=other_channel)

    @property
    def other_channel(self) -> int:
        """Other channel index captured at operation construction time."""
        return self._config_value("other_channel")

    def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
        other_channel = self.other_channel
        if not -data.shape[0] <= other_channel < data.shape[0]:
            raise IndexError("Channel index out of range")
        (values,) = integer_arithmetic_as_float64(data)
        return values - values[other_channel]

Attributes

name = 'channel_difference' class-attribute instance-attribute

other_channel property

Other channel index captured at operation construction time.

Functions

__init__(sampling_rate, other_channel=0)

Initialize channel difference calculation

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate (Hz)	required
other_channel	int	int. Channel to calculate difference with, default is 0	0

Source code in wandas/processing/stats.py

def __init__(self, sampling_rate: float, other_channel: int = 0):
    """
    Initialize channel difference calculation

    Args:
        sampling_rate: float. Sampling rate (Hz)
        other_channel: int. Channel to calculate difference with, default is 0
    """
    super().__init__(sampling_rate, other_channel=other_channel)

process(data, *inputs)

Source code in wandas/processing/stats.py

def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
    other_channel = self.other_channel
    if not -data.shape[0] <= other_channel < data.shape[0]:
        raise IndexError("Channel index out of range")
    (values,) = integer_arithmetic_as_float64(data)
    return values - values[other_channel]

Functions

wandas.processing.filters

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

HighPassFilter

Bases: _ButterworthFilter

High-pass filter operation

Source code in wandas/processing/filters.py

class HighPassFilter(_ButterworthFilter):
    """High-pass filter operation"""

    name = "highpass_filter"
    _btype = "high"
    _display = "hpf"

Attributes

name = 'highpass_filter' class-attribute instance-attribute

LowPassFilter

Bases: _ButterworthFilter

Low-pass filter operation

Source code in wandas/processing/filters.py

class LowPassFilter(_ButterworthFilter):
    """Low-pass filter operation"""

    name = "lowpass_filter"
    _btype = "low"
    _display = "lpf"

Attributes

name = 'lowpass_filter' class-attribute instance-attribute

BandPassFilter

Bases: _ButterworthFilter

Band-pass filter operation

Source code in wandas/processing/filters.py

class BandPassFilter(_ButterworthFilter):
    """Band-pass filter operation"""

    name = "bandpass_filter"
    _btype = "band"
    _display = "bpf"

    def __init__(
        self,
        sampling_rate: float,
        low_cutoff: float,
        high_cutoff: float,
        order: int = 4,
    ):
        """
        Initialize band-pass filter

        Args:
            sampling_rate: float. Sampling rate (Hz)
            low_cutoff: float. Lower cutoff frequency (Hz). Must be between 0 and Nyquist frequency.
            high_cutoff: float. Higher cutoff frequency (Hz). Must be between 0 and Nyquist frequency
                and greater than low_cutoff.
            order: int, optional. Order of the Butterworth prototype, default is 4.
                The resulting band-pass filter has order ``2 * order``, and because it
                is applied forward and backward the magnitude response is squared.

        Raises:
            ValueError: If either cutoff frequency is not within valid range (0 < cutoff < Nyquist),
                or if low_cutoff >= high_cutoff
        """
        # Skip single-cutoff _ButterworthFilter.__init__
        AudioOperation.__init__(self, sampling_rate, low_cutoff=low_cutoff, high_cutoff=high_cutoff, order=order)

    @property
    def low_cutoff(self) -> float:
        """Lower cutoff frequency captured at operation construction time."""
        return self._config_value("low_cutoff")

    @property
    def high_cutoff(self) -> float:
        """Higher cutoff frequency captured at operation construction time."""
        return self._config_value("high_cutoff")

    def validate_params(self) -> None:
        """Validate parameters"""
        low_cutoff = self.low_cutoff
        high_cutoff = self.high_cutoff
        _validate_cutoff(low_cutoff, self.sampling_rate, "Lower cutoff")
        _validate_cutoff(high_cutoff, self.sampling_rate, "Higher cutoff")
        if low_cutoff >= high_cutoff:
            raise ValueError(
                f"Invalid bandpass filter cutoff frequencies\n"
                f"  Lower cutoff: {low_cutoff} Hz\n"
                f"  Higher cutoff: {high_cutoff} Hz\n"
                f"  Problem: Lower cutoff must be less than higher cutoff\n"
                f"A bandpass filter passes frequencies between low and high\n"
                f"  cutoffs.\n"
                f"Ensure low_cutoff < high_cutoff\n"
                f"  (e.g., low_cutoff=100, high_cutoff=1000)"
            )

    def _normalized_cutoff(self) -> float | list[float]:
        nyquist = 0.5 * self.sampling_rate
        return [self.low_cutoff / nyquist, self.high_cutoff / nyquist]

Attributes

name = 'bandpass_filter' class-attribute instance-attribute

low_cutoff property

Lower cutoff frequency captured at operation construction time.

high_cutoff property

Higher cutoff frequency captured at operation construction time.

Functions

__init__(sampling_rate, low_cutoff, high_cutoff, order=4)

Initialize band-pass filter

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate (Hz)	required
low_cutoff	float	float. Lower cutoff frequency (Hz). Must be between 0 and Nyquist frequency.	required
high_cutoff	float	float. Higher cutoff frequency (Hz). Must be between 0 and Nyquist frequency and greater than low_cutoff.	required
order	int	int, optional. Order of the Butterworth prototype, default is 4. The resulting band-pass filter has order 2 * order, and because it is applied forward and backward the magnitude response is squared.	4

Raises:

Type	Description
ValueError	If either cutoff frequency is not within valid range (0 < cutoff < Nyquist), or if low_cutoff >= high_cutoff

Source code in wandas/processing/filters.py

def __init__(
    self,
    sampling_rate: float,
    low_cutoff: float,
    high_cutoff: float,
    order: int = 4,
):
    """
    Initialize band-pass filter

    Args:
        sampling_rate: float. Sampling rate (Hz)
        low_cutoff: float. Lower cutoff frequency (Hz). Must be between 0 and Nyquist frequency.
        high_cutoff: float. Higher cutoff frequency (Hz). Must be between 0 and Nyquist frequency
            and greater than low_cutoff.
        order: int, optional. Order of the Butterworth prototype, default is 4.
            The resulting band-pass filter has order ``2 * order``, and because it
            is applied forward and backward the magnitude response is squared.

    Raises:
        ValueError: If either cutoff frequency is not within valid range (0 < cutoff < Nyquist),
            or if low_cutoff >= high_cutoff
    """
    # Skip single-cutoff _ButterworthFilter.__init__
    AudioOperation.__init__(self, sampling_rate, low_cutoff=low_cutoff, high_cutoff=high_cutoff, order=order)

validate_params()

Validate parameters

Source code in wandas/processing/filters.py

def validate_params(self) -> None:
    """Validate parameters"""
    low_cutoff = self.low_cutoff
    high_cutoff = self.high_cutoff
    _validate_cutoff(low_cutoff, self.sampling_rate, "Lower cutoff")
    _validate_cutoff(high_cutoff, self.sampling_rate, "Higher cutoff")
    if low_cutoff >= high_cutoff:
        raise ValueError(
            f"Invalid bandpass filter cutoff frequencies\n"
            f"  Lower cutoff: {low_cutoff} Hz\n"
            f"  Higher cutoff: {high_cutoff} Hz\n"
            f"  Problem: Lower cutoff must be less than higher cutoff\n"
            f"A bandpass filter passes frequencies between low and high\n"
            f"  cutoffs.\n"
            f"Ensure low_cutoff < high_cutoff\n"
            f"  (e.g., low_cutoff=100, high_cutoff=1000)"
        )

AWeighting

Bases: ChannelIndependentAudioOperation[NDArrayReal, NDArrayReal]

Apply the implemented digital A-frequency-weighting curve.

The output is a linear waveform. This operation does not calculate RMS, convert to dB, or establish sound-level-meter conformance.

Source code in wandas/processing/filters.py

class AWeighting(ChannelIndependentAudioOperation[NDArrayReal, NDArrayReal]):
    """Apply the implemented digital A-frequency-weighting curve.

    The output is a linear waveform. This operation does not calculate RMS,
    convert to dB, or establish sound-level-meter conformance.
    """

    name = "a_weighting"
    _display = "Aw"

    def __init__(self, sampling_rate: float):
        """
        Initialize A-weighting filter

        Args:
            sampling_rate: float. Sampling rate (Hz)
        """
        super().__init__(sampling_rate)

    _keeps_input_precision = True

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        return precision_real_dtype(input_dtype)

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Create processor function for A-weighting filter"""
        logger.debug(f"Applying A-weighting to array with shape: {x.shape}")
        result = A_weight(x, self.sampling_rate)

        # Handle case where A_weight returns a tuple
        if isinstance(result, tuple):
            # Use the first element of the tuple
            result = result[0]

        logger.debug(f"A-weighting applied, returning result with shape: {result.shape}")
        return np.array(result)

Attributes

name = 'a_weighting' class-attribute instance-attribute

Functions

__init__(sampling_rate)

Initialize A-weighting filter

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate (Hz)	required

Source code in wandas/processing/filters.py

def __init__(self, sampling_rate: float):
    """
    Initialize A-weighting filter

    Args:
        sampling_rate: float. Sampling rate (Hz)
    """
    super().__init__(sampling_rate)

calculate_output_dtype(input_dtype, *input_dtypes)

Source code in wandas/processing/filters.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    return precision_real_dtype(input_dtype)

Functions

wandas.processing.effects

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

HpssHarmonic

Bases: _HpssBase

HPSS Harmonic operation

Source code in wandas/processing/effects.py

class HpssHarmonic(_HpssBase):
    """HPSS Harmonic operation"""

    name = "hpss_harmonic"
    _extract_func = "harmonic"
    _display = "Hrm"

Attributes

name = 'hpss_harmonic' class-attribute instance-attribute

HpssPercussive

Bases: _HpssBase

HPSS Percussive operation

Source code in wandas/processing/effects.py

class HpssPercussive(_HpssBase):
    """HPSS Percussive operation"""

    name = "hpss_percussive"
    _extract_func = "percussive"
    _display = "Prc"

Attributes

name = 'hpss_percussive' class-attribute instance-attribute

Normalize

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Signal normalization operation.

Source code in wandas/processing/effects.py

class Normalize(AudioOperation[NDArrayReal, NDArrayReal]):
    """Signal normalization operation."""

    name = "normalize"
    _display = "norm"

    @staticmethod
    def _output_dtype(input_dtype: np.dtype[Any], norm: float | None) -> np.dtype[Any]:
        dtype = np.dtype(input_dtype)
        if norm is None:
            return dtype
        if dtype.kind == "f" and norm in {np.inf, -np.inf}:
            return dtype
        return np.dtype(np.float64)

    def __init__(
        self,
        sampling_rate: float,
        norm: float | None = np.inf,
        axis: int | None = -1,
        threshold: float | None = None,
        fill: bool | None = None,
    ):
        """
        Initialize normalization operation

        Args:
            sampling_rate: float. Sampling rate (Hz)
            norm: float or np.inf, default=np.inf. Norm type. Supported values:
                - np.inf: Maximum absolute value normalization
                - -np.inf: Minimum absolute value normalization
                - 0: Pseudo L0 normalization (divide by number of non-zero elements)
                - float: Lp norm
                - None: No normalization
            axis: int or None, default=-1. Axis along which to normalize.
                - -1: Normalize along time axis (each channel independently)
                - None: Global normalization across all axes
                - int: Normalize along specified axis
            threshold: float or None, optional. Threshold below which values are considered zero.
                If None, no threshold is applied.
            fill: bool or None, optional. Value to fill when the norm is zero.
                If None, the zero vector remains zero.

        Raises:
            ValueError: If norm parameter is invalid or threshold is negative
        """
        # Validate norm parameter
        if norm is not None and not isinstance(norm, int | float):
            raise ValueError(
                f"Invalid normalization method\n"
                f"  Got: {type(norm).__name__} ({norm})\n"
                f"  Expected: float, int, np.inf, -np.inf, or None\n"
                f"Norm parameter must be a numeric value or None.\n"
                f"Common values: np.inf (max norm), 2 (L2 norm),\n"
                f"1 (L1 norm), 0 (pseudo L0)"
            )

        # Validate that norm is non-negative (except for -np.inf which is valid)
        if norm is not None and norm < 0 and not np.isneginf(norm):
            raise ValueError(
                f"Invalid normalization method\n"
                f"  Got: {norm}\n"
                f"  Expected: Non-negative value, np.inf, -np.inf, or None\n"
                f"Norm parameter must be non-negative (except -np.inf for min norm).\n"
                f"Common values: np.inf (max norm), 2 (L2 norm),\n"
                f"1 (L1 norm), 0 (pseudo L0)"
            )

        # Validate threshold
        if threshold is not None and threshold <= 0:
            raise ValueError(
                f"Invalid threshold for normalization\n"
                f"  Got: {threshold}\n"
                f"  Expected: Positive value or None\n"
                f"Threshold must be strictly positive.\n"
                f"Typical values: 1e-10 (small threshold), 1e-6 (larger threshold)"
            )

        super().__init__(sampling_rate, norm=norm, axis=axis, threshold=threshold, fill=fill)
        logger.debug(
            f"Initialized Normalize operation with norm={norm}, axis={axis}, threshold={threshold}, fill={fill}"
        )

    @property
    def norm(self) -> float | None:
        """Norm captured at operation construction time."""
        return self._config_value("norm")

    @property
    def axis(self) -> int | None:
        """Axis captured at operation construction time."""
        return self._config_value("axis")

    @property
    def threshold(self) -> float | None:
        """Threshold captured at operation construction time."""
        return self._config_value("threshold")

    @property
    def fill(self) -> bool | None:
        """Fill behavior captured at operation construction time."""
        return self._config_value("fill")

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Perform normalization processing"""
        logger.debug(f"Applying normalization to array with shape: {x.shape}, norm={self.norm}, axis={self.axis}")

        result = _normalize_array(
            x,
            norm=self.norm,
            axis=self.axis,
            threshold=self.threshold,
            fill=self.fill,
        )

        logger.debug(f"Normalization applied, returning result with shape: {result.shape}")
        return result

    def _build_execution_graph(
        self,
        data: DaArray,
        inputs: tuple[DaArray, ...],
        *,
        output_shape: tuple[int, ...],
        output_dtype: np.dtype[Any],
    ) -> DaArray:
        axis = self.axis
        if self.norm is None or not isinstance(axis, int | np.integer) or axis not in {-1, data.ndim - 1}:
            return super()._build_execution_graph(
                data,
                inputs,
                output_shape=output_shape,
                output_dtype=output_dtype,
            )
        result = _try_build_channelwise_graph(
            self,
            data,
            inputs,
            output_shape=output_shape,
            output_dtype=output_dtype,
        )
        if result is not None:
            return result
        return super()._build_execution_graph(
            data,
            inputs,
            output_shape=output_shape,
            output_dtype=output_dtype,
        )

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        """Return normalization output dtype metadata."""
        return self._output_dtype(input_dtype, self.norm)

Attributes

name = 'normalize' class-attribute instance-attribute

norm property

Norm captured at operation construction time.

axis property

Axis captured at operation construction time.

threshold property

Threshold captured at operation construction time.

fill property

Fill behavior captured at operation construction time.

Functions

__init__(sampling_rate, norm=np.inf, axis=-1, threshold=None, fill=None)

Initialize normalization operation

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate (Hz)	required
norm	float | None	float or np.inf, default=np.inf. Norm type. Supported values: - np.inf: Maximum absolute value normalization - -np.inf: Minimum absolute value normalization - 0: Pseudo L0 normalization (divide by number of non-zero elements) - float: Lp norm - None: No normalization	inf
axis	int | None	int or None, default=-1. Axis along which to normalize. - -1: Normalize along time axis (each channel independently) - None: Global normalization across all axes - int: Normalize along specified axis	-1
threshold	float | None	float or None, optional. Threshold below which values are considered zero. If None, no threshold is applied.	None
fill	bool | None	bool or None, optional. Value to fill when the norm is zero. If None, the zero vector remains zero.	None

Raises:

Type	Description
ValueError	If norm parameter is invalid or threshold is negative

Source code in wandas/processing/effects.py

def __init__(
    self,
    sampling_rate: float,
    norm: float | None = np.inf,
    axis: int | None = -1,
    threshold: float | None = None,
    fill: bool | None = None,
):
    """
    Initialize normalization operation

    Args:
        sampling_rate: float. Sampling rate (Hz)
        norm: float or np.inf, default=np.inf. Norm type. Supported values:
            - np.inf: Maximum absolute value normalization
            - -np.inf: Minimum absolute value normalization
            - 0: Pseudo L0 normalization (divide by number of non-zero elements)
            - float: Lp norm
            - None: No normalization
        axis: int or None, default=-1. Axis along which to normalize.
            - -1: Normalize along time axis (each channel independently)
            - None: Global normalization across all axes
            - int: Normalize along specified axis
        threshold: float or None, optional. Threshold below which values are considered zero.
            If None, no threshold is applied.
        fill: bool or None, optional. Value to fill when the norm is zero.
            If None, the zero vector remains zero.

    Raises:
        ValueError: If norm parameter is invalid or threshold is negative
    """
    # Validate norm parameter
    if norm is not None and not isinstance(norm, int | float):
        raise ValueError(
            f"Invalid normalization method\n"
            f"  Got: {type(norm).__name__} ({norm})\n"
            f"  Expected: float, int, np.inf, -np.inf, or None\n"
            f"Norm parameter must be a numeric value or None.\n"
            f"Common values: np.inf (max norm), 2 (L2 norm),\n"
            f"1 (L1 norm), 0 (pseudo L0)"
        )

    # Validate that norm is non-negative (except for -np.inf which is valid)
    if norm is not None and norm < 0 and not np.isneginf(norm):
        raise ValueError(
            f"Invalid normalization method\n"
            f"  Got: {norm}\n"
            f"  Expected: Non-negative value, np.inf, -np.inf, or None\n"
            f"Norm parameter must be non-negative (except -np.inf for min norm).\n"
            f"Common values: np.inf (max norm), 2 (L2 norm),\n"
            f"1 (L1 norm), 0 (pseudo L0)"
        )

    # Validate threshold
    if threshold is not None and threshold <= 0:
        raise ValueError(
            f"Invalid threshold for normalization\n"
            f"  Got: {threshold}\n"
            f"  Expected: Positive value or None\n"
            f"Threshold must be strictly positive.\n"
            f"Typical values: 1e-10 (small threshold), 1e-6 (larger threshold)"
        )

    super().__init__(sampling_rate, norm=norm, axis=axis, threshold=threshold, fill=fill)
    logger.debug(
        f"Initialized Normalize operation with norm={norm}, axis={axis}, threshold={threshold}, fill={fill}"
    )

calculate_output_dtype(input_dtype, *input_dtypes)

Return normalization output dtype metadata.

Source code in wandas/processing/effects.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    """Return normalization output dtype metadata."""
    return self._output_dtype(input_dtype, self.norm)

RemoveDC

Bases: ChannelIndependentAudioOperation[NDArrayReal, NDArrayReal]

Remove DC component (DC offset) from the signal.

This operation removes the DC component by subtracting the mean value from each channel, centering the signal around zero.

Source code in wandas/processing/effects.py

class RemoveDC(ChannelIndependentAudioOperation[NDArrayReal, NDArrayReal]):
    """Remove DC component (DC offset) from the signal.

    This operation removes the DC component by subtracting the mean value
    from each channel, centering the signal around zero.
    """

    name = "remove_dc"
    _display = "dcRM"

    def __init__(self, sampling_rate: float):
        """Initialize DC removal operation.

        Args:
            sampling_rate: float. Sampling rate (Hz)
        """
        super().__init__(sampling_rate)
        logger.debug("Initialized RemoveDC operation")

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        if np.issubdtype(input_dtype, np.integer):
            return np.dtype(np.float64)
        return np.dtype(input_dtype)

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Perform DC removal processing.

        Args:
            x: NDArrayReal. Input signal array (channels, samples)

        Returns:
            NDArrayReal: Signal with DC component removed
        """
        logger.debug(f"Removing DC component from array with shape: {x.shape}")

        # Subtract mean along time axis (axis=1 for channel data)
        mean_values = x.mean(axis=-1, keepdims=True)
        result: NDArrayReal = x - mean_values

        logger.debug(f"DC removal applied, returning result with shape: {result.shape}")
        return result

Attributes

name = 'remove_dc' class-attribute instance-attribute

Functions

__init__(sampling_rate)

Initialize DC removal operation.

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate (Hz)	required

Source code in wandas/processing/effects.py

def __init__(self, sampling_rate: float):
    """Initialize DC removal operation.

    Args:
        sampling_rate: float. Sampling rate (Hz)
    """
    super().__init__(sampling_rate)
    logger.debug("Initialized RemoveDC operation")

calculate_output_dtype(input_dtype, *input_dtypes)

Source code in wandas/processing/effects.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    if np.issubdtype(input_dtype, np.integer):
        return np.dtype(np.float64)
    return np.dtype(input_dtype)

AddWithSNR

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Addition operation considering SNR

Source code in wandas/processing/effects.py

class AddWithSNR(AudioOperation[NDArrayReal, NDArrayReal]):
    """Addition operation considering SNR"""

    name = "add_with_snr"
    _display = "+SNR"
    _expected_input_count = 2

    def __init__(self, sampling_rate: float, snr: float = 1.0):
        """
        Initialize addition operation considering SNR

        Args:
            sampling_rate: float. Sampling rate (Hz)
            snr: float. Signal-to-noise ratio (dB)
        """
        super().__init__(sampling_rate, snr=snr)
        logger.debug(f"Initialized AddWithSNR operation with SNR: {snr} dB")

    @property
    def snr(self) -> float:
        """Signal-to-noise ratio captured at operation construction time."""
        return self._config_value("snr")

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        """Promote SNR mixing to at least float32 precision."""
        return np.result_type(input_dtype, *input_dtypes, np.float32)

    def _process(self, x: NDArrayReal, other: NDArrayReal) -> NDArrayReal:
        """Perform addition processing considering SNR."""
        logger.debug(f"Applying SNR-based addition with shape: {x.shape}")
        output_dtype = self.calculate_output_dtype(x.dtype, other.dtype)
        clean = np.asarray(x, dtype=output_dtype)
        noise = np.asarray(other, dtype=output_dtype)

        clean_rms = util.calculate_rms(clean)
        other_rms = util.calculate_rms(noise)
        desired_noise_rms = util.calculate_desired_noise_rms(clean_rms, self.snr)
        gain = np.zeros_like(desired_noise_rms, dtype=output_dtype)
        np.divide(desired_noise_rms, other_rms, out=gain, where=other_rms != 0)
        result: NDArrayReal = clean + noise * gain
        return np.asarray(result, dtype=output_dtype)

Attributes

name = 'add_with_snr' class-attribute instance-attribute

snr property

Signal-to-noise ratio captured at operation construction time.

Functions

__init__(sampling_rate, snr=1.0)

Initialize addition operation considering SNR

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate (Hz)	required
snr	float	float. Signal-to-noise ratio (dB)	1.0

Source code in wandas/processing/effects.py

def __init__(self, sampling_rate: float, snr: float = 1.0):
    """
    Initialize addition operation considering SNR

    Args:
        sampling_rate: float. Sampling rate (Hz)
        snr: float. Signal-to-noise ratio (dB)
    """
    super().__init__(sampling_rate, snr=snr)
    logger.debug(f"Initialized AddWithSNR operation with SNR: {snr} dB")

calculate_output_dtype(input_dtype, *input_dtypes)

Promote SNR mixing to at least float32 precision.

Source code in wandas/processing/effects.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    """Promote SNR mixing to at least float32 precision."""
    return np.result_type(input_dtype, *input_dtypes, np.float32)

Fade

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Fade operation using a Tukey (tapered cosine) window.

This operation applies symmetric fade-in and fade-out with the same duration. The Tukey window alpha parameter is computed from the fade duration so that the tapered portion equals the requested fade length at each end.

Source code in wandas/processing/effects.py

class Fade(AudioOperation[NDArrayReal, NDArrayReal]):
    """Fade operation using a Tukey (tapered cosine) window.

    This operation applies symmetric fade-in and fade-out with the same
    duration. The Tukey window alpha parameter is computed from the fade
    duration so that the tapered portion equals the requested fade length
    at each end.
    """

    name = "fade"
    _display = "fade"

    def __init__(self, sampling_rate: float, fade_ms: float = 50) -> None:
        fade_ms = float(fade_ms)
        super().__init__(sampling_rate, fade_ms=fade_ms)

    @property
    def fade_ms(self) -> float:
        """Fade duration captured at operation construction time."""
        return self._config_value("fade_ms")

    def validate_params(self) -> None:
        if self.fade_ms < 0:
            raise ValueError("fade_ms must be non-negative")

    def _fade_len_for_sampling_rate(self) -> int:
        return round(self.fade_ms * self.sampling_rate / 1000.0)

    @staticmethod
    def calculate_tukey_alpha(fade_len: int, n_samples: int) -> float:
        """Calculate Tukey window alpha parameter from fade length.

        The alpha parameter determines what fraction of the window is tapered.
        For symmetric fade-in/fade-out, alpha = 2 * fade_len / n_samples ensures
        that each side's taper has exactly fade_len samples.

        Args:
            fade_len: int. Desired fade length in samples for each end (in and out).
            n_samples: int. Total number of samples in the signal.

        Returns:
            float: Alpha parameter for scipy.signal.windows.tukey, clamped to [0, 1].

        Examples:
            >>> Fade.calculate_tukey_alpha(fade_len=20, n_samples=200)
            0.2
            >>> Fade.calculate_tukey_alpha(fade_len=100, n_samples=100)
            1.0
        """
        alpha = float(2 * fade_len) / float(n_samples)
        return min(1.0, alpha)

    _keeps_input_precision = True

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        del input_dtypes
        if self._fade_len_for_sampling_rate() <= 0 or np.dtype(input_dtype) == np.float32:
            return np.dtype(input_dtype)
        return np.result_type(input_dtype, np.float64)

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        logger.debug(f"Applying Tukey Fade to array with shape: {x.shape}")

        arr = x
        if arr.ndim == 1:
            arr = arr.reshape(1, -1)

        n_samples = int(arr.shape[-1])

        # If no fade requested, return input
        fade_len = self._fade_len_for_sampling_rate()

        if fade_len <= 0:
            return arr

        if 2 * fade_len >= n_samples:
            raise ValueError("Fade length too long: 2*fade_ms must be less than signal length")

        # Calculate Tukey window alpha parameter
        alpha = self.calculate_tukey_alpha(fade_len, n_samples)

        # Create tukey window (numpy) and apply
        env = scipy.signal.windows.tukey(n_samples, alpha=alpha)

        result: NDArrayReal = arr * env[None, :]
        logger.debug("Tukey fade applied")
        return result

Attributes

name = 'fade' class-attribute instance-attribute

fade_ms property

Fade duration captured at operation construction time.

Functions

__init__(sampling_rate, fade_ms=50)

Source code in wandas/processing/effects.py

def __init__(self, sampling_rate: float, fade_ms: float = 50) -> None:
    fade_ms = float(fade_ms)
    super().__init__(sampling_rate, fade_ms=fade_ms)

validate_params()

Source code in wandas/processing/effects.py

def validate_params(self) -> None:
    if self.fade_ms < 0:
        raise ValueError("fade_ms must be non-negative")

calculate_tukey_alpha(fade_len, n_samples) staticmethod

Calculate Tukey window alpha parameter from fade length.

The alpha parameter determines what fraction of the window is tapered. For symmetric fade-in/fade-out, alpha = 2 * fade_len / n_samples ensures that each side's taper has exactly fade_len samples.

Parameters:

Name	Type	Description	Default
fade_len	int	int. Desired fade length in samples for each end (in and out).	required
n_samples	int	int. Total number of samples in the signal.	required

Returns:

Name	Type	Description
float	float	Alpha parameter for scipy.signal.windows.tukey, clamped to [0, 1].

Examples:

>>> Fade.calculate_tukey_alpha(fade_len=20, n_samples=200)
0.2
>>> Fade.calculate_tukey_alpha(fade_len=100, n_samples=100)
1.0

Source code in wandas/processing/effects.py

@staticmethod
def calculate_tukey_alpha(fade_len: int, n_samples: int) -> float:
    """Calculate Tukey window alpha parameter from fade length.

    The alpha parameter determines what fraction of the window is tapered.
    For symmetric fade-in/fade-out, alpha = 2 * fade_len / n_samples ensures
    that each side's taper has exactly fade_len samples.

    Args:
        fade_len: int. Desired fade length in samples for each end (in and out).
        n_samples: int. Total number of samples in the signal.

    Returns:
        float: Alpha parameter for scipy.signal.windows.tukey, clamped to [0, 1].

    Examples:
        >>> Fade.calculate_tukey_alpha(fade_len=20, n_samples=200)
        0.2
        >>> Fade.calculate_tukey_alpha(fade_len=100, n_samples=100)
        1.0
    """
    alpha = float(2 * fade_len) / float(n_samples)
    return min(1.0, alpha)

calculate_output_dtype(input_dtype, *input_dtypes)

Source code in wandas/processing/effects.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    del input_dtypes
    if self._fade_len_for_sampling_rate() <= 0 or np.dtype(input_dtype) == np.float32:
        return np.dtype(input_dtype)
    return np.result_type(input_dtype, np.float64)

Functions