Processing Module / 処理モジュール

The wandas.processing module provides various processing capabilities for audio data. wandas.processing モジュールは、オーディオデータに対する様々な処理機能を提供します。

Base Processing / 基本処理

Provides basic processing operations. 基本的な処理操作を提供します。

wandas.processing.base

Attributes

logger = logging.getLogger(__name__) module-attribute

InputArrayType = TypeVar('InputArrayType', NDArrayReal, NDArrayComplex) module-attribute

OutputArrayType = TypeVar('OutputArrayType', NDArrayReal, NDArrayComplex) module-attribute

Classes

AudioOperation

Bases: Generic[InputArrayType, OutputArrayType]

Numerical Dask operation with an operation-owned config snapshot.

Source code in wandas/processing/base.py

class AudioOperation(Generic[InputArrayType, OutputArrayType]):
    """Numerical Dask operation with an operation-owned config snapshot."""

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

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

        Parameters
        ----------
        input_shape : tuple
            Input data shape

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

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

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

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

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

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

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

Attributes

name class-attribute

pure = pure instance-attribute

sampling_rate property

Sampling rate captured at operation construction time.

params property

Return a read-only defensive snapshot of operation parameters.

Functions

__init_subclass__(**kwargs)

Ensure subclass process overrides keep the base input contract.

Source code in wandas/processing/base.py

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

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

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

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

Initialize AudioOperation.

Parameters

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

Source code in wandas/processing/base.py

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

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

    # Validate parameters during initialization
    self.validate_params()

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

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

to_params()

Return operation parameters used for lineage and display.

Source code in wandas/processing/base.py

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

validate_params()

Validate parameters (raises exception if invalid)

Source code in wandas/processing/base.py

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

get_metadata_updates()

Get metadata updates to apply after processing.

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

Returns

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.

Source code in wandas/processing/base.py

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

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

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

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

    >>> return {}

    Return new sampling rate for operations that resample:

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

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

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

get_display_name()

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

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

Source code in wandas/processing/base.py

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

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

calculate_output_shape(input_shape)

Calculate output data shape after operation.

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

Parameters

input_shape : tuple Input data shape

Returns

tuple Output data shape

Source code in wandas/processing/base.py

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

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

    Parameters
    ----------
    input_shape : tuple
        Input data shape

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

calculate_output_dtype(input_dtype, *input_dtypes)

Calculate output dtype metadata after operation.

Source code in wandas/processing/base.py

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

process(data, *inputs)

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

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

Source code in wandas/processing/base.py

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

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

Functions

register_operation(operation_class)

Register a new operation type

Source code in wandas/processing/base.py

def register_operation(operation_class: type) -> None:
    """Register a new operation type"""

    if not issubclass(operation_class, AudioOperation):
        raise TypeError("Strategy class must inherit from AudioOperation.")
    if inspect.isabstract(operation_class):
        raise TypeError("Cannot register abstract AudioOperation class.")

    existing = _OPERATION_REGISTRY.get(operation_class.name)
    if (
        existing is not None
        and existing.__module__ == operation_class.__module__
        and existing.__qualname__ == operation_class.__qualname__
    ):
        return

    _OPERATION_REGISTRY[operation_class.name] = operation_class

register_lazy_operation(name, module_name)

Register an operation that can be loaded from module_name on demand.

Source code in wandas/processing/base.py

def register_lazy_operation(name: str, module_name: str) -> None:
    """Register an operation that can be loaded from *module_name* on demand."""
    _OPERATION_MODULES[name] = module_name

get_operation(name)

Get operation class by name

Source code in wandas/processing/base.py

def get_operation(name: str) -> type[AudioOperation[Any, Any]]:
    """Get operation class by name"""
    if name not in _OPERATION_REGISTRY and name in _OPERATION_MODULES:
        importlib.import_module(_OPERATION_MODULES[name])
    if name not in _OPERATION_REGISTRY:
        raise ValueError(f"Unknown operation type: {name}")
    return _OPERATION_REGISTRY[name]

create_operation(name, sampling_rate, **params)

Create operation instance from name and parameters

Source code in wandas/processing/base.py

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

Effects / エフェクト

Provides audio effect processing. オーディオエフェクト処理を提供します。

wandas.processing.effects

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

HpssHarmonic

Bases: _HpssBase

HPSS Harmonic operation

Source code in wandas/processing/effects.py

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

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

Attributes

name = 'hpss_harmonic' class-attribute instance-attribute

HpssPercussive

Bases: _HpssBase

HPSS Percussive operation

Source code in wandas/processing/effects.py

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

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

Attributes

name = 'hpss_percussive' class-attribute instance-attribute

Normalize

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Signal normalization operation.

Source code in wandas/processing/effects.py

