codingutf8 2023 the huggingface inc team licensed under the apache license version 2 0 the license you may not use this file except in compliance with the license you may obtain a copy of the license at http www apache orglicenseslicense2 0 unless required by applicable law or agreed to in writing software distributed under the license is distributed on an as is basis without warranties or conditions of any kind either express or implied see the license for the specific language governing permissions and limitations under the license feature extractor class for seamlessm4t modelinputnames inputfeatures attentionmask def init self featuresize80 samplingrate16000 nummelbins80 paddingvalue0 0 stride2 kwargs self nummelbins nummelbins self returnattentionmask true self stride stride melfilters melfilterbank numfrequencybins256 nummelfiltersself nummelbins minfrequency20 maxfrequencysamplingrate 2 samplingratesamplingrate normnone melscalekaldi triangularizeinmelspacetrue self melfilters np padmelfilters 0 1 0 0 self window windowfunction400 povey periodicfalse super initfeaturesizefeaturesize samplingratesamplingrate paddingvaluepaddingvalue kwargs staticmethod copied from transformers models wav2vec2 featureextractionwav2vec2 wav2vec2featureextractor zeromeanunitvarnorm def zeromeanunitvarnorm inputvalues listnp ndarray attentionmask listnp ndarray paddingvalue float 0 0 listnp ndarray if attentionmask is not none attentionmask np arrayattentionmask np int32 normedinputvalues for vector length in zipinputvalues attentionmask sum1 normedslice vector vector length mean np sqrtvector length var 1e7 if length normedslice shape0 normedslicelength paddingvalue normedinputvalues appendnormedslice else normedinputvalues x x mean np sqrtx var 1e7 for x in inputvalues return normedinputvalues def extractfbankfeatures self waveform np ndarray np ndarray by default it extracts the left channel if stereo if lenwaveform shape 2 waveform waveform0 waveform np squeezewaveform 215 kaldi compliance 16bit signed integers features spectrogram waveform self window framelength400 hoplength160 fftlength512 power2 0 centerfalse preemphasis0 97 melfiltersself melfilters logmellog melfloor1 192092955078125e07 removedcoffsettrue t return features def call self rawspeech unionnp ndarray listfloat listnp ndarray listlistfloat padding unionbool str paddingstrategy true padtomultipleof optionalint 2 maxlength optionalint none truncation bool false returntensors optionalunionstr tensortype none samplingrate optionalint none returnattentionmask optionalbool none donormalizepermelbins optionalbool true kwargs batchfeature if samplingrate is not none if samplingrate self samplingrate raise valueerror fthe model corresponding to this feature extractor self was trained using a sampling rate of f self samplingrate please make sure that the provided rawspeech input was sampled with f self samplingrate and not samplingrate else logger warning it is strongly recommended to pass the samplingrate argument to this function failing to do so can result in silent errors that might be hard to debug isbatchednumpy isinstancerawspeech np ndarray and lenrawspeech shape 1 if isbatchednumpy and lenrawspeech shape 3 raise valueerrorfonly monochannel or stereochannel audio is supported for input to self isbatched isbatchednumpy or isinstancerawspeech list tuple and isinstancerawspeech0 np ndarray tuple list if isbatched rawspeech np asarrayspeech dtypenp float32 for speech in rawspeech elif not isbatched and not isinstancerawspeech np ndarray rawspeech np asarrayrawspeech dtypenp float32 elif isinstancerawspeech np ndarray and rawspeech dtype is np dtypenp float64 rawspeech rawspeech astypenp float32 always return batch if not isbatched rawspeech rawspeech extract fbank features features self extractfbankfeatureswaveform for waveform in rawspeech if donormalizepermelbins torch defaults to ddof1 and numpy defaults to ddof0 features x np expanddimsx mean0 0 np sqrtnp expanddimsx var0 ddof1 0 1e7 for x in features convert into correct format for padding encodedinputs batchfeatureinputfeatures features paddedinputs self pad encodedinputs paddingpadding maxlengthmaxlength truncationtruncation padtomultipleofpadtomultipleof returnattentionmaskreturnattentionmask returntensorsnp seamlessm4t needs to process extracted features inputfeatures paddedinputs getinputfeatures attentionmask paddedinputs getattentionmask batchsize numframes numchannels inputfeatures shape remainder numframes self stride if remainder 0 inputfeatures inputfeatures numframes attentionmask attentionmask numframes inputfeatures np reshape inputfeatures batchsize numframes self stride numchannels self stride indices np arange0 numframes attentionmask attentionmask indices self stride 1 paddedinputsinputfeatures inputfeatures paddedinputsattentionmask attentionmask if returntensors is not none paddedinputs paddedinputs converttotensorsreturntensors return paddedinputs coding utf 8 2023 the huggingface inc team licensed under the apache license version 2 0 the license you may not use this file except in compliance with the license you may obtain a copy of the license at http www apache org licenses license 2 0 unless required by applicable law or agreed to in writing software distributed under the license is distributed on an as is basis without warranties or conditions of any kind either express or implied see the license for the specific language governing permissions and limitations under the license feature extractor class for seamlessm4t constructs a seamlessm4t feature extractor this feature extractor inherits from sequencefeatureextractor which contains most of the main methods users should refer to this superclass for more information regarding those methods this class extracts mel filter bank features from raw speech args feature_size int optional defaults to 80 the feature dimension of the extracted features sampling_rate int optional defaults to 16000 the sampling rate at which the audio files should be digitalized expressed in hertz hz num_mel_bins int optional defaults to 80 number of mel frequency bins padding_value float optional defaults to 0 0 the value that is used to fill the padding vectors stride int optional defaults to 2 stride used to reshape audios from shape batch_size num_frames num_mel_bins to batch_size num_frames stride num_mel_bins stride copied from transformers models wav2vec2 feature_extraction_wav2vec2 wav2vec2featureextractor zero_mean_unit_var_norm every array in the list is normalized to have zero mean and unit variance get mel filter bank features using torchaudio note that torchaudio requires 16 bit signed integers as inputs and hence the waveform should not be normalized before feature extraction by default it extracts the left channel if stereo kaldi compliance 16 bit signed integers main method to featurize and prepare for the model one or several sequence s args raw_speech np ndarray list float list np ndarray list list float list list list float the sequence or batch of sequences to be padded each sequence can be a numpy array a list of float values a list of numpy arrays a list of list of float values or a list of a list of list of float values if raw_speech is a one dimensional np ndarray or a list float raw_speech is considered a single channel single sample sound in all other cases the first dimension of raw_speech whether from an np ndarray or a list corresponds to the number of samples in the batch and the number of channels i e mono or stereo character is derived from the other dimensions 1d single channel waveform batches 2d stereo channel waveform batches padding bool str or utils paddingstrategy optional defaults to true select a strategy to pad the returned sequences according to the model s padding side and padding index among true or longest pad to the longest sequence in the batch or no padding if only a single sequence if provided max_length pad to a maximum length specified with the argument max_length or to the maximum acceptable input length for the model if that argument is not provided false or do_not_pad default no padding i e can output a batch with sequences of different lengths pad_to_multiple_of int optional defaults to 2 if set will pad the sequence to a multiple of the provided value this is especially useful to enable the use of tensor cores on nvidia hardware with compute capability 7 5 volta or on tpus which benefit from having sequence lengths be a multiple of 128 max_length int optional maximum length of the returned list and optionally padding length see above truncation bool activates truncation to cut input sequences longer than max_length to max_length return_attention_mask bool optional whether to return the attention mask if left to the default will return the attention mask according to the specific feature_extractor s default what are attention masks glossary attention mask tip for seamlessm4t models attention_mask should always be passed for batched inference to avoid subtle bugs tip return_tensors str or utils tensortype optional if set will return tensors instead of list of python integers acceptable values are tf return tensorflow tf constant objects pt return pytorch torch tensor objects np return numpy np ndarray objects sampling_rate int optional the sampling rate at which the raw_speech input was sampled it is strongly recommended to pass sampling_rate at the forward call to prevent silent errors do_normalize_per_mel_bins bool optional defaults to true whether or not to zero mean unit variance normalize the input per mel channel kwargs optional remaining dictionary of keyword arguments that will be passed to the tokenizer or the feature extractor always return batch extract fbank features torch defaults to ddof 1 and numpy defaults to ddof 0 convert into correct format for padding seamlessm4t needs to process extracted features
from typing import List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


