backend/preprocess_wav.py

import soundfile
import io
from typing import Any, Tuple, Union, Optional
import numpy as np
import torch

def preprocess_wav(data: Any, incoming_sample_rate) -> Tuple[np.ndarray, int]:
        segment, sample_rate = soundfile.read(
            io.BytesIO(data),
            dtype="float32",
            always_2d=True,
            frames=-1,
            start=0,
            format="RAW",
            subtype="PCM_16",
            samplerate=incoming_sample_rate,
            channels=1,
        )
        return segment, sample_rate

def convert_waveform(
        waveform: Union[np.ndarray, torch.Tensor],
        sample_rate: int,
        normalize_volume: bool = False,
        to_mono: bool = False,
        to_sample_rate: Optional[int] = None,
    ) -> Tuple[Union[np.ndarray, torch.Tensor], int]:
        """convert a waveform:
        - to a target sample rate
        - from multi-channel to mono channel
        - volume normalization

        Args:
            waveform (numpy.ndarray or torch.Tensor): 2D original waveform
                (channels x length)
            sample_rate (int): original sample rate
            normalize_volume (bool): perform volume normalization
            to_mono (bool): convert to mono channel if having multiple channels
            to_sample_rate (Optional[int]): target sample rate
        Returns:
            waveform (numpy.ndarray): converted 2D waveform (channels x length)
            sample_rate (float): target sample rate
        """
        try:
            import torchaudio.sox_effects as ta_sox
        except ImportError:
            raise ImportError("Please install torchaudio: pip install torchaudio")

        effects = []
        if normalize_volume:
            effects.append(["gain", "-n"])
        if to_sample_rate is not None and to_sample_rate != sample_rate:
            effects.append(["rate", f"{to_sample_rate}"])
        if to_mono and waveform.shape[0] > 1:
            effects.append(["channels", "1"])
        if len(effects) > 0:
            is_np_input = isinstance(waveform, np.ndarray)
            _waveform = torch.from_numpy(waveform) if is_np_input else waveform
            converted, converted_sample_rate = ta_sox.apply_effects_tensor(
                _waveform, sample_rate, effects
            )
            if is_np_input:
                converted = converted.numpy()
            return converted, converted_sample_rate
        return waveform, sample_rate