Skip to content

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', 'FFT', 'IFFT', 'ISTFT', 'NOctSpectrum', 'NOctSynthesis', 'STFT', 'TransferFunction', '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 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
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

    _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 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
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 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
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 
240
241
242
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 
252
253
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 
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
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 
290
291
292
293
294
295
296
297
298
299
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 
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
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 
339
340
341
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 
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
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 
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
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 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
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 
101
102
103
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 
110
111
112
113
114
115
116
117
118
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 
120
121
122
123
124
125
126
127
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 
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
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 
309
310
311
312
313
314
315
316
317
318
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 
325
326
327
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 
88
89
90
91
92
93
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 
 96
 97
 98
 99
100
101
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 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
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)

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

    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 
194
195
196
197
198
199
200
201
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 
203
204
def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    return np.dtype(np.float64)

HighPassFilter

Bases: _ButterworthFilter

High-pass filter operation

Source code in wandas/processing/filters.py 
94
95
96
97
98
99
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 
102
103
104
105
106
107
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 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
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:
        return da.abs(data)
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 
18
19
20
21
22
23
24
25
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 
27
28
def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
    return da.abs(data)

ChannelDifference

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Channel difference calculation operation

Source code in wandas/processing/stats.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
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")
        return data - data[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 
87
88
89
90
91
92
93
94
95
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 
102
103
104
105
106
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")
    return data - data[other_channel]

Mean

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Mean calculation

Source code in wandas/processing/stats.py 
71
72
73
74
75
76
77
78
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 
77
78
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 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
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:
        return da.power(data, self.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 
37
38
39
40
41
42
43
44
45
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 
57
58
def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
    return da.power(data, self.exponent)

Sum

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Sum calculation

Source code in wandas/processing/stats.py 
61
62
63
64
65
66
67
68
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 
67
68
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 
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
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 = resample_poly(x, up, down, axis=-1)
        else:
            result = 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 
444
445
446
447
448
449
450
451
452
453
454
455
456
457
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 
464
465
466
467
468
469
470
471
472
473
474
475
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 
477
478
479
480
481
482
483
484
485
486
487
488
489
490
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 
512
513
514
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 
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
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)

    @staticmethod
    def _output_dtype() -> np.dtype[Any]:
        return np.dtype(np.float64)

    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 self._output_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 
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
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 
748
749
750
751
752
753
754
755
756
757
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 
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
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 
857
858
859
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 self._output_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 
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
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 = 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 = 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 
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
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 
1039
1040
1041
1042
1043
1044
1045
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 
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
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 
1160
1161
1162
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 
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
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 
527
528
529
530
531
532
533
534
535
536
537
538
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 
560
561
562
563
564
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 
701
702
703
704
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 
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
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 
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
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 
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
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 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
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 
108
109
110
111
112
113
114
115
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 
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
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 = resample_poly(x, up, down, axis=-1)
        else:
            result = 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 
444
445
446
447
448
449
450
451
452
453
454
455
456
457
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 
464
465
466
467
468
469
470
471
472
473
474
475
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 
477
478
479
480
481
482
483
484
485
486
487
488
489
490
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 
512
513
514
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 
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
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 
527
528
529
530
531
532
533
534
535
536
537
538
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 
560
561
562
563
564
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 
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
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 
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
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 
606
607
608
609
610
611
612
613
614
615
616
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 
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
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)

    @staticmethod
    def _output_dtype() -> np.dtype[Any]:
        return np.dtype(np.float64)

    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 self._output_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 
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
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 
748
749
750
751
752
753
754
755
756
757
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 
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
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 
857
858
859
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 self._output_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 
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
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 = 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 = 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 
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
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 
1039
1040
1041
1042
1043
1044
1045
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 
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
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 
1160
1161
1162
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 or zero-padded to n_fft before the selected window is applied. 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 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
class FFT(AudioOperation[NDArrayReal, NDArrayComplex]):
    """One-sided, coherent-gain-normalized peak-amplitude FFT.

    The input is truncated or zero-padded to ``n_fft`` before the selected
    window is applied. 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)

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

    def _process(self, x: NDArrayReal) -> NDArrayComplex:
        """Apply FFT to the input array."""
        fft_size = int(x.shape[-1]) if self.n_fft is None else self.n_fft
        if x.shape[-1] >= fft_size:
            x = x[..., :fft_size]
        else:
            x = np.pad(x, [(0, 0)] * (x.ndim - 1) + [(0, fft_size - x.shape[-1])])

        win = get_window(self.window, fft_size)
        x = x * win
        result: NDArrayComplex = np.fft.rfft(x, n=fft_size, axis=-1)
        scaling_factor = np.sum(win)
        return _normalize_rfft_amplitude(
            result,
            n_fft=fft_size,
            window_gain=float(scaling_factor),
        )
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 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
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 
257
258
259
260
261
262
263
264
265
266
267
268
269
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 
271
272
def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    return np.dtype(np.complex128)

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 and window, the result is the truncated-or-zero-padded input multiplied by that analysis window. 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 
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
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`` and
    ``window``, the result is the truncated-or-zero-padded input multiplied by
    that analysis window. 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"):
        """
        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'
        """
        super().__init__(sampling_rate, n_fft=n_fft, window=window)

    @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)

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

    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
        win = get_window(self.window, fft_size)
        _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
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')

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'
Source code in wandas/processing/spectral.py 
330
331
332
333
334
335
336
337
338
339
def __init__(self, sampling_rate: float, n_fft: int | None = None, window: str = "hann"):
    """
    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'
    """
    super().__init__(sampling_rate, n_fft=n_fft, window=window)
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 
351
352
353
354
355
356
357
358
359
360
361
362
363
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 
365
366
def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    return np.dtype(np.float64)

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 
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
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",
    ):
        """
        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'

        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, "STFT")

        self._SFT = ShortTimeFFT(
            win=get_window(window, actual_win_length),
            hop=actual_hop_length,
            fs=sampling_rate,
            mfft=n_fft,
            scale_to="magnitude",
        )
        super().__init__(
            sampling_rate,
            n_fft=n_fft,
            win_length=actual_win_length,
            hop_length=actual_hop_length,
            window=window,
        )

    @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")

    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)

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

    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)
        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.

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.

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

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'

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 
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
def __init__(
    self,
    sampling_rate: float,
    n_fft: int = 2048,
    hop_length: int | None = None,
    win_length: int | None = None,
    window: str = "hann",
):
    """
    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'

    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, "STFT")

    self._SFT = ShortTimeFFT(
        win=get_window(window, actual_win_length),
        hop=actual_hop_length,
        fs=sampling_rate,
        mfft=n_fft,
        scale_to="magnitude",
    )
    super().__init__(
        sampling_rate,
        n_fft=n_fft,
        win_length=actual_win_length,
        hop_length=actual_hop_length,
        window=window,
    )
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 
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
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 
493
494
def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    return np.dtype(np.complex128)

ISTFT

Bases: AudioOperation[NDArrayComplex, NDArrayReal]

Inverse Short-Time Fourier Transform operation

Source code in wandas/processing/spectral.py 
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
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 = ShortTimeFFT(
            win=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)

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

    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 
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
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 = ShortTimeFFT(
        win=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 
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
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 
661
662
def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    return np.dtype(np.float64)
process(data, *inputs)

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

Source code in wandas/processing/spectral.py 
692
693
694
695
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 
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
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.")

        _, result = ss.welch(
            x,
            nperseg=self.win_length,
            noverlap=self.noverlap,
            nfft=self.n_fft,
            window=self.window,
            average=self.average,
            detrend=self.detrend,
            scaling="spectrum",
        )

        # 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 
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
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 
790
791
792
793
794
795
796
797
798
799
800
801
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 
803
804
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 
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
class NOctSpectrum(_NOctBase, ChannelIndependentAudioOperation[NDArrayReal, NDArrayReal]):
    """N-octave spectrum operation"""

    name = "noct_spectrum"
    _display = "Oct"

    def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        """Advertise the float64 output produced by MoSQITo."""
        return np.dtype(np.float64)

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Create processor function for octave spectrum"""
        logger.debug(f"Applying NoctSpectrum to array with shape: {x.shape}")
        spec, frequencies = noct_spectrum(
            sig=x.T,
            fs=self.sampling_rate,
            fmin=self.fmin,
            fmax=self.fmax,
            n=self.n,
            G=self.G,
            fr=self.fr,
        )
        spec = np.asarray(spec).reshape(np.asarray(frequencies).size, 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)