logger = logging.get_logger(__name__)


class SeamlessM4TFeatureExtractor(SequenceFeatureExtractor):
    r

    model_input_names = ["input_features", "attention_mask"]

    def __init__(
        self,
        feature_size=80,
        sampling_rate=16000,
        num_mel_bins=80,
        padding_value=0.0,
        stride=2,
        **kwargs,
    ):
        self.num_mel_bins = num_mel_bins
        self.return_attention_mask = True
        self.stride = stride

        mel_filters = mel_filter_bank(
            num_frequency_bins=256,
            num_mel_filters=self.num_mel_bins,
            min_frequency=20,
            max_frequency=sampling_rate // 2,
            sampling_rate=sampling_rate,
            norm=None,
            mel_scale="kaldi",
            triangularize_in_mel_space=True,
        )

        self.mel_filters = np.pad(mel_filters, ((0, 1), (0, 0)))
        self.window = window_function(400, "povey", periodic=False)

        super().__init__(feature_size=feature_size, sampling_rate=sampling_rate, padding_value=padding_value, **kwargs)

    @staticmethod
    
    def zero_mean_unit_var_norm(
        input_values: List[np.ndarray], attention_mask: List[np.ndarray], padding_value: float = 0.0
    ) -> List[np.ndarray]:
        
        if attention_mask is not None:
            attention_mask = np.array(attention_mask, np.int32)
            normed_input_values = []

            for vector, length in zip(input_values, attention_mask.sum(-1)):
                normed_slice = (vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1e-7)
                if length < normed_slice.shape[0]:
                    normed_slice[length:] = padding_value

                normed_input_values.append(normed_slice)
        else:
            normed_input_values = [(x - x.mean()) / np.sqrt(x.var() + 1e-7) for x in input_values]

        return normed_input_values

    def _extract_fbank_features(
        self,
        waveform: np.ndarray,
    ) -> np.ndarray:
        
        
        if len(waveform.shape) == 2:
            waveform = waveform[0]

        waveform = np.squeeze(waveform) * (2**15)  
        features = spectrogram(
            waveform,
            self.window,
            frame_length=400,
            hop_length=160,
            fft_length=512,
            power=2.0,
            center=False,
            preemphasis=0.97,
            mel_filters=self.mel_filters,
            log_mel="log",
            mel_floor=1.192092955078125e-07,
            remove_dc_offset=True,
        ).T
        return features

    def __call__(
        self,
        raw_speech: Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]],
        padding: Union[bool, str, PaddingStrategy] = True,
        pad_to_multiple_of: Optional[int] = 2,
        max_length: Optional[int] = None,
        truncation: bool = False,
        return_tensors: Optional[Union[str, TensorType]] = None,
        sampling_rate: Optional[int] = None,
        return_attention_mask: Optional[bool] = None,
        do_normalize_per_mel_bins: Optional[bool] = True,
        **kwargs,
    ) -> BatchFeature:
        
        if sampling_rate is not None:
            if sampling_rate != self.sampling_rate:
                raise ValueError(
                    f"The model corresponding to this feature extractor: {self} was trained using a sampling rate of"
                    f" {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with"
                    f" {self.sampling_rate} and not {sampling_rate}."
                )
        else:
            logger.warning(
                "It is strongly recommended to pass the `sampling_rate` argument to this function. "
                "Failing to do so can result in silent errors that might be hard to debug."
            )

        is_batched_numpy = isinstance(raw_speech, np.ndarray) and len(raw_speech.shape) > 1
        if is_batched_numpy and len(raw_speech.shape) > 3:
            raise ValueError(f"Only mono-channel or stereo-channel audio is supported for input to {self}")

        is_batched = is_batched_numpy or (
            isinstance(raw_speech, (list, tuple)) and (isinstance(raw_speech[0], (np.ndarray, tuple, list)))
        )

        if is_batched:
            raw_speech = [np.asarray(speech, dtype=np.float32) for speech in raw_speech]
        elif not is_batched and not isinstance(raw_speech, np.ndarray):
            raw_speech = np.asarray(raw_speech, dtype=np.float32)
        elif isinstance(raw_speech, np.ndarray) and raw_speech.dtype is np.dtype(np.float64):
            raw_speech = raw_speech.astype(np.float32)

        
        if not is_batched:
            raw_speech = [raw_speech]

        
        features = [self._extract_fbank_features(waveform) for waveform in raw_speech]

        if do_normalize_per_mel_bins:
            
            features = [
                (x - np.expand_dims(x.mean(0), 0)) / np.sqrt(np.expand_dims(x.var(0, ddof=1), 0) + 1e-7)
                for x in features
            ]

        
        encoded_inputs = BatchFeature({"input_features": features})

        padded_inputs = self.pad(
            encoded_inputs,
            padding=padding,
            max_length=max_length,
            truncation=truncation,
            pad_to_multiple_of=pad_to_multiple_of,
            return_attention_mask=return_attention_mask,
            return_tensors="np",
        )

        
        input_features = padded_inputs.get("input_features")
        attention_mask = padded_inputs.get("attention_mask")

        batch_size, num_frames, num_channels = input_features.shape

        remainder = num_frames % self.stride
        if remainder != 0:
            input_features = input_features[:, :num_frames, :]
            attention_mask = attention_mask[:, :num_frames]

        input_features = np.reshape(
            input_features, (batch_size, num_frames // self.stride, num_channels * self.stride)
        )

        indices = np.arange(0, num_frames)
        attention_mask = attention_mask[:, indices % self.stride == 1]

        padded_inputs["input_features"] = input_features
        padded_inputs["attention_mask"] = attention_mask

        if return_tensors is not None:
            padded_inputs = padded_inputs.convert_to_tensors(return_tensors)

        return padded_inputs