class Normalize(AudioOperation[NDArrayReal, NDArrayReal]):
    """Signal normalization operation."""

    name = "normalize"
    _display = "norm"

    @staticmethod
    def _output_dtype(input_dtype: np.dtype[Any], norm: float | None) -> np.dtype[Any]:
        dtype = np.dtype(input_dtype)
        if norm is None:
            return dtype
        if dtype.kind == "f" and norm in {np.inf, -np.inf}:
            return dtype
        return np.dtype(np.float64)

    def __init__(
        self,
        sampling_rate: float,
        norm: float | None = np.inf,
        axis: int | None = -1,
        threshold: float | None = None,
        fill: bool | None = None,
    ):
        """
        Initialize normalization operation

        Parameters
        ----------
        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 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

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

Source code in wandas/processing/effects.py

def __init__(
    self,
    sampling_rate: float,
    norm: float | None = np.inf,
    axis: int | None = -1,
    threshold: float | None = None,
    fill: bool | None = None,
):
    """
    Initialize normalization operation

    Parameters
    ----------
    sampling_rate : float
        Sampling rate (Hz)
    norm : float or np.inf, default=np.inf
        Norm type. Supported values:
        - np.inf: Maximum absolute value normalization
        - -np.inf: Minimum absolute value normalization
        - 0: Pseudo L0 normalization (divide by number of non-zero elements)
        - float: Lp norm
        - None: No normalization
    axis : int or None, default=-1
        Axis along which to normalize.
        - -1: Normalize along time axis (each channel independently)
        - None: Global normalization across all axes
        - int: Normalize along specified axis
    threshold : float or None, optional
        Threshold below which values are considered zero.
        If None, no threshold is applied.
    fill : bool or None, optional
        Value to fill when the norm is zero.
        If None, the zero vector remains zero.

    Raises
    ------
    ValueError
        If norm parameter is invalid or threshold is negative
    """
    # Validate norm parameter
    if norm is not None and not isinstance(norm, int | float):
        raise ValueError(
            f"Invalid normalization method\n"
            f"  Got: {type(norm).__name__} ({norm})\n"
            f"  Expected: float, int, np.inf, -np.inf, or None\n"
            f"Norm parameter must be a numeric value or None.\n"
            f"Common values: np.inf (max norm), 2 (L2 norm),\n"
            f"1 (L1 norm), 0 (pseudo L0)"
        )

    # Validate that norm is non-negative (except for -np.inf which is valid)
    if norm is not None and norm < 0 and not np.isneginf(norm):
        raise ValueError(
            f"Invalid normalization method\n"
            f"  Got: {norm}\n"
            f"  Expected: Non-negative value, np.inf, -np.inf, or None\n"
            f"Norm parameter must be non-negative (except -np.inf for min norm).\n"
            f"Common values: np.inf (max norm), 2 (L2 norm),\n"
            f"1 (L1 norm), 0 (pseudo L0)"
        )

    # Validate threshold
    if threshold is not None and threshold <= 0:
        raise ValueError(
            f"Invalid threshold for normalization\n"
            f"  Got: {threshold}\n"
            f"  Expected: Positive value or None\n"
            f"Threshold must be strictly positive.\n"
            f"Typical values: 1e-10 (small threshold), 1e-6 (larger threshold)"
        )

    super().__init__(sampling_rate, norm=norm, axis=axis, threshold=threshold, fill=fill)
    logger.debug(
        f"Initialized Normalize operation with norm={norm}, axis={axis}, threshold={threshold}, fill={fill}"
    )

calculate_output_dtype(input_dtype, *input_dtypes)

Return normalization output dtype metadata.

Source code in wandas/processing/effects.py

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

RemoveDC

Bases: _ChannelIndependentAudioOperation[NDArrayReal, NDArrayReal]

Remove DC component (DC offset) from the signal.

This operation removes the DC component by subtracting the mean value from each channel, centering the signal around zero.

Source code in wandas/processing/effects.py

class RemoveDC(_ChannelIndependentAudioOperation[NDArrayReal, NDArrayReal]):
    """Remove DC component (DC offset) from the signal.

    This operation removes the DC component by subtracting the mean value
    from each channel, centering the signal around zero.
    """

    name = "remove_dc"
    _display = "dcRM"

    def __init__(self, sampling_rate: float):
        """Initialize DC removal operation.

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

        Parameters
        ----------
        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

sampling_rate : float Sampling rate (Hz)

Source code in wandas/processing/effects.py

def __init__(self, sampling_rate: float):
    """Initialize DC removal operation.

    Parameters
    ----------
    sampling_rate : float
        Sampling rate (Hz)
    """
    super().__init__(sampling_rate)
    logger.debug("Initialized RemoveDC operation")

calculate_output_dtype(input_dtype, *input_dtypes)

Source code in wandas/processing/effects.py

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

AddWithSNR

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Addition operation considering SNR

Source code in wandas/processing/effects.py

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

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

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

        Parameters
        ----------
        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

sampling_rate : float Sampling rate (Hz) snr : float Signal-to-noise ratio (dB)

Source code in wandas/processing/effects.py

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

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

calculate_output_dtype(input_dtype, *input_dtypes)

Promote SNR mixing to at least float32 precision.

Source code in wandas/processing/effects.py

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

Fade

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Fade operation using a Tukey (tapered cosine) window.

This operation applies symmetric fade-in and fade-out with the same duration. The Tukey window alpha parameter is computed from the fade duration so that the tapered portion equals the requested fade length at each end.

Source code in wandas/processing/effects.py