Advertise the float64 output produced by MoSQITo.

Source code in wandas/processing/spectral.py 
961
962
963
def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    """Advertise the float64 output produced by MoSQITo."""
    return np.dtype(np.float64)

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.

Source code in wandas/processing/spectral.py 
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
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.
    """

    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,
    ) -> 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.

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

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

    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."
            )

    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_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
        """Advertise the real float64 output produced by MoSQITo."""
        return np.dtype(np.float64)

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Create processor function for octave synthesis."""
        logger.debug(f"Applying NoctSynthesis to array with shape: {x.shape}")
        result = self._synthesize(x, 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.

Functions
__init__(sampling_rate, fmin, fmax, n=3, G=10, fr=1000, *, n_fft)

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

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 or G is not 2 or 10.
Source code in wandas/processing/spectral.py 
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
def __init__(
    self,
    sampling_rate: float,
    fmin: float,
    fmax: float,
    n: int = 3,
    G: int = 10,
    fr: int = 1000,
    *,
    n_fft: int,
) -> 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.

    Raises:
        TypeError: If ``n_fft`` is not an integer or ``G`` is not an
            integer ratio convention.
        ValueError: If ``n_fft`` is not positive or ``G`` is not 2 or 10.
    """
    AudioOperation.__init__(
        self,
        sampling_rate,
        fmin=fmin,
        fmax=fmax,
        n=n,
        G=G,
        fr=fr,
        n_fft=n_fft,
    )
validate_params()

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

Source code in wandas/processing/spectral.py 
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
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."
        )
calculate_output_dtype(input_dtype, *input_dtypes)

Advertise the real float64 output produced by MoSQITo.

Source code in wandas/processing/spectral.py 
1072
1073
1074
def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    """Advertise the real float64 output produced by MoSQITo."""
    return np.dtype(np.float64)

Coherence

Bases: _CrossSpectralBase

Coherence estimation operation

Source code in wandas/processing/spectral.py 
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
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}")
        from scipy import signal as ss

        _, coh = ss.coherence(
            x=x[:, np.newaxis],
            y=x[np.newaxis, :],
            fs=self.sampling_rate,
            nperseg=self.win_length,
            noverlap=self.noverlap,
            nfft=self.n_fft,
            window=self.window,
            detrend=self.detrend,
        )

        # SciPy returns (input, output, frequency); expose output-major pairs.
        result: NDArrayReal = flatten_output_input_pairs(as_output_input_pairs(coh))

        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 
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
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}")
        from scipy import signal as ss

        # Calculate all combinations using scipy's csd function
        _, csd_result = ss.csd(
            x=x[:, np.newaxis],
            y=x[np.newaxis, :],
            fs=self.sampling_rate,
            nperseg=self.win_length,
            noverlap=self.noverlap,
            nfft=self.n_fft,
            window=self.window,
            detrend=self.detrend,
            scaling=self.scaling,
            average=self.average,
        )

        # SciPy returns (input, output, frequency); expose output-major pairs.
        result: NDArrayComplex = flatten_output_input_pairs(as_output_input_pairs(csd_result))

        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 
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
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}")
        from scipy import signal as ss

        # Calculate cross-spectral density between all channels
        _f, p_yx = ss.csd(
            x=x[:, np.newaxis, :],
            y=x[np.newaxis, :, :],
            fs=self.sampling_rate,
            nperseg=self.win_length,
            noverlap=self.noverlap,
            nfft=self.n_fft,
            window=self.window,
            detrend=self.detrend,
            scaling=self.scaling,
            average=self.average,
            axis=-1,
        )
        # p_yx shape: (input, output, frequency)

        # Calculate power spectral density for each channel
        _f, p_xx = ss.welch(
            x=x,
            fs=self.sampling_rate,
            nperseg=self.win_length,
            noverlap=self.noverlap,
            nfft=self.n_fft,
            window=self.window,
            detrend=self.detrend,
            scaling=self.scaling,
            average=self.average,
            axis=-1,
        )
        # p_xx shape: (num_channels, num_frequencies)

        # Calculate H[output, input] = P_out_in / P_in_in. Exact zero
        # denominators remain complex NaN; nonzero near-zero bins are untouched.
        h_f = transfer_function_ratio(as_output_input_pairs(p_yx), p_xx)
        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

Functions

noct_spectrum(*args, **kwargs)

Source code in wandas/processing/spectral.py 
23
24
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 
27
28
def noct_synthesis(*args: Any, **kwargs: Any) -> Any:
    return require_mosqito_sound_level_meter("noct_synthesis").noct_synthesis(*args, **kwargs)

validate_noct_recipe_params(params)

Validate portable N-octave parameters without importing MoSQITo.

Source code in wandas/processing/spectral.py 
56
57
58
59
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"])

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 each channel, 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 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
class Cepstrum(AudioOperation[NDArrayReal, NDArrayReal]):
    """Calculate a normalized real cepstrum.

    The operation windows each channel, 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)

    def calculate_output_dtype(
        self,
        input_dtype: np.dtype[Any],
        *input_dtypes: np.dtype[Any],
    ) -> np.dtype[Any]:
        """Return NumPy FFT's real output dtype."""
        return np.dtype(np.float64)

    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)
        if analysis.shape[-1] < n_fft:
            analysis = np.pad(
                analysis,
                [(0, 0)] * (analysis.ndim - 1) + [(0, n_fft - analysis.shape[-1])],
            )
        window_values = get_window(self.window, n_fft)
        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 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
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 
154
155
156
157
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 NumPy FFT's real output dtype.

