"""Signal post-processing effects.
This module provides :class:`Effect` subclasses that can be applied to
complex sensor-signal arrays after synthesis but before beamforming or
output. Each effect operates independently on each sensor channel.
Current implementations:
- :class:`Reverb` — convolutional reverb via a synthetic exponentially-decaying
impulse response, controlled by ``duration_s`` and ``wet_dry_mix``.
"""
from abc import ABC, abstractmethod
import numpy as np
from stonesoup.base import Base, Property
from .._seed import _spawn_rng
from .base import ComplexArray
[docs]
class Effect(Base, ABC):
"""A base class for all signal post-processing effects."""
[docs]
@abstractmethod
def apply(self, signals: ComplexArray, sampling_rate_hz: int) -> ComplexArray:
"""Apply the effect to the signal. Must be implemented by subclasses."""
...
[docs]
class Reverb(Effect):
"""Applies a simple convolutional reverb effect to the signal."""
duration_s: float = Property(default=1.0, doc="The decay time of the reverb tail in seconds.")
wet_dry_mix: float = Property(
default=0.3,
doc="Mix between wet (reverb) and dry signal (0=dry, 1=wet).",
)
seed: int | None = Property(
default=None,
doc="Seed for the impulse-response RNG. ``None`` defers to the global seed set by "
"``bluepebble.set_seed()`` (different reproducible tail per call) or gives a "
"non-deterministic tail if no global seed is set; an explicit integer produces the "
"same tail for the same ``duration_s`` and ``sampling_rate_hz`` across calls and runs.",
)
[docs]
def apply(self, signals: ComplexArray, sampling_rate_hz: int) -> ComplexArray:
"""Apply a simple convolutional reverb effect to the signal.
Parameters
----------
signals : ComplexArray
An array of complex signals with shape (num_sensors, num_samples).
sampling_rate_hz : int
The sampling rate in Hertz.
Returns
-------
ComplexArray
An array of complex signals with the reverb effect applied,
with shape (num_sensors, num_samples).
Raises
------
ValueError
If ``duration_s`` is not positive or ``wet_dry_mix`` is not in ``[0, 1]``.
"""
if self.duration_s <= 0:
msg = f"duration_s must be positive, got {self.duration_s}"
raise ValueError(msg)
if not 0 <= self.wet_dry_mix <= 1:
msg = f"wet_dry_mix must be in [0, 1], got {self.wet_dry_mix}"
raise ValueError(msg)
# Generate a synthetic Impulse Response (IR) for the reverb effect.
# Re-create the RNG from seed each call so the same seed always produces the same IR,
# regardless of how many times apply() has been called previously.
rng = _spawn_rng(self.seed)
ir_samples = int(self.duration_s * sampling_rate_hz)
time = np.arange(ir_samples) / sampling_rate_hz
decay = np.exp(-5.0 * time / self.duration_s) # Exponential decay
ir = decay * rng.standard_normal(ir_samples) # White noise modulated by decay
ir_max = np.max(np.abs(ir))
if ir_max > 0:
ir /= ir_max # Normalise the IR
# Apply reverb to each sensor channel independently
reverbed_signals = np.copy(signals)
for i in range(signals.shape[0]):
dry_signal = signals[i, :]
wet_signal = np.convolve(dry_signal, ir, mode="same")
reverbed_signals[i, :] = (
1 - self.wet_dry_mix
) * dry_signal + self.wet_dry_mix * wet_signal
return reverbed_signals