class Fade(AudioOperation[NDArrayReal, NDArrayReal]):
    """Fade operation using a Tukey (tapered cosine) window.

    This operation applies symmetric fade-in and fade-out with the same
    duration. The Tukey window alpha parameter is computed from the fade
    duration so that the tapered portion equals the requested fade length
    at each end.
    """

    name = "fade"
    _display = "fade"

    def __init__(self, sampling_rate: float, fade_ms: float = 50) -> None:
        fade_ms = float(fade_ms)
        super().__init__(sampling_rate, fade_ms=fade_ms)

    @property
    def fade_ms(self) -> float:
        """Fade duration captured at operation construction time."""
        return self._config_value("fade_ms")

    def validate_params(self) -> None:
        if self.fade_ms < 0:
            raise ValueError("fade_ms must be non-negative")

    def _fade_len_for_sampling_rate(self) -> int:
        return round(self.fade_ms * self.sampling_rate / 1000.0)

    @staticmethod
    def calculate_tukey_alpha(fade_len: int, n_samples: int) -> float:
        """Calculate Tukey window alpha parameter from fade length.

        The alpha parameter determines what fraction of the window is tapered.
        For symmetric fade-in/fade-out, alpha = 2 * fade_len / n_samples ensures
        that each side's taper has exactly fade_len samples.

        Parameters
        ----------
        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

def __init__(self, sampling_rate: float, fade_ms: float = 50) -> None:
    fade_ms = float(fade_ms)
    super().__init__(sampling_rate, fade_ms=fade_ms)

validate_params()

Source code in wandas/processing/effects.py

def validate_params(self) -> None:
    if self.fade_ms < 0:
        raise ValueError("fade_ms must be non-negative")

calculate_tukey_alpha(fade_len, n_samples) staticmethod

Calculate Tukey window alpha parameter from fade length.

The alpha parameter determines what fraction of the window is tapered. For symmetric fade-in/fade-out, alpha = 2 * fade_len / n_samples ensures that each side's taper has exactly fade_len samples.

Parameters

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

Source code in wandas/processing/effects.py

@staticmethod
def calculate_tukey_alpha(fade_len: int, n_samples: int) -> float:
    """Calculate Tukey window alpha parameter from fade length.

    The alpha parameter determines what fraction of the window is tapered.
    For symmetric fade-in/fade-out, alpha = 2 * fade_len / n_samples ensures
    that each side's taper has exactly fade_len samples.

    Parameters
    ----------
    fade_len : int
        Desired fade length in samples for each end (in and out).
    n_samples : int
        Total number of samples in the signal.

    Returns
    -------
    float
        Alpha parameter for scipy.signal.windows.tukey, clamped to [0, 1].

    Examples
    --------
    >>> Fade.calculate_tukey_alpha(fade_len=20, n_samples=200)
    0.2
    >>> Fade.calculate_tukey_alpha(fade_len=100, n_samples=100)
    1.0
    """
    alpha = float(2 * fade_len) / float(n_samples)
    return min(1.0, alpha)

calculate_output_dtype(input_dtype, *input_dtypes)

Source code in wandas/processing/effects.py

def calculate_output_dtype(self, input_dtype: np.dtype[Any], *input_dtypes: np.dtype[Any]) -> np.dtype[Any]:
    del input_dtypes
    if self._fade_len_for_sampling_rate() <= 0:
        return np.dtype(input_dtype)
    return np.result_type(input_dtype, np.float64)

Functions

Modules

Filters / フィルター

Provides various audio filter processing. 様々なオーディオフィルター処理を提供します。

wandas.processing.filters

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

HighPassFilter

Bases: _ButterworthFilter

High-pass filter operation

Source code in wandas/processing/filters.py

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

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

Attributes

name = 'highpass_filter' class-attribute instance-attribute

LowPassFilter

Bases: _ButterworthFilter

Low-pass filter operation

Source code in wandas/processing/filters.py

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

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

Attributes

name = 'lowpass_filter' class-attribute instance-attribute

BandPassFilter

Bases: _ButterworthFilter

Band-pass filter operation

Source code in wandas/processing/filters.py

class BandPassFilter(_ButterworthFilter):
    """Band-pass filter operation"""

    name = "bandpass_filter"
    _btype = "band"
    _display = "bpf"

    def __init__(
        self,
        sampling_rate: float,
        low_cutoff: float,
        high_cutoff: float,
        order: int = 4,
    ):
        """
        Initialize band-pass filter

        Parameters
        ----------
        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

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

Source code in wandas/processing/filters.py

def __init__(
    self,
    sampling_rate: float,
    low_cutoff: float,
    high_cutoff: float,
    order: int = 4,
):
    """
    Initialize band-pass filter

    Parameters
    ----------
    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

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: AudioOperation[NDArrayReal, NDArrayReal]

A-weighting filter operation

Source code in wandas/processing/filters.py

class AWeighting(AudioOperation[NDArrayReal, NDArrayReal]):
    """A-weighting filter operation"""

    name = "a_weighting"
    _display = "Aw"

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

        Parameters
        ----------
        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

sampling_rate : float Sampling rate (Hz)

Source code in wandas/processing/filters.py

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

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

calculate_output_dtype(input_dtype, *input_dtypes)

Source code in wandas/processing/filters.py

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

Functions

Spectral Processing / スペクトル処理

Provides spectral analysis and processing capabilities. スペクトル解析と処理機能を提供します。

wandas.processing.spectral

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

FFT

Bases: AudioOperation[NDArrayReal, NDArrayComplex]

FFT (Fast Fourier Transform) operation

Source code in wandas/processing/spectral.py

class FFT(AudioOperation[NDArrayReal, NDArrayComplex]):
    """FFT (Fast Fourier Transform) operation"""

    name = "fft"
    _display = "FFT"

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

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

        Parameters
        ----------
        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."""
        from scipy.signal import get_window

        n_fft = self.n_fft
        if n_fft is not None and x.shape[-1] > n_fft:
            # If n_fft is specified and input length exceeds it, truncate
            x = x[..., :n_fft]

        win = get_window(self.window, x.shape[-1])
        x = x * win
        fft_size = int(x.shape[-1]) if n_fft is None else n_fft
        result: NDArrayComplex = np.fft.rfft(x, n=fft_size, axis=-1)
        # Window function scaling correction
        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

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

Source code in wandas/processing/spectral.py

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

    Parameters
    ----------
    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

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

Returns

tuple Output data shape (channels, freqs).

Source code in wandas/processing/spectral.py

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

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

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

calculate_output_dtype(input_dtype, *input_dtypes)

Source code in wandas/processing/spectral.py

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

IFFT

Bases: AudioOperation[NDArrayComplex, NDArrayReal]

IFFT (Inverse Fast Fourier Transform) operation

Source code in wandas/processing/spectral.py