Source code in wandas/processing/cepstral.py 
159
160
161
162
163
164
165
def calculate_output_dtype(
    self,
    input_dtype: np.dtype[Any],
    *input_dtypes: np.dtype[Any],
) -> np.dtype[Any]:
    """Return NumPy FFT's real output dtype."""
    return np.dtype(np.float64)

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 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
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]))

    def calculate_output_dtype(
        self,
        input_dtype: np.dtype[Any],
        *input_dtypes: np.dtype[Any],
    ) -> np.dtype[Any]:
        """Return NumPy FFT's real output dtype."""
        return np.dtype(np.float64)

    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 
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
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 
303
304
305
306
307
308
309
310
311
312
313
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 NumPy FFT's real output dtype.

Source code in wandas/processing/cepstral.py 
315
316
317
318
319
320
321
def calculate_output_dtype(
    self,
    input_dtype: np.dtype[Any],
    *input_dtypes: np.dtype[Any],
) -> np.dtype[Any]:
    """Return NumPy FFT's real output dtype."""
    return np.dtype(np.float64)

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 
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
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 
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
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 
406
407
408
409
410
411
412
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 
414
415
416
417
418
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 
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
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)

    def calculate_output_dtype(
        self,
        input_dtype: np.dtype[Any],
        *input_dtypes: np.dtype[Any],
    ) -> np.dtype[Any]:
        """Return the complex dtype used by ``SpectralFrame``."""
        return np.dtype(np.complex128)

    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 
