Repository: claritychallenge/clarity (language: python)

Issue:
Multiband Dynamic Range Compressor

Update PR #374 from main

Multiband Dynamic Range Compressor



Discussion / hints:

Fix patch:
diff --git a/clarity/enhancer/multiband_compressor/__init__.py b/clarity/enhancer/multiband_compressor/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/clarity/enhancer/multiband_compressor/compressor_qmul.py b/clarity/enhancer/multiband_compressor/compressor_qmul.py
new file mode 100644
index 000000000..acbdc8b9e
--- /dev/null
+++ b/clarity/enhancer/multiband_compressor/compressor_qmul.py
@@ -0,0 +1,169 @@
+""""Module for the Compressor class."""
+
+from __future__ import annotations
+
+import warnings
+
+import numpy as np
+
+
+class Compressor:
+    """Compressor
+    Based in the compressor from [1].
+    Code adapted from JUCE C++ source code.
+    Optimization using IIR filters.
+
+    References:
+    [1] Giannoulis, D., Massberg, M., & Reiss, J. D. (2012).
+    Digital dynamic range compressor design - A tutorial and analysis.
+    Journal of the Audio Engineering Society, 60(6), 399-408.
+
+    Example:
+    >>> import librosa
+    >>> import matplotlib.pyplot as plt
+
+    >>> signal, sr = librosa.load(
+    ...     librosa.ex("brahms"),
+    ...     sr=None,
+    ...     duration=10,
+    ...     mono=False
+    ... )
+    >>> if signal.ndim == 1:
+    >>>      signal = signal[np.newaxis, :]
+
+    >>> compressor = Compressor(
+    ...    threshold=-30.0,
+    ...    ratio=4.0,
+    ...    attack=10.0,
+    ...    release=100.0,
+    ...    makeup_gain=1.25,
+    ...    sample_rate=sr,
+    ...    knee_width=10.0,
+    ...)
+
+    >>> compressed_signal = compressor(signal)
+    >>> fig, axes = plt.subplots(2, 1)
+    >>> axes[0].specgram(signal[0], Fs=sr, NFFT=512, noverlap=256)
+    >>> axes[0].set_title("original signal")
+    >>> axes[1].specgram(compressed_signal[0], Fs=sr, NFFT=512, noverlap=256)
+    >>> axes[1].set_title("Compressed signal")
+    >>> plt.yticks([x for x in range(0, int(sr / 2), 1000)])
+    >>> plt.tight_layout()
+    >>> plt.show()
+    >>> plt.close()
+
+    >>> plt.figure()
+    >>> plt.plot(signal[0])
+    >>> plt.plot(compressed_signal[0])
+    >>> plt.show()
+    """
+
+    def __init__(
+        self,
+        threshold: float = 0.0,
+        ratio: float = 1.0,
+        attack: float = 15.0,
+        release: float = 100.0,
+        makeup_gain: float = 0.0,
+        knee_width: float = 0.0,
+        sample_rate: float = 44100.0,
+    ) -> None:
+        if threshold > 0 or threshold < -60:
+            warnings.warn(
+                "Threshold outside the recommended range [0.0, -60.0] dB."
+                f" {threshold} dB was provided.",
+                stacklevel=1,
+            )
+        if ratio < 1 or ratio > 20:
+            warnings.warn(
+                "Ratio outside the recommended range [1.0, 20.0]."
+                f" {ratio} was provided.",
+                stacklevel=1,
+            )
+
+        if attack < 0.1 or attack > 80.0:
+            warnings.warn(
+                "Attack outside the recommended range [0.1, 80.0] ms."
+                f" {attack} ms was provided.",
+                stacklevel=1,
+            )
+
+        if release < 0.1 or release > 1000.0:
+            warnings.warn(
+                "Release outside the recommended range [0.1, 1000.0] ms."
+                f" {release} ms was provided.",
+                stacklevel=1,
+            )
+
+        if makeup_gain < 0 or makeup_gain > 24:
+            warnings.warn(
+                "Make-up gain outside the recommended range [0.0, 24.0] dB."
+                f" {makeup_gain} dB was provided.",
+                stacklevel=1,
+            )
+
+        self.threshold = float(threshold)
+        self.ratio = float(ratio)
+        self.attack = float(attack)
+        self.release = float(release)
+        self.makeup_gain = float(makeup_gain)
+        self.sample_rate = float(sample_rate)
+        self.knee_width = float(knee_width)
+
+        self.alpha_attack = np.exp(-1.0 / (0.001 * self.sample_rate * self.attack))
+        self.alpha_release = np.exp(-1.0 / (0.001 * self.sample_rate * self.release))
+
+    def __call__(self, input_signal: np.ndarray) -> np.ndarray:
+        """Process the signal.
+        The method processes the input signal and returns the output signal.
+
+        Args:
+            input_signal (np.ndarray): The input signal. (channels, samples)
+
+        Returns:
+            np.ndarray: The output signal.(channels, samples)
+        """
+
+        # Compute the instantaneous desired levels
+        x_g = 20 * np.log10(np.abs(input_signal))
+        x_g[x_g < -120] = -120
+        y_g = self.threshold + (x_g - self.threshold) / self.ratio
+
+        if self.knee_width == 0:
+            # If knee width is zero, apply hard knee
+            y_g[x_g < self.threshold] = x_g[x_g < self.threshold]
+        else:
+            # Apply soft knee
+            index = 2 * np.abs(x_g - self.threshold) <= self.knee_width
+            y_g[index] = x_g[index] + (
+                (1 / self.ratio - 1)
+                * (x_g[index] - self.threshold + self.knee_width / 2) ** 2
+            ) / (2 * self.knee_width)
+
+            index = 2 * (x_g - self.threshold) < -self.knee_width
+            y_g[index] = x_g[index]
+
+        y_l = x_g - y_g
+
+        # Do the filtering - cannot easily vectorise this part
+        filtered = np.zeros(input_signal.shape)
+        for channel, filtered_signal in enumerate(filtered):
+            out = 0
+            for i, sample in enumerate(y_l[channel]):
+                alpha = self.alpha_attack if sample > out else self.alpha_release
+                out = alpha * out + (1 - alpha) * sample
+                filtered_signal[i] = out
+
+        # Compute the gains and apply to the input signal
+        c = 10 ** ((self.makeup_gain - filtered) / 20.0)
+        output_signal = input_signal * c
+
+        return output_signal
+
+    def __str__(self) -> str:
+        """Return the string representation of the object."""
+        return (
+            f"Compressor: threshold={self.threshold}, ratio={self.ratio},"
+            f" attack={self.attack}, release={self.release}, makeup"
+            f" gain={self.makeup_gain}, sample_rate={self.sample_rate}"
+        )
diff --git a/clarity/enhancer/multiband_compressor/crossover.py b/clarity/enhancer/multiband_compressor/crossover.py
new file mode 100644
index 000000000..575b0a577
--- /dev/null
+++ b/clarity/enhancer/multiband_compressor/crossover.py
@@ -0,0 +1,265 @@
+"""Class compute crossover filter for one crossover frequency."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import matplotlib.pyplot as plt
+import numpy as np
+from scipy.signal import butter, lfilter, tf2zpk, zpk2tf
+
+if TYPE_CHECKING:
+    from numpy import ndarray
+
+
+def compute_coefficients(
+    xover_freqs: ndarray, sample_rate: float = 44100, order: int = 4
+) -> tuple[ndarray, ndarray, ndarray, ndarray]:
+    """Compute the filter coefficients.
+    These are independent of the signal.
+
+    Args:
+        xover_freqs (ndarray): The crossover frequencies.
+        sample_rate (float): The sample rate of the signal.
+        order (int): The order of the filter.
+
+    Returns:
+        bsotre (ndarray): The numerator of the filter.
+        astore (ndarray): The denominator of the filter.
+        bstore_phi (ndarray): The phase correction for the numerator of the
+            all pass filter.
+        astore_phi (ndarray): The phase correction for the denominator of the
+            all pass filter.
+    """
+    bstore = np.zeros((order + 1, xover_freqs.shape[0], 2))
+    astore = np.zeros((order + 1, xover_freqs.shape[0], 2))
+    bstore_phi = np.zeros((2 * order + 1, xover_freqs.shape[0]))
+    astore_phi = np.zeros((2 * order + 1, xover_freqs.shape[0]))
+
+    for i, freq in enumerate(xover_freqs):
+        for ifilt in range(2):
+            if ifilt == 0:
+                btype = "low"
+            else:
+                btype = "high"
+            b, a = linkwitz_riley(freq / (sample_rate / 2), btype, order)
+            bstore[:, i, ifilt] = b
+            astore[:, i, ifilt] = a
+
+        bstore_phi[:, i], astore_phi[:, i] = make_all_pass(
+            bstore[:, i, 0],
+            astore[:, i, 0],
+            bstore[:, i, 1],
+            astore[:, i, 1],
+        )
+
+    return bstore, astore, bstore_phi, astore_phi
+
+
+def make_all_pass(
+    b1: ndarray, a1: ndarray, b2: ndarray, a2: ndarray
+) -> tuple[ndarray, ndarray]:
+    """
+    Take two filters [b1,a1] and [b2,a2]
+    and calculate the coefficient of a filter that is equivalent to
+    passing the input through each filter in parallel and summing the result
+
+    Args:
+        b1: Numerator of the first filter
+        a1: Denominator of the first filter
+        b2: Numerator of the second filter
+        a2: Denominator of the second filter
+
+    Returns:
+        np.ndarray: The numerator and denominator of the all pass filter.
+    """
+    # to poles and zeros
+    r1, p1, k1 = tf2zpk(b1, a1)
+    r2, p2, k2 = tf2zpk(b2, a2)
+    roots1 = np.concatenate((r1, p2))
+    roots2 = np.concatenate((r2, p1))
+    poly1 = np.poly(roots1)
+    poly2 = np.poly(roots2)
+    bt = poly1 * k1 + poly2 * k2
+    at = np.poly(np.concatenate((p1, p2)))
+    return bt, at
+
+
+def linkwitz_riley(
+    xover_freqs: float, btype: str, order: int = 4
+) -> tuple[ndarray, ndarray]:
+    """
+    Compute the Linkwitz-Riley filter.
+    Makes a filter that is equivalent to passing through butterworth
+    twice to get linkwitz-riley
+
+    Args:
+        xover_freqs: The crossover frequency.
+        btype: The type of filter.
+        order: The order of the filter.
+
+    Returns:
+        np.ndarray: The numerator and denominator of the filter.
+    """
+    # filter coefficients for Butterworth
+    b, a = butter(order // 2, xover_freqs, btype=btype)
+    # get poles, zeros and gain
+    r, p, k = tf2zpk(b, a)
+    # duplicate zeros
+    r = np.concatenate((r, r))
+    # duplicate poles
+    p = np.concatenate((p, p))
+    # square the gain
+    k = k**2
+    # convert poles, zeros and gain back to filter coefficients
+    b, a = zpk2tf(r, p, k)
+    return b, a
+
+
+class Crossover:
+    """
+    A class to compute crossover filter for two or more crossover frequencies.
+    This is based on [1]
+
+    References:
+    [1] D'Appolito, J. A. (1984, October). Active realization of multiway all-pass
+    crossover systems. In Audio Engineering Society Convention 76.
+    Audio Engineering Society.
+
+    Example:
+    >>> xover_freqs = np.array(
+    ...    [250, 500, 1000, 1500, 2000, 3000, 4000, 5000, 6000, 8000]
+    ...) * np.sqrt(2)
+    >>> xover = Crossover(xover_freqs)
+    >>> signal = np.random.randn(1000)
+    >>> filtered_signal = xover(signal)
+    >>> xover.plot_filter()
+    """
+
+    FILTER_ORDER = 4  # Implementation only for filters order 4
+
+    def __init__(
+        self,
+        freq_crossover: list | ndarray | int,
+        sample_rate: float = 44100,
+    ) -> None:
+        """Initialize the crossover filter.
+
+        Args:
+            freq_crossover (list | ndarray | int): The crossover frequency (Hz).
+            sample_rate (float): The sample rate of the signal (Hz).
+        """
+        if isinstance(freq_crossover, int):
+            freq_crossover = [freq_crossover]
+
+        self.xover_freqs = np.array(freq_crossover)
+        self.sample_rate = sample_rate
+
+        # Initialize the filter coefficients
+
+        self.bstore, self.astore, self.bstore_phi, self.astore_phi = (
+            compute_coefficients(self.xover_freqs, self.sample_rate, self.FILTER_ORDER)
+        )
+
+    def __call__(self, signal: np.ndarray, axis: int = -1) -> np.ndarray:
+        """Apply the crossover filter to the signal.
+
+        Args:
+            signal (np.ndarray): The input signal.
+            axis (int): The axis along which to apply the filter.
+              Default is -1. More information in ```scipy.signal.lfilter```
+              documentation.
+
+        Returns:
+            np.ndarray: The filtered signal.
+        """
+        output_signal = []
+        for filter_idx in range(self.xover_freqs.shape[0] + 1):
+            output_signal.append(
+                self.xover_component(
+                    signal, filter_idx, len(self.xover_freqs), axis=axis
+                )
+            )
+        return np.array(output_signal)
+
+    def xover_component(
+        self, signal: ndarray, filter_idx: int, len_xover_freqs: int, axis: int = -1
+    ) -> ndarray:
+        """Apply the crossover filter to the signal.
+
+        Args:
+            signal (np.ndarray): The input signal.
+            filter_idx (int): The index of the filter to apply.
+            len_xover_freqs (int): The number of crossover frequencies.
+            axis (int): The axis along which to apply the filter.
+              Default is -1. ```More information in scipy.signal.lfilter```
+              documentation.
+
+        Returns:
+            np.ndarray: The filtered signal.
+        """
+        # The low pass filter component
+        if filter_idx < len_xover_freqs:
+            signal = lfilter(
+                self.bstore[:, filter_idx, 0],
+                self.astore[:, filter_idx, 0],
+                signal,
+                axis=axis,
+            )
+        # The high pass filter component
+        if filter_idx > 0:
+            if filter_idx == 1:
+                signal = lfilter(
+                    self.bstore[:, 0, 1], self.astore[:, 0, 1], signal, axis=axis
+                )
+            else:
+                for m in range(filter_idx):
+                    signal = lfilter(
+                        self.bstore[:, m, 1], self.astore[:, m, 1], signal, axis=axis
+                    )
+
+        # The phi filter component
+        if len_xover_freqs + 1 > 2:
+            if filter_idx == len_xover_freqs - 2:
+                signal = lfilter(
+                    self.bstore_phi[:, filter_idx + 1],
+                    self.astore_phi[:, filter_idx + 1],
+                    signal,
+                    axis=axis,
+                )
+            elif filter_idx < len_xover_freqs - 2:
+                for m in range(filter_idx + 1, len_xover_freqs):
+                    signal = lfilter(
+                        self.bstore_phi[:, m], self.astore_phi[:, m], signal, axis=axis
+                    )
+
+        return signal
+
+    def __str__(self):
+        """Method to print the crossover filter."""
+        return f"Crossover frequencies: {[round(x, 4) for x in self.xover_freqs]} Hz"
+
+    def plot_filter(self):
+        """Method to plot the frequency response of the filter.
+        This can help to validate the Class is generating the expected filters
+        """
+
+        x = np.concatenate(([1], np.zeros(65500)))
+        y = self(x)
+
+        fplt = np.linspace(0, self.sample_rate, len(x))
+        plt.figure()
+        for i in range(len(self.xover_freqs) + 1):
+            Y = np.fft.fft(y[i, :])
+            plt.plot(fplt, 20 * np.log10(np.abs(Y)))
+
+        ytotal = np.sum(y, axis=0)
+        Ytotal = np.fft.fft(ytotal)
+        plt.plot(
+            fplt,
+            20 * np.log10(np.abs(Ytotal)),
+        )
+        print(np.sum(20 * np.log10(np.abs(Ytotal))))
+        plt.xlim([-10, int(self.sample_rate / 2)])
+        plt.ylim([-250, 10])
+        plt.show()
diff --git a/clarity/enhancer/multiband_compressor/multiband_compressor.py b/clarity/enhancer/multiband_compressor/multiband_compressor.py
new file mode 100644
index 000000000..7e9804255
--- /dev/null
+++ b/clarity/enhancer/multiband_compressor/multiband_compressor.py
@@ -0,0 +1,261 @@
+"""An implementation for Multiband Dynamic Range Compressor."""
+
+from __future__ import annotations
+
+import numpy as np
+
+from clarity.enhancer.multiband_compressor.compressor_qmul import Compressor
+from clarity.enhancer.multiband_compressor.crossover import Crossover
+
+
+class MultibandCompressor:
+    """Multiband Compressor."""
+
+    def __init__(
+        self,
+        crossover_frequencies: int | float | list | np.ndarray,
+        sample_rate: float = 44100,
+        compressors_params: dict | None = None,
+    ) -> None:
+        """
+        Initialize the multiband compressor.
+
+        Args:
+            crossover_frequencies (list | np.ndarray): The crossover frequencies for the
+                different compressors.
+            order (int): The order of the crossover filters. Expected to be 4
+            sample_rate (float): The sample rate of the signal.
+            compressors_params (dict): A dictionary with the compressors parameters.
+                By default, all compressors are initialized with the same parameters.
+                if a dictionary is provided, it must have the following keys:
+                - attack (float | list): The attack time in milliseconds.
+                - release (float | list): The release time in milliseconds.
+                - threshold (float | list): The threshold in dB.
+                - ratio (float | list): The ratio.
+                - gain (float | list): The make-up gain in dB.
+                If the values are floats, all compressors will have the same value. If
+                the values are lists, the length must be the same as the number of
+                compressors (len(crossover_frequencies) + 1).
+
+        Example:
+        >>> import librosa
+        >>> import matplotlib.pyplot as plt
+
+        >>> signal, sr = librosa.load(
+        ...    librosa.ex("brahms"),
+        ...    sr=None,
+        ...    duration=10,
+        ...    mono=False
+        ... )
+
+        >>> signal = np.vstack((signal, signal * 0.8))
+
+        >>> HL = np.array([20, 20, 30, 40, 50, 60])
+        >>> mbc = MultibandCompressor(
+        >>>     crossover_frequencies=(
+        ...         np.array([250, 500, 1000, 2000, 4000]) * np.sqrt(2)
+        ...     ),
+        >>>     sample_rate=sr,
+        >>>     compressors_params={
+        ...         "attack": [11, 11, 14, 13, 11, 11],
+        ...         "release": [80, 80, 80, 80, 100, 100],
+        ...         "threshold": -40,
+        ...         "ratio": 4.0,
+        ...         "makeup_gain": np.maximum((HL - 20) / 3, 0),
+        ...         "knee_width": 0,
+        ...     }
+        ... )
+
+        >>> compressed_signal, compressed_bands = mbc(signal, return_bands=True)
+
+        >>> for chan in range(signal.shape[0]):
+        >>>     fig, axes = plt.subplots(3, 1, figsize=(10, 10))
+        >>>     axes[0].specgram(signal[chan], Fs=sr, NFFT=512, noverlap=256)
+        >>>     axes[0].set_title("Original Signal")
+
+        >>>     axes[1].specgram(compressed_signal[chan], Fs=sr, NFFT=512, noverlap=256)
+        >>>     axes[1].set_title("Compressed Signal")
+
+        >>>     axes[2].plot(signal[chan])
+        >>>     axes[2].plot(compressed_signal[chan])
+        >>>     axes[2].set_title("Time Domain")
+
+        >>>     fig.suptitle(f"Channel {chan}", fontsize=20)
+
+        >>>     plt.tight_layout()
+        >>>     plt.show()
+        >>>     plt.close()
+        """
+
+        if isinstance(crossover_frequencies, (int, float)):
+            crossover_frequencies = [crossover_frequencies]
+
+        self.xover_freqs = np.array(crossover_frequencies)
+        self.num_compressors = len(self.xover_freqs) + 1
+
+        # Compute the crossover filters
+        self.crossover = Crossover(self.xover_freqs, sample_rate)
+
+        self.sample_rate = sample_rate
+
+        # Set the params for the compressors
+        if compressors_params is None:
+            self.attack: float = 15.0
+            self.release: float = 100.0
+            self.threshold: float = 0.0
+            self.ratio: float = 1.0
+            self.makeup_gain: float = 0.0
+            self.knee_width: float = 0.0
+        else:
+            self.attack = compressors_params.get("attack", 15.0)
+            self.release = compressors_params.get("release", 100.0)
+            self.threshold = compressors_params.get("threshold", 0.0)
+            self.ratio = compressors_params.get("ratio", 1.0)
+            self.makeup_gain = compressors_params.get("makeup_gain", 0.0)
+            self.knee_width = compressors_params.get("knee_width", 0.0)
+
+        # Initialize the compressors
+        self.compressor: list = []
+        self.set_compressors(
+            attack=self.attack,
+            release=self.release,
+            threshold=self.threshold,
+            ratio=self.ratio,
+            makeup_gain=self.makeup_gain,
+            knee_width=self.knee_width,
+        )
+
+    def set_compressors(
+        self,
+        attack: list | float = 15.0,
+        release: list | float = 100.0,
+        threshold: list | float = 0.0,
+        ratio: list | float = 1.0,
+        makeup_gain: list | float = 0.0,
+        knee_width: list | float = 0.0,
+    ) -> None:
+        """Set the compressors parameters.
+
+        Parameters can be a float or a list with the same length as center_frequencies.
+
+        Args:
+            attack (list | float): The attack time in milliseconds.
+            release (list | float): The release time in milliseconds.
+            threshold (list | float): The threshold in dB.
+            ratio (list | float): The ratio.
+            gain (list | float): The make-up gain in dB.
+            knee_width (list | float): The knee width in dB.
+
+        Returns:
+            list[Compressor]: The compressors.
+
+        Raises:
+            ValueError: If the parameters are not the same length as
+            crossover frequencies + 1.
+        """
+
+        if isinstance(attack, (int, float)):
+            attack = [float(attack)] * self.num_compressors
+        if isinstance(release, (int, float)):
+            release = [float(release)] * self.num_compressors
+        if isinstance(threshold, (int, float)):
+            threshold = [float(threshold)] * self.num_compressors
+        if isinstance(ratio, (int, float)):
+            ratio = [float(ratio)] * self.num_compressors
+        if isinstance(makeup_gain, (int, float)):
+            makeup_gain = [float(makeup_gain)] * self.num_compressors
+        if isinstance(knee_width, (int, float)):
+            knee_width = [float(knee_width)] * self.num_compressors
+
+        if len(attack) != self.num_compressors:
+            raise ValueError(
+                "Attack must be a float or have the same "
+                "length as crossover frequencies + 1. "
+                f"{len(attack)} was provided, {self.num_compressors} expected."
+            )
+        if len(release) != self.num_compressors:
+            raise ValueError(
+                "Release must be a float or have the same "
+                "length as crossover frequencies + 1. "
+                f"{len(release)} was provided, {self.num_compressors} expected."
+            )
+        if len(threshold) != self.num_compressors:
+            raise ValueError(
+                "Threshold must be a float or have the same "
+                "length as crossover frequencies + 1. "
+                f"{len(threshold)} was provided, {self.num_compressors} expected."
+            )
+        if len(ratio) != self.num_compressors:
+            raise ValueError(
+                "Ratio must be a float or have the same "
+                "length as crossover frequencies + 1. "
+                f"{len(ratio)} was provided, {self.num_compressors} expected."
+            )
+        if len(makeup_gain) != self.num_compressors:
+            raise ValueError(
+                "Gain must be a float or have the same length as "
+                "crossover frequencies + 1. "
+                f"{len(makeup_gain)} was provided, {self.num_compressors} expected."
+            )
+        if len(knee_width) != self.num_compressors:
+            raise ValueError(
+                "Knee width must be a float or have the same length as "
+                "crossover frequencies + 1. "
+                f"{len(knee_width)} was provided, {self.num_compressors} expected."
+            )
+
+        self.compressor = [
+            Compressor(
+                attack=attack[i],
+                release=release[i],
+                threshold=threshold[i],
+                ratio=ratio[i],
+                makeup_gain=makeup_gain[i],
+                knee_width=knee_width[i],
+                sample_rate=self.sample_rate,
+            )
+            for i in range(self.num_compressors)
+        ]
+
+    def __call__(
+        self, signal: np.ndarray, return_bands: bool = False
+    ) -> np.ndarray | tuple:
+        """Compress the signal.
+
+        Args:
+            signal (np.ndarray): The input signal.
+                (channel, samples)
+            return_bands (bool): If True, return the compressed bands.
+
+        Returns:
+            np.ndarray: The compressed signal.
+                (channel, samples)
+            np.ndarray: The compressed bands if return_bands is True.
+                (bands, channel, samples)
+        """
+
+        if signal.ndim == 1:
+            # add dimension to mono signals
+            signal = signal[np.newaxis, :]
+
+        bands_signal = self.crossover(signal, axis=-1)
+
+        compressed_bands = np.zeros_like(bands_signal)
+        for idx, sig in enumerate(bands_signal):
+            compressed_bands[idx, :, :] = self.compressor[idx](sig)
+
+        compressed_signal = np.sum(compressed_bands, axis=0)
+        if return_bands:
+            return compressed_signal, compressed_bands
+        return compressed_signal
+
+    def __str__(self):
+        """Return the string representation of the object."""
+        out_text = "Multiband Compressor Summary:\n"
+        out_text += f"-Crossover Filter: Frequencies {str(self.crossover)}\n"
+        out_text += f"-Compressors: {self.num_compressors}\n"
+        for idx, comp in enumerate(self.compressor):
+            out_text += f"  -Compressor {idx + 1}: "
+            out_text += f"{comp}\n"
+
+        return out_text
diff --git a/recipes/cad2/README.md b/recipes/cad2/README.md
new file mode 100644
index 000000000..e69de29bb
diff --git a/recipes/cad2/__init__.py b/recipes/cad2/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/recipes/cad2/task1/__init__.py b/recipes/cad2/task1/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/recipes/cad2/task1/baseline/__init__.py b/recipes/cad2/task1/baseline/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/recipes/cad2/task1/baseline/config.yaml b/recipes/cad2/task1/baseline/config.yaml
new file mode 100644
index 000000000..9a169a109
--- /dev/null
+++ b/recipes/cad2/task1/baseline/config.yaml
@@ -0,0 +1,7 @@
+path:
+  root: ??
+  exp_folder: ./exp_${separator.model} # folder to store enhanced signals and final results
+
+input_sample_rate: 44100   # sample rate of the input mixture
+output_sample_rate: 32000  # sample rate output stems
+
diff --git a/recipes/cad2/task1/baseline/enhance.py b/recipes/cad2/task1/baseline/enhance.py
new file mode 100644
index 000000000..dfd2a4eef
--- /dev/null
+++ b/recipes/cad2/task1/baseline/enhance.py
@@ -0,0 +1,18 @@
+"""Baseline enhancement for CAD2 task1."""
+
+from __future__ import annotations
+
+import hydra
+from omegaconf import DictConfig
+
+
+@hydra.main(config_path="", config_name="config")
+def enhance(config: DictConfig) -> None:
+    print(config)
+
+    pass
+
+
+# pylint: disable = no-value-for-parameter
+if __name__ == "__main__":
+    enhance()
diff --git a/recipes/cad2/task1/baseline/evaluate.py b/recipes/cad2/task1/baseline/evaluate.py
new file mode 100644
index 000000000..e69de29bb


Test patch:
diff --git a/tests/enhancer/multiband_compressor/test_compressor_qmul.py b/tests/enhancer/multiband_compressor/test_compressor_qmul.py
new file mode 100644
index 000000000..38dfd6a83
--- /dev/null
+++ b/tests/enhancer/multiband_compressor/test_compressor_qmul.py
@@ -0,0 +1,94 @@
+import numpy as np
+import pytest
+
+from clarity.enhancer.multiband_compressor.compressor_qmul import Compressor
+
+
+@pytest.fixture
+def default_compressor():
+    """Return a Compressor object with default parameters."""
+    return Compressor()
+
+
+@pytest.fixture
+def custom_compressor():
+    """Return a Compressor object with custom parameters."""
+    return Compressor(
+        threshold=-30.0,
+        ratio=4.0,
+        attack=10.0,
+        release=100.0,
+        makeup_gain=1.25,
+        sample_rate=44100.0,
+        knee_width=10.0,
+    )
+
+
+@pytest.fixture
+def random_signal():
+    """Generate a random signal for testing."""
+    return np.random.randn(1, 1000)
+
+
+def test_initialization_default(default_compressor):
+    """Test the initialization of the Compressor class."""
+    assert default_compressor.threshold == 0.0
+    assert default_compressor.ratio == 1.0
+    assert default_compressor.attack == 15.0
+    assert default_compressor.release == 100.0
+    assert default_compressor.makeup_gain == 0.0
+    assert default_compressor.sample_rate == 44100.0
+    assert default_compressor.knee_width == 0.0
+
+
+def test_initialization_custom(custom_compressor):
+    """Test the initialization of the Compressor class with custom parameters."""
+    assert custom_compressor.threshold == -30.0
+    assert custom_compressor.ratio == 4.0
+    assert custom_compressor.attack == 10.0
+    assert custom_compressor.release == 100.0
+    assert custom_compressor.makeup_gain == 1.25
+    assert custom_compressor.sample_rate == 44100.0
+    assert custom_compressor.knee_width == 10.0
+
+
+def test_warning_threshold():
+    """Test the warning for a threshold outside the recommended range."""
+    with pytest.warns(UserWarning, match="Threshold outside the recommended range"):
+        Compressor(threshold=10.0)
+
+
+def test_warning_ratio():
+    """Test the warning for a ratio outside the recommended range."""
+    with pytest.warns(UserWarning, match="Ratio outside the recommended range"):
+        Compressor(ratio=0.5)
+
+
+def test_warning_attack():
+    """Test the warning for an attack outside the recommended range."""
+    with pytest.warns(UserWarning, match="Attack outside the recommended range"):
+        Compressor(attack=0.05)
+
+
+def test_warning_release():
+    """Test the warning for a release outside the recommended range."""
+    with pytest.warns(UserWarning, match="Release outside the recommended range"):
+        Compressor(release=1500.0)
+
+
+def test_warning_gain():
+    """Test the warning for a gain outside the recommended range."""
+    with pytest.warns(UserWarning, match="Make-up gain outside the recommended range"):
+        Compressor(makeup_gain=25.0)
+
+
+def test_call_default(default_compressor, random_signal):
+    """Test the call method of the Compressor class with default parameters."""
+    processed_signal = default_compressor(random_signal)
+    assert processed_signal.shape == random_signal.shape
+
+
+def test_call_custom(custom_compressor, random_signal):
+    """Test the call method of the Compressor class with custom parameters."""
+    processed_signal = custom_compressor(random_signal)
+    assert processed_signal.shape == random_signal.shape
diff --git a/tests/enhancer/multiband_compressor/test_crossover.py b/tests/enhancer/multiband_compressor/test_crossover.py
new file mode 100644
index 000000000..baea6595b
--- /dev/null
+++ b/tests/enhancer/multiband_compressor/test_crossover.py
@@ -0,0 +1,107 @@
+""" Tests for the CrossoverTwoOrMore class """
+
+import numpy as np
+import pytest
+from matplotlib import pyplot as plt
+
+from clarity.enhancer.multiband_compressor.crossover import (
+    Crossover,
+    compute_coefficients,
+)
+
+
+@pytest.fixture
+def random_signal():
+    """Generate a random signal for testing."""
+    return np.random.randn(1000)
+
+
+@pytest.fixture
+def single_freq_crossover():
+    """Return a Crossover object with a single frequency."""
+    xover_freqs = 1000
+    return Crossover(xover_freqs)
+
+
+@pytest.fixture
+def multi_freq_crossover():
+    """Return a Crossover object with multiple frequencies."""
+    xover_freqs = [1000, 2000, 3000]
+    return Crossover(xover_freqs)
+
+
+@pytest.fixture
+def complex_multi_freq_crossover():
+    """Return a Crossover object with multiple frequencies."""
+    xover_freqs = [250, 500, 1000, 1500, 2000, 3000, 4000, 5000, 6000, 8000]
+    return Crossover(xover_freqs)
+
+
+def test_initialization_single_frequency(single_freq_crossover):
+    """Test the initialization of the Crossover class with a single frequency."""
+    assert single_freq_crossover.xover_freqs.shape[0] == 1
+
+
+def test_initialization_multiple_frequencies(multi_freq_crossover):
+    """Test the initialization of the Crossover class with multiple frequencies."""
+    assert multi_freq_crossover.xover_freqs.shape[0] == 3
+
+
+def test_compute_coefficients():
+    """Test the compute_coefficients method of the Crossover class."""
+    bstore, astore, bstore_phi, astore_phi = compute_coefficients(
+        np.array([1000, 2000, 3000]), 8000, 4
+    )
+    # Check that the coefficients are not None
+    assert np.sum(bstore) == pytest.approx(
+        6.70591633, rel=pytest.rel_tolerance, abs=pytest.abs_tolerance
+    )
+    assert np.sum(astore) == pytest.approx(
+        13.411832670, rel=pytest.rel_tolerance, abs=pytest.abs_tolerance
+    )
+    assert np.sum(bstore_phi) == pytest.approx(
+        28.748181628, rel=pytest.rel_tolerance, abs=pytest.abs_tolerance
+    )
+    assert np.sum(astore_phi) == pytest.approx(
+        28.748181628, rel=pytest.rel_tolerance, abs=pytest.abs_tolerance
+    )
+
+
+def test_filter_application(multi_freq_crossover, random_signal):
+    """Test the application of the filter to a signal."""
+    filtered_signal = multi_freq_crossover(random_signal)
+    # Check the shape of the filtered signal
+    assert filtered_signal.shape == (
+        len(multi_freq_crossover.xover_freqs) + 1,
+        random_signal.shape[0],
+    )
+
+
+def test_filter_with_different_lengths(multi_freq_crossover):
+    """Test the application of the filter to signals of different lengths."""
+    for length in [100, 1000, 10000]:
+        signal = np.random.randn(length)
+        filtered_signal = multi_freq_crossover(signal)
+        # Check the shape of the filtered signal
+        assert filtered_signal.shape == (
+            len(multi_freq_crossover.xover_freqs) + 1,
+            length,
+        )
+
+
+def test_plot_filter(multi_freq_crossover, monkeypatch):
+    """Test the plot_filter method of the CrossoverTwoOrMore class."""
+    monkeypatch.setattr(plt, "show", lambda: None)
+    # Check if plot_filter runs without error
+    multi_freq_crossover.plot_filter()
+
+
+def test_complex_multi_freq_crossover(complex_multi_freq_crossover):
+    """Test the application of the filter to a signal."""
+    signal = np.random.randn(1000)
+    filtered_signal = complex_multi_freq_crossover(signal)
+    # Check the shape of the filtered signal
+    assert filtered_signal.shape == (
+        len(complex_multi_freq_crossover.xover_freqs) + 1,
+        signal.shape[0],
+    )
diff --git a/tests/enhancer/multiband_compressor/test_multiband_compresor.py b/tests/enhancer/multiband_compressor/test_multiband_compresor.py
new file mode 100644
index 000000000..294ff903b
--- /dev/null
+++ b/tests/enhancer/multiband_compressor/test_multiband_compresor.py
@@ -0,0 +1,205 @@
+import numpy as np
+import pytest
+
+from clarity.enhancer.multiband_compressor.compressor_qmul import Compressor
+from clarity.enhancer.multiband_compressor.multiband_compressor import (
+    MultibandCompressor,
+)
+
+
+@pytest.fixture
+def default_compressor():
+    return Compressor()
+
+
+def test_compressor_initialization(default_compressor):
+    """Test the initialization of the Compressor class."""
+    assert default_compressor.threshold == 0.0
+    assert default_compressor.ratio == 1.0
+    assert default_compressor.attack == 15.0
+    assert default_compressor.release == 100.0
+    assert default_compressor.makeup_gain == 0.0
+    assert default_compressor.sample_rate == 44100.0
+
+
+def test_compressor_signal_processing(default_compressor):
+    """Test the signal processing of the Compressor class."""
+    input_signal = np.array([[1, 2, 3, 4, 5]])
+    output_signal = default_compressor(input_signal)
+    # Add assertions to verify the output signal
+    assert isinstance(output_signal, np.ndarray)
+    assert len(output_signal) == len(input_signal)
+    assert np.sum(output_signal) == pytest.approx(
+        15, rel=pytest.rel_tolerance, abs=pytest.abs_tolerance
+    )
+
+
+@pytest.fixture
+def default_multiband_compressor():
+    """Fixture for the MultibandCompressor class with default parameters."""
+    return MultibandCompressor(
+        crossover_frequencies=np.array([250, 500, 1000, 2000, 4000]) * np.sqrt(2)
+    )
+
+
+def test_multiband_compressor_onefreq():
+    """Test the initialization of the MultibandCompressor class with one frequency."""
+    multiband_compressor = MultibandCompressor(crossover_frequencies=250)
+    assert multiband_compressor.num_compressors == 2
+    assert multiband_compressor.xover_freqs.shape[0] == 1
+
+
+def test_multiband_compressor_initialization(default_multiband_compressor):
+    """Test the initialization of the MultibandCompressor class."""
+    assert np.sum(default_multiband_compressor.xover_freqs) == pytest.approx(
+        np.sum(
+            np.array(
+                [
+                    250,
+                    500,
+                    1000,
+                    2000,
+                    4000,
+                ]
+            )
+            * np.sqrt(2)
+        ),
+        rel=pytest.rel_tolerance,
+        abs=pytest.abs_tolerance,
+    )
+
+    assert np.sum(default_multiband_compressor.xover_freqs) == pytest.approx(
+        np.sum(np.array(default_multiband_compressor.xover_freqs)),
+        rel=pytest.rel_tolerance,
+        abs=pytest.abs_tolerance,
+    )
+    assert default_multiband_compressor.sample_rate == 44100.0
+    assert default_multiband_compressor.num_compressors == 6
+    assert default_multiband_compressor.attack == 15.0
+    assert default_multiband_compressor.release == 100.0
+    assert default_multiband_compressor.threshold == 0.0
+    assert default_multiband_compressor.ratio == 1.0
+    assert default_multiband_compressor.makeup_gain == 0.0
+    assert default_multiband_compressor.knee_width == 0.0
+
+
+def test_multiband_compressor_set_compressors(default_multiband_compressor):
+    """Test the set_compressors method of the MultibandCompressor class."""
+    default_multiband_compressor.set_compressors(
+        attack=10.0, release=50.0, threshold=-20.0, ratio=2.0, makeup_gain=6.0
+    )
+    assert len(default_multiband_compressor.compressor) == 6
+
+
+def test_multiband_compressor_call(default_multiband_compressor):
+    """Test the __call__ method of the MultibandCompressor class."""
+    np.random.seed(0)
+    signal = np.random.rand(1000)
+
+    default_multiband_compressor.set_compressors()
+    compressed_signal = default_multiband_compressor(signal)
+    assert isinstance(compressed_signal, np.ndarray)
+
+    compressed_signal, bands = default_multiband_compressor(signal, return_bands=True)
+    assert isinstance(compressed_signal, np.ndarray)
+    assert isinstance(bands, np.ndarray)
+    assert bands.shape[0] == 6
+
+    assert np.sum(compressed_signal) == pytest.approx(
+        441.09490986, rel=pytest.rel_tolerance, abs=pytest.abs_tolerance
+    )
+
+
+def test_multiband_compressor_str(default_multiband_compressor):
+    """Test the __str__ method of the MultibandCompressor class."""
+    summary_text = str(default_multiband_compressor)
+    assert "Multiband Compressor Summary:" in summary_text
+
+
+def test_multiband_wrong_number_attack():
+    """Test the initialization of the MultibandCompressor class with
+    invalid parameters.
+    """
+    with pytest.raises(ValueError):
+        MultibandCompressor(
+            crossover_frequencies=np.array([250]),
+            compressors_params={
+                "attack": [1],
+            },
+        )
+
+
+def test_multiband_wrong_number_release():
+    """Test the initialization of the MultibandCompressor class with
+    invalid parameters.
+    """
+    with pytest.raises(ValueError):
+        MultibandCompressor(
+            crossover_frequencies=np.array([250]),
+            compressors_params={
+                "release": [1],
+            },
+        )
+
+
+def test_multiband_wrong_number_threshold():
+    """Test the initialization of the MultibandCompressor class with
+    invalid parameters.
+    """
+    with pytest.raises(ValueError):
+        MultibandCompressor(
+            crossover_frequencies=np.array([250]),
+            compressors_params={
+                "threshold": [1],
+            },
+        )
+
+
+def test_multiband_wrong_number_ratio():
+    """Test the initialization of the MultibandCompressor class with
+    invalid parameters.
+    """
+    with pytest.raises(ValueError):
+        MultibandCompressor(
+            crossover_frequencies=np.array([250]),
+            compressors_params={
+                "ratio": [1],
+            },
+        )
+
+
+def test_multiband_wrong_number_gain():
+    """Test the initialization of the MultibandCompressor class with
+    invalid parameters.
+    """
+    with pytest.raises(ValueError):
+        MultibandCompressor(
+            crossover_frequencies=np.array([250]),
+            compressors_params={
+                "makeup_gain": [1],
+            },
+        )
+
+
+def test_multiband_wrong_number_knee_width():
+    """Test the initialization of the MultibandCompressor class with
+    invalid parameters.
+    """
+    with pytest.raises(ValueError):
+        MultibandCompressor(
+            crossover_frequencies=np.array([250]),
+            compressors_params={
+                "knee_width": [1],
+            },
+        )
+
+
+def test_multiband_stereo_signal(default_multiband_compressor):
+    """Test the __call__ method of the MultibandCompressor class with
+    stereo signal."""
+    signal = np.random.rand(2, 1000)
+
+    default_multiband_compressor.set_compressors()
+    compressed_signal = default_multiband_compressor(signal)
+    assert isinstance(compressed_signal, np.ndarray)
+    assert compressed_signal.shape == signal.shape