class IFFT(AudioOperation[NDArrayComplex, NDArrayReal]):
    """IFFT (Inverse Fast Fourier Transform) operation"""

    name = "ifft"
    _display = "iFFT"

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

        Parameters
        ----------
        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

        Parameters
        ----------
        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:
        """Create processor function for IFFT operation"""
        logger.debug(f"Applying IFFT to array with shape: {x.shape}")

        # Restore frequency component scaling (remove the 2.0 multiplier applied in FFT)
        fft_size = 2 * (int(x.shape[-1]) - 1) if self.n_fft is None else self.n_fft
        _x = _denormalize_rfft_amplitude(
            x,
            n_fft=fft_size,
            window_gain=1.0,
        )

        # Execute IFFT
        result: NDArrayReal = np.fft.irfft(_x, n=self.n_fft, axis=-1)

        # Window function correction (inverse of FFT operation)
        from scipy.signal import get_window

        win = get_window(self.window, result.shape[-1])

        # Correct the FFT window function scaling
        scaling_factor = np.sum(win) / result.shape[-1]
        result = result / scaling_factor

        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

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'

Source code in wandas/processing/spectral.py

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

    Parameters
    ----------
    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

input_shape : tuple Input data shape (channels, freqs)

Returns

tuple Output data shape (channels, samples)

Source code in wandas/processing/spectral.py

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

    Parameters
    ----------
    input_shape : tuple
        Input data shape (channels, freqs)

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

calculate_output_dtype(input_dtype, *input_dtypes)

Source code in wandas/processing/spectral.py

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

STFT

Bases: AudioOperation[NDArrayReal, NDArrayComplex]

Short-Time Fourier Transform operation

Source code in wandas/processing/spectral.py

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

    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

        Parameters
        ----------
        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

        Parameters
        ----------
        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

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

Source code in wandas/processing/spectral.py

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

    Parameters
    ----------
    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

input_shape : tuple Input data shape

Returns

tuple Output data shape

Source code in wandas/processing/spectral.py

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

    Parameters
    ----------
    input_shape : tuple
        Input data shape

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

calculate_output_dtype(input_dtype, *input_dtypes)

Source code in wandas/processing/spectral.py

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

ISTFT

Bases: AudioOperation[NDArrayComplex, NDArrayReal]

Inverse Short-Time Fourier Transform operation

Source code in wandas/processing/spectral.py

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

    name = "istft"
    _display = "iSTFT"

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

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

        Parameters
        ----------
        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

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

Source code in wandas/processing/spectral.py

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

    Parameters
    ----------
    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

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

Source code in wandas/processing/spectral.py

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

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

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

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

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

    .. math::

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

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

    The output length is then:

    .. math::

        output\\_samples = k_1 - k_0

    where k0 defaults to 0 and k1 defaults to k_max.

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

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

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

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

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

    output_samples = k1 - k0

    return (n_channels, output_samples)

calculate_output_dtype(input_dtype, *input_dtypes)

Source code in wandas/processing/spectral.py

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

process(data, *inputs)

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

Source code in wandas/processing/spectral.py

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

Welch

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Welch method for power spectral density estimation.

Computes the one-sided amplitude spectrum using Welch's method for consistency with FFT and STFT methods. For a sine wave with amplitude A, the peak value at its frequency will be approximately A.

Notes

Internally uses scipy.signal.welch with scaling='spectrum' and converts the power spectrum to amplitude spectrum: - DC component (f=0): A = sqrt(P) - AC components (f>0): A = sqrt(2*P)

Source code in wandas/processing/spectral.py