480
481
482
483
484
485
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 
492
493
494
495
496
497
498
499
500
501
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 used by SpectralFrame.

Source code in wandas/processing/cepstral.py 
503
504
505
506
507
508
509
def calculate_output_dtype(
    self,
    input_dtype: np.dtype[Any],
    *input_dtypes: np.dtype[Any],
) -> np.dtype[Any]:
    """Return the complex dtype used by ``SpectralFrame``."""
    return np.dtype(np.complex128)

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 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
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 
101
102
103
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 
110
111
112
113
114
115
116
117
118
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 
120
121
122
123
124
125
126
127
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

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 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
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:
        return da.abs(data)
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 
18
19
20
21
22
23
24
25
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 
27
28
def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
    return da.abs(data)

Power

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Power operation

Source code in wandas/processing/stats.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
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:
        return da.power(data, self.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 
37
38
39
40
41
42
43
44
45
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 
57
58
def process(self, data: DaArray, *inputs: DaArray) -> DaArray:
    return da.power(data, self.exponent)

Sum

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Sum calculation

Source code in wandas/processing/stats.py 
61
62
63
64
65
66
67
68
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 
67
68
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 
71
72
73
74
75
76
77
78
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 
77
78
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 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
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")
        return data - data[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 
87
88
89
90
91
92
93
94
95
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 
102
103
104
105
106
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")
    return data - data[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 
94
95
96
97
98
99
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 
102
103
104
105
106
107
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 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
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. Filter order, default is 4

        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 _setup_processor(self) -> None:
        """Set up band-pass filter processor"""
        nyquist = 0.5 * self.sampling_rate
        low_normal_cutoff = self.low_cutoff / nyquist
        high_normal_cutoff = self.high_cutoff / nyquist

        # Precompute and save filter coefficients
        self._b, self._a = signal.butter(
            self.order,
            [low_normal_cutoff, high_normal_cutoff],
            btype="band",
        )
        logger.debug(f"Bandpass filter coefficients calculated: b={self._b}, a={self._a}")
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. Filter order, default is 4

 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 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
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. Filter order, default is 4

    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 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
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 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
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)

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

    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 
194
195
196
197
198
199
200
201
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 
203
204
def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    return np.dtype(np.float64)

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 
88
89
90
91
92
93
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 
 96
 97
 98
 99
100
101
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 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
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 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
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 
254
255
256
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 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
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 
269
270
271
272
273
274
275
276
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 
278
279
280
281
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 
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
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 
309
310
311
312
313
314
315
316
317
318
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 
325
326
327
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 
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
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)

    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:
            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 = sp_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 
357
358
359
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 
366
367
368
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 
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
@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 
397
398
399
400
401
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:
        return np.dtype(input_dtype)
    return np.result_type(input_dtype, np.float64)

Functions