class Welch(AudioOperation[NDArrayReal, NDArrayReal]):
    """Welch method for power spectral density estimation.

    Computes the one-sided amplitude spectrum using Welch's method for
    consistency with FFT and STFT methods. For a sine wave with amplitude A,
    the peak value at its frequency will be approximately A.

    Notes
    -----
    Internally uses scipy.signal.welch with scaling='spectrum' and converts
    the power spectrum to amplitude spectrum:
    - DC component (f=0): A = sqrt(P)
    - AC components (f>0): 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

        Parameters
        ----------
        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

        Parameters
        ----------
        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:
        """Create processor function for Welch operation.

        Converts power spectrum from scipy.signal.welch to one-sided
        amplitude spectrum 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)  # Convert to amplitude
        result[..., 1:-1] *= np.sqrt(2)  # Apply factor of sqrt(2) for AC components

        return np.array(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

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

Source code in wandas/processing/spectral.py

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

    Parameters
    ----------
    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

input_shape : tuple Input data shape (channels, samples)

Returns

tuple Output data shape (channels, freqs)

Source code in wandas/processing/spectral.py

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

    Parameters
    ----------
    input_shape : tuple
        Input data shape (channels, samples)

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

calculate_output_dtype(input_dtype, *input_dtypes)

Source code in wandas/processing/spectral.py

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

NOctSpectrum

Bases: _NOctBase

N-octave spectrum operation

Source code in wandas/processing/spectral.py

class NOctSpectrum(_NOctBase):
    """N-octave spectrum operation"""

    name = "noct_spectrum"
    _display = "Oct"

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Create processor function for octave spectrum"""
        logger.debug(f"Applying NoctSpectrum to array with shape: {x.shape}")
        spec, _ = 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.expand_dims(spec, axis=0) if spec.ndim == 1 else spec.T
        logger.debug(f"NoctSpectrum applied, returning result with shape: {spec.shape}")
        return np.array(spec)

Attributes

name = 'noct_spectrum' class-attribute instance-attribute

NOctSynthesis

Bases: _NOctBase

Octave synthesis operation

Source code in wandas/processing/spectral.py

class NOctSynthesis(_NOctBase):
    """Octave synthesis operation"""

    name = "noct_synthesis"
    _display = "Octs"

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Create processor function for octave synthesis"""
        logger.debug(f"Applying NoctSynthesis to array with shape: {x.shape}")
        # Calculate n from shape[-1]
        n = x.shape[-1]  # Calculate n from shape[-1]
        n = n * 2 - 1 if n % 2 == 0 else (n - 1) * 2
        freqs = np.fft.rfftfreq(n, d=1 / self.sampling_rate)
        result, _ = noct_synthesis(
            spectrum=np.abs(x).T,
            freqs=freqs,
            fmin=self.fmin,
            fmax=self.fmax,
            n=self.n,
            G=self.G,
            fr=self.fr,
        )
        result = result.T
        logger.debug(f"NoctSynthesis applied, returning result with shape: {result.shape}")
        return np.array(result)

Attributes

name = 'noct_synthesis' class-attribute instance-attribute

Coherence

Bases: _CrossSpectralBase

Coherence estimation operation

Source code in wandas/processing/spectral.py

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

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

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Processor function for coherence estimation operation"""
        logger.debug(f"Applying coherence estimation to array with shape: {x.shape}")
        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,
        )

        # Reshape result to (n_channels * n_channels, n_freqs)
        result: NDArrayReal = coh.transpose(1, 0, 2).reshape(-1, coh.shape[-1])

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

Attributes

name = 'coherence' class-attribute instance-attribute

CSD

Bases: _ScaledCrossSpectralBase

Cross-spectral density estimation operation

Source code in wandas/processing/spectral.py

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

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

    def _process(self, x: NDArrayReal) -> NDArrayComplex:
        """Processor function for cross-spectral density estimation operation"""
        logger.debug(f"Applying CSD estimation to array with shape: {x.shape}")
        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,
        )

        # Reshape result to (n_channels * n_channels, n_freqs)
        result: NDArrayComplex = csd_result.transpose(1, 0, 2).reshape(-1, csd_result.shape[-1])

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

Attributes

name = 'csd' class-attribute instance-attribute

TransferFunction

Bases: _ScaledCrossSpectralBase

Transfer function estimation operation

Source code in wandas/processing/spectral.py

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

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

    def _process(self, x: NDArrayReal) -> NDArrayComplex:
        """Processor function for transfer function estimation operation"""
        logger.debug(f"Applying transfer function estimation to array with shape: {x.shape}")
        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: (num_channels, num_channels, num_frequencies)

        # 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 transfer function H(f) = P_yx / P_xx
        h_f = p_yx / p_xx[np.newaxis, :, :]
        result: NDArrayComplex = h_f.transpose(1, 0, 2).reshape(-1, h_f.shape[-1])

        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

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

noct_synthesis(*args, **kwargs)

Source code in wandas/processing/spectral.py

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

Cepstral Processing / ケプストラム処理

Provides real-cepstrum analysis, symmetric liftering, and spectral-envelope reconstruction. Most users should use the typed Frame methods described in the cepstral analysis guide. 実ケプストラム解析、対称リフタリング、スペクトル包絡再構成を提供します。通常は ケプストラム解析ガイドの型付きFrameメソッドを利用してください。

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

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.

Source code in wandas/processing/cepstral.py

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

    Parameters
    ----------
    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)
        window_values = get_window(self.window, analysis.shape[-1])
        window_gain = float(np.sum(window_values))
        if not np.isfinite(window_gain) or window_gain == 0:
            raise ValueError(
                "Invalid window gain for cepstrum\n"
                f"  Window: {self.window!r}\n"
                f"  Gain: {window_gain}\n"
                "Use a window with a finite non-zero coherent gain."
            )
        spectrum = np.fft.rfft(analysis * window_values, n=n_fft, axis=-1)
        normalized_spectrum = _normalize_rfft_amplitude(
            spectrum,
            n_fft=n_fft,
            window_gain=window_gain,
        )
        magnitude = np.abs(normalized_spectrum)
        log_magnitude = np.log(np.maximum(magnitude, self.floor))
        return np.asarray(np.fft.irfft(log_magnitude, n=n_fft, axis=-1), dtype=np.float64)

Attributes

name = 'cepstrum' class-attribute instance-attribute

n_fft property

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

window property

Return the configured analysis-window name.

floor property

Return the positive log-magnitude floor.

Functions

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

Source code in wandas/processing/cepstral.py

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

calculate_output_shape(input_shape)

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

Source code in wandas/processing/cepstral.py

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

calculate_output_dtype(input_dtype, *input_dtypes)

Return NumPy FFT's real output dtype.

Source code in wandas/processing/cepstral.py

def calculate_output_dtype(
    self,
    input_dtype: np.dtype[Any],
    *input_dtypes: np.dtype[Any],
) -> np.dtype[Any]:
    """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

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.

Source code in wandas/processing/cepstral.py

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

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

    Parameters
    ----------
    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

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

calculate_output_shape(input_shape)

Replace the frequency axis with a complete quefrency axis.

Source code in wandas/processing/cepstral.py

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

calculate_output_dtype(input_dtype, *input_dtypes)

Return NumPy FFT's real output dtype.

Source code in wandas/processing/cepstral.py

def calculate_output_dtype(
    self,
    input_dtype: np.dtype[Any],
    *input_dtypes: np.dtype[Any],
) -> np.dtype[Any]:
    """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

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.

Source code in wandas/processing/cepstral.py

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

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

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

    name = "lifter"
    _display = "lifter"

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

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

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

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

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

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

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

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

Attributes

name = 'lifter' class-attribute instance-attribute

cutoff property

Return the quefrency cutoff in seconds.

mode property

Return the selected low- or high-quefrency mode.

axis property

Return the configured quefrency axis.

Functions

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

Source code in wandas/processing/cepstral.py

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

calculate_output_dtype(input_dtype, *input_dtypes)

Preserve a real floating input dtype.

Source code in wandas/processing/cepstral.py

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

calculate_output_shape(input_shape)

Validate the cutoff against the known cepstrum length.

Source code in wandas/processing/cepstral.py

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

SpectralEnvelope

Bases: AudioOperation[NDArrayReal, NDArrayComplex]

Reconstruct a normalized one-sided spectral envelope.

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

Parameters

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.

Source code in wandas/processing/cepstral.py

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

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

    Parameters
    ----------
    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

def __init__(self, sampling_rate: float, *, axis: int = -1) -> None:
    normalized_axis = _normalize_transform_axis(
        axis,
        operation_name="spectral envelope",
    )
    super().__init__(sampling_rate, axis=normalized_axis)

calculate_output_shape(input_shape)

Replace the quefrency axis with its one-sided frequency axis.

Source code in wandas/processing/cepstral.py

def calculate_output_shape(self, input_shape: tuple[int, ...]) -> tuple[int, ...]:
    """Replace the quefrency axis with its one-sided frequency axis."""
    axis = _resolve_transform_axis(
        self.axis,
        len(input_shape),
        operation_name="spectral envelope",
    )
    output_shape = list(input_shape)
    output_shape[axis] = int(input_shape[axis]) // 2 + 1
    return tuple(output_shape)

calculate_output_dtype(input_dtype, *input_dtypes)

Return the complex dtype used by SpectralFrame.

Source code in wandas/processing/cepstral.py

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

Functions

Statistical Processing / 統計処理

Provides statistical analysis functions for audio data. オーディオデータの統計分析機能を提供します。

wandas.processing.stats

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

ABS

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Absolute value operation

Source code in wandas/processing/stats.py

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

    name = "abs"
    _display = "abs"

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

        Parameters
        ----------
        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

sampling_rate : float Sampling rate (Hz)

Source code in wandas/processing/stats.py

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

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

process(data, *inputs)

Source code in wandas/processing/stats.py

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

Power

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Power operation

Source code in wandas/processing/stats.py

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

    name = "power"
    _display = "pow"

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

        Parameters
        ----------
        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

sampling_rate : float Sampling rate (Hz) exponent : float Power exponent

Source code in wandas/processing/stats.py

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

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

process(data, *inputs)

Source code in wandas/processing/stats.py

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

Sum

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Sum calculation

Source code in wandas/processing/stats.py

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

    name = "sum"
    _display = "sum"

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

Attributes

name = 'sum' class-attribute instance-attribute

Functions

process(data, *inputs)

Source code in wandas/processing/stats.py

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

Mean

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Mean calculation

Source code in wandas/processing/stats.py

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

    name = "mean"
    _display = "mean"

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

Attributes

name = 'mean' class-attribute instance-attribute

Functions

process(data, *inputs)

Source code in wandas/processing/stats.py

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

ChannelDifference

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Channel difference calculation operation

Source code in wandas/processing/stats.py

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

    name = "channel_difference"
    _display = "diff"

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

        Parameters
        ----------
        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

sampling_rate : float Sampling rate (Hz) other_channel : int Channel to calculate difference with, default is 0

Source code in wandas/processing/stats.py

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

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

process(data, *inputs)

Source code in wandas/processing/stats.py

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

Functions

Temporal Processing / 時間領域処理

Provides time-domain processing capabilities. 時間領域の処理機能を提供します。

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: AudioOperation[NDArrayReal, NDArrayReal]

Resampling operation

Source code in wandas/processing/temporal.py

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

    name = "resampling"
    _display = "rs"

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

        Parameters
        ----------
        sampling_rate : float
            Sampling rate (Hz)
        target_sampling_rate : float
            Target sampling rate (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 new sampling rate

        Notes
        -----
        Resampling always produces output at target_sr, regardless of input
        sampling rate. All necessary parameters are provided at initialization.
        """
        return {"sampling_rate": self.target_sr}

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

        Parameters
        ----------
        input_shape : tuple
            Input data shape

        Returns
        -------
        tuple
            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 resampling operation

Parameters

sampling_rate : float Sampling rate (Hz) target_sampling_rate : float Target sampling rate (Hz)

Raises

ValueError If sampling_rate or target_sr is not positive

Source code in wandas/processing/temporal.py

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

    Parameters
    ----------
    sampling_rate : float
        Sampling rate (Hz)
    target_sampling_rate : float
        Target sampling rate (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

dict Metadata updates with new sampling rate

Notes

Resampling always produces output at target_sr, regardless of input sampling rate. All necessary parameters are provided at initialization.

Source code in wandas/processing/temporal.py

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

    Returns
    -------
    dict
        Metadata updates with new sampling rate

    Notes
    -----
    Resampling always produces output at target_sr, regardless of input
    sampling rate. All necessary parameters are provided at initialization.
    """
    return {"sampling_rate": self.target_sr}

calculate_output_shape(input_shape)

Calculate output data shape after operation

Parameters

input_shape : tuple Input data shape

Returns

tuple Output data shape

Source code in wandas/processing/temporal.py

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

    Parameters
    ----------
    input_shape : tuple
        Input data shape

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

calculate_output_dtype(input_dtype, *input_dtypes)

Return resampling output dtype metadata.

Source code in wandas/processing/temporal.py

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

Trim

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Trimming operation

Source code in wandas/processing/temporal.py

class Trim(AudioOperation[NDArrayReal, NDArrayReal]):
    """Trimming operation"""

    name = "trim"
    _display = "trim"

    def __init__(
        self,
        sampling_rate: float,
        start: float,
        end: float,
    ):
        """
        Initialize trimming operation

        Parameters
        ----------
        sampling_rate : float
            Sampling rate (Hz)
        start : float
            Start time for trimming (seconds)
        end : float
            End time for trimming (seconds)
        """
        super().__init__(sampling_rate, start=start, end=end)
        logger.debug(f"Initialized Trim operation with 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, ...]:
        """
        Calculate output data shape after operation

        Parameters
        ----------
        input_shape : tuple
            Input data shape

        Returns
        -------
        tuple
            Output data shape
        """
        # Calculate length after trimming
        # Exclude parts where there is no signal
        end_sample = min(self.end_sample, input_shape[-1])
        n_samples = end_sample - self.start_sample
        return (*input_shape[:-1], n_samples)

    def _process(self, x: NDArrayReal) -> NDArrayReal:
        """Create processor function for trimming operation"""
        logger.debug(f"Applying trim to array with shape: {x.shape}")
        # Apply trimming
        result = x[..., self.start_sample : self.end_sample]
        logger.debug(f"Trim applied, returning result with shape: {result.shape}")
        return result

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)

Initialize trimming operation

Parameters

sampling_rate : float Sampling rate (Hz) start : float Start time for trimming (seconds) end : float End time for trimming (seconds)

Source code in wandas/processing/temporal.py

def __init__(
    self,
    sampling_rate: float,
    start: float,
    end: float,
):
    """
    Initialize trimming operation

    Parameters
    ----------
    sampling_rate : float
        Sampling rate (Hz)
    start : float
        Start time for trimming (seconds)
    end : float
        End time for trimming (seconds)
    """
    super().__init__(sampling_rate, start=start, end=end)
    logger.debug(f"Initialized Trim operation with start: {start}, end: {end}")

calculate_output_shape(input_shape)

Calculate output data shape after operation

Parameters

input_shape : tuple Input data shape

Returns

tuple Output data shape

Source code in wandas/processing/temporal.py

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

    Parameters
    ----------
    input_shape : tuple
        Input data shape

    Returns
    -------
    tuple
        Output data shape
    """
    # Calculate length after trimming
    # Exclude parts where there is no signal
    end_sample = min(self.end_sample, input_shape[-1])
    n_samples = end_sample - self.start_sample
    return (*input_shape[:-1], n_samples)

FixLength

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Operation to adjust signal length to a specified length.

Source code in wandas/processing/temporal.py

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

    name = "fix_length"
    _display = "fix"

    def __init__(
        self,
        sampling_rate: float,
        length: int | None = None,
        duration: float | None = None,
    ):
        """
        Initialize fix length operation

        Parameters
        ----------
        sampling_rate : float
            Sampling rate (Hz)
        length : Optional[int]
            Target length for fixing
        duration : Optional[float]
            Target length for fixing
        """
        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, ...]:
        """
        Calculate output data shape after operation

        Parameters
        ----------
        input_shape : tuple
            Input data shape

        Returns
        -------
        tuple
            Output data shape
        """
        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 fix length operation

Parameters

sampling_rate : float Sampling rate (Hz) length : Optional[int] Target length for fixing duration : Optional[float] Target length for fixing

Source code in wandas/processing/temporal.py

def __init__(
    self,
    sampling_rate: float,
    length: int | None = None,
    duration: float | None = None,
):
    """
    Initialize fix length operation

    Parameters
    ----------
    sampling_rate : float
        Sampling rate (Hz)
    length : Optional[int]
        Target length for fixing
    duration : Optional[float]
        Target length for fixing
    """
    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)

Calculate output data shape after operation

Parameters

input_shape : tuple Input data shape

Returns

tuple Output data shape

Source code in wandas/processing/temporal.py

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

    Parameters
    ----------
    input_shape : tuple
        Input data shape

    Returns
    -------
    tuple
        Output data shape
    """
    return (*input_shape[:-1], self.target_length)

RmsTrend

Bases: AudioOperation[NDArrayReal, NDArrayReal]

RMS calculation

Source code in wandas/processing/temporal.py

class RmsTrend(AudioOperation[NDArrayReal, NDArrayReal]):
    """RMS calculation"""

    name = "rms_trend"
    _display = "RMS"

    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,
    ) -> None:
        """
        Initialize RMS calculation

        Parameters
        ----------
        sampling_rate : float
            Sampling rate (Hz)
        frame_length : int
            Frame length, default is 2048
        hop_length : int
            Hop length, default is 512
        ref : Union[list[float], float]
            Reference value(s) for dB calculation
        dB : bool
            Whether to convert to decibels
        Aw : bool
            Whether to apply A-weighting before RMS calculation
        """
        ref_array = np.array(ref if isinstance(ref, list) else [ref])
        super().__init__(
            sampling_rate,
            frame_length=frame_length,
            hop_length=hop_length,
            dB=dB,
            Aw=Aw,
            ref=ref_array,
        )

    @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 get_metadata_updates(self) -> dict[str, Any]:
        """
        Update sampling rate based on hop length.

        Returns
        -------
        dict
            Metadata updates with new sampling rate based on hop length

        Notes
        -----
        The output sampling rate is determined by downsampling the input
        by hop_length. All necessary parameters are provided at initialization.
        """
        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 output data shape after operation

        Parameters
        ----------
        input_shape : tuple
            Input data shape (channels, samples)

        Returns
        -------
        tuple
            Output data shape (channels, frames)
        """
        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}")

        if self.Aw:
            # Apply A-weighting
            _x = A_weight(x, self.sampling_rate)
            if isinstance(_x, np.ndarray):
                # Use the first element if A_weight returns a tuple
                x = _x
            elif isinstance(_x, tuple):
                # Use the first element if A_weight returns a tuple
                x = _x[0]
            else:
                raise ValueError("A_weighting returned an unexpected type.")

        # Calculate RMS
        result: NDArrayReal = _frame_rms(
            x,
            frame_length=self.frame_length,
            hop_length=self.hop_length,
        )

        if self.dB:
            # Convert to dB
            result = 20 * np.log10(np.maximum(result / self._config["ref"][..., np.newaxis], DB_FLOOR))
        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)

Initialize RMS calculation

Parameters

sampling_rate : float Sampling rate (Hz) frame_length : int Frame length, default is 2048 hop_length : int Hop length, default is 512 ref : Union[list[float], float] Reference value(s) for dB calculation dB : bool Whether to convert to decibels Aw : bool Whether to apply A-weighting before RMS calculation

Source code in wandas/processing/temporal.py

def __init__(
    self,
    sampling_rate: float,
    frame_length: int = 2048,
    hop_length: int = 512,
    ref: list[float] | float = 1.0,
    dB: bool = False,
    Aw: bool = False,
) -> None:
    """
    Initialize RMS calculation

    Parameters
    ----------
    sampling_rate : float
        Sampling rate (Hz)
    frame_length : int
        Frame length, default is 2048
    hop_length : int
        Hop length, default is 512
    ref : Union[list[float], float]
        Reference value(s) for dB calculation
    dB : bool
        Whether to convert to decibels
    Aw : bool
        Whether to apply A-weighting before RMS calculation
    """
    ref_array = np.array(ref if isinstance(ref, list) else [ref])
    super().__init__(
        sampling_rate,
        frame_length=frame_length,
        hop_length=hop_length,
        dB=dB,
        Aw=Aw,
        ref=ref_array,
    )

get_metadata_updates()

Update sampling rate based on hop length.

Returns

dict Metadata updates with new sampling rate based on hop length

Notes

The output sampling rate is determined by downsampling the input by hop_length. All necessary parameters are provided at initialization.

Source code in wandas/processing/temporal.py

def get_metadata_updates(self) -> dict[str, Any]:
    """
    Update sampling rate based on hop length.

    Returns
    -------
    dict
        Metadata updates with new sampling rate based on hop length

    Notes
    -----
    The output sampling rate is determined by downsampling the input
    by hop_length. All necessary parameters are provided at initialization.
    """
    new_sr = self.sampling_rate / self.hop_length
    return {"sampling_rate": new_sr}

calculate_output_shape(input_shape)

Calculate output data shape after operation

Parameters

input_shape : tuple Input data shape (channels, samples)

Returns

tuple Output data shape (channels, frames)

Source code in wandas/processing/temporal.py

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

    Parameters
    ----------
    input_shape : tuple
        Input data shape (channels, samples)

    Returns
    -------
    tuple
        Output data shape (channels, frames)
    """
    n_frames = _centered_frame_count(
        input_shape[-1],
        self.frame_length,
        self.hop_length,
    )
    return (*input_shape[:-1], n_frames)

calculate_output_dtype(input_dtype, *input_dtypes)

Return RMS trend output dtype metadata.

Source code in wandas/processing/temporal.py

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

SoundLevel

Bases: AudioOperation[NDArrayReal, NDArrayReal]

Time-weighted RMS or sound level with frequency and time weighting.

Source code in wandas/processing/temporal.py

class SoundLevel(AudioOperation[NDArrayReal, NDArrayReal]):
    """Time-weighted RMS or sound level with frequency and time weighting."""

    name = "sound_level"

    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,
    ) -> None:
        validate_sampling_rate(sampling_rate)
        ref_array = np.atleast_1d(np.array(ref, dtype=float, copy=True))
        if np.any(ref_array <= 0):
            raise ValueError(
                "Invalid sound level reference\n"
                f"  Got: {ref_array.tolist()}\n"
                "  Expected: Positive reference values\n"
                "Sound pressure level requires a positive reference pressure."
            )
        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,
        )

    @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 a supported IEC-style weighting curve before calculating sound level."
            )
        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 a supported sound level meter time constant before calculating sound level."
        )

    @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_squared(self, n_channels: int) -> NDArrayReal:
        """Return squared reference pressure 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(np.square(ref), dtype=np.float64)

    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)
        freq_weighting = self.freq_weighting
        if freq_weighting == "Z":
            weighted = weighted_input
        else:
            weighted = frequency_weight(weighted_input, self.sampling_rate, curve=freq_weighting)
        squared = np.square(weighted)
        alpha = np.asarray(np.exp(-1.0 / (self.sampling_rate * self.time_constant)), dtype=np.float64).item()
        smoothed = lfilter([1.0 - alpha], [1.0, -alpha], squared, axis=-1)
        if self.dB:
            ref_squared_broadcast = self._reference_squared(smoothed.shape[0])[:, np.newaxis]
            result = 10.0 * np.log10(np.maximum(smoothed / ref_squared_broadcast, MIN_SOUND_LEVEL_POWER_RATIO))
        else:
            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)

Source code in wandas/processing/temporal.py

def __init__(
    self,
    sampling_rate: float,
    ref: list[float] | float | NDArrayReal = 1.0,
    freq_weighting: str | None = "Z",
    time_weighting: str = "Fast",
    dB: bool = False,
) -> None:
    validate_sampling_rate(sampling_rate)
    ref_array = np.atleast_1d(np.array(ref, dtype=float, copy=True))
    if np.any(ref_array <= 0):
        raise ValueError(
            "Invalid sound level reference\n"
            f"  Got: {ref_array.tolist()}\n"
            "  Expected: Positive reference values\n"
            "Sound pressure level requires a positive reference pressure."
        )
    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,
    )

get_display_name()

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

Source code in wandas/processing/temporal.py

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

calculate_output_dtype(input_dtype, *input_dtypes)

Return sound level output dtype metadata.

Source code in wandas/processing/temporal.py

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

Functions