codingutf8 2021 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 sequence feature extraction class for common feature extractors to preprocess sequences this is a general feature extraction class for speech recognition args featuresize int the feature dimension of the extracted features samplingrate int the sampling rate at which the audio files should be digitalized expressed in hertz hz paddingvalue float the value that is used to fill the padding values vectors pad input values input vectors or a batch of input values input vectors up to predefined length or to the max sequence length in the batch padding side leftright padding values are defined at the feature extractor level with self paddingside self paddingvalue tip if the processedfeatures passed are dictionary of numpy arrays pytorch tensors or tensorflow tensors the result will use the same type unless you provide a different tensor type with returntensors in the case of pytorch tensors you will lose the specific device of your tensors however tip args processedfeatures batchfeature list of batchfeature dictstr listfloat dictstr listlistfloat or listdictstr listfloat processed inputs can represent one input batchfeature or dictstr listfloat or a batch of input values vectors list of batchfeature dictstr listlistfloat or listdictstr listfloat so you can use this method during preprocessing as well as in a pytorch dataloader collate function instead of listfloat you can have tensors numpy arrays pytorch tensors or tensorflow tensors see the note above for the return type 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 maxlength pad to a maximum length specified with the argument maxlength or to the maximum acceptable input length for the model if that argument is not provided false or donotpad default no padding i e can output a batch with sequences of different lengths maxlength int optional maximum length of the returned list and optionally padding length see above truncation bool activates truncation to cut input sequences longer than maxlength to maxlength padtomultipleof int optional 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 returnattentionmask bool optional whether to return the attention mask if left to the default will return the attention mask according to the specific featureextractor s default what are attention masks glossaryattentionmask returntensors 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 if we have a list of dicts let s convert it in a dict of lists we do this to allow using this method as a collatefn function in pytorch dataloader the model s main input name usually inputvalues has be passed for padding if we have pytorchtf tensors or lists as inputs we cast them as numpy arrays and rebuild them afterwards if no returntensors is specified note that we lose the specific device the tensor may be on for pytorch firstelement might be an empty listtuple in some edge cases so we grab the first non empty element convert paddingstrategy in paddingstrategy truncation make sure that maxlength cannot be longer than the longest truncated length padding pad inputs on leftright and up to predefined length or max length in the batch args processedfeatures uniondictstr np ndarray batchfeature dictionary of input values np ndarrayfloat input vectors listnp ndarrayfloat or batch of inputs values listnp ndarrayint input vectors listnp ndarrayint maxlength int optional maximum length of the returned list and optionally padding length see below paddingstrategy paddingstrategy optional default to paddingstrategy donotpad paddingstrategy to use for padding paddingstrategy longest pad to the longest sequence in the batch paddingstrategy maxlength pad to the max length default paddingstrategy donotpad do not pad the featureextractor padding sides are defined in self paddingside left pads on the left of the sequences right pads on the right of the sequences padtomultipleof int optional integer if set will pad the sequence to a multiple of the provided value this is especially useful to enable the use of tensor core on nvidia hardware with compute capability 7 5 volta or on tpus which benefit from having sequence lengths be a multiple of 128 returnattentionmask bool optional set to false to avoid returning attention mask default set to model specifics truncate inputs to predefined length or max length in the batch args processedfeaturesuniondictstr np ndarray batchfeature dictionary of input values np ndarrayfloat input vectors listnp ndarrayfloat or batch of inputs values listnp ndarrayint input vectors listnp ndarrayint maxlength int optional maximum length of the returned list and optionally padding length see below padtomultipleof int optional integer if set will pad the sequence to a multiple of the provided value this is especially useful to enable the use of tensor core on nvidia hardware with compute capability 7 5 volta or on tpus which benefit from having sequence lengths be a multiple of 128 truncation bool optional activates truncation to cut input sequences longer than maxlength to maxlength find maxlength that fits padtomultipleof find the correct padding strategy get padding strategy set max length if needed test if we have a padding value coding utf 8 2021 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 sequence feature extraction class for common feature extractors to preprocess sequences this is a general feature extraction class for speech recognition args feature_size int the feature dimension of the extracted features sampling_rate int the sampling rate at which the audio files should be digitalized expressed in hertz hz padding_value float the value that is used to fill the padding values vectors pad input values input vectors or a batch of input values input vectors up to predefined length or to the max sequence length in the batch padding side left right padding values are defined at the feature extractor level with self padding_side self padding_value tip if the processed_features passed are dictionary of numpy arrays pytorch tensors or tensorflow tensors the result will use the same type unless you provide a different tensor type with return_tensors in the case of pytorch tensors you will lose the specific device of your tensors however tip args processed_features batchfeature list of batchfeature dict str list float dict str list list float or list dict str list float processed inputs can represent one input batchfeature or dict str list float or a batch of input values vectors list of batchfeature dict str list list float or list dict str list float so you can use this method during preprocessing as well as in a pytorch dataloader collate function instead of list float you can have tensors numpy arrays pytorch tensors or tensorflow tensors see the note above for the return type 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 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 pad_to_multiple_of int optional 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 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 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 if we have a list of dicts let s convert it in a dict of lists we do this to allow using this method as a collate_fn function in pytorch dataloader the model s main input name usually input_values has be passed for padding if we have pytorch tf tensors or lists as inputs we cast them as numpy arrays and rebuild them afterwards if no return_tensors is specified note that we lose the specific device the tensor may be on for pytorch first_element might be an empty list tuple in some edge cases so we grab the first non empty element convert padding_strategy in paddingstrategy truncation make sure that max_length cannot be longer than the longest truncated length padding pad inputs on left right and up to predefined length or max length in the batch args processed_features union dict str np ndarray batchfeature dictionary of input values np ndarray float input vectors list np ndarray float or batch of inputs values list np ndarray int input vectors list np ndarray int max_length int optional maximum length of the returned list and optionally padding length see below padding_strategy paddingstrategy optional default to paddingstrategy do_not_pad paddingstrategy to use for padding paddingstrategy longest pad to the longest sequence in the batch paddingstrategy max_length pad to the max length default paddingstrategy do_not_pad do not pad the feature_extractor padding sides are defined in self padding_side left pads on the left of the sequences right pads on the right of the sequences pad_to_multiple_of int optional integer if set will pad the sequence to a multiple of the provided value this is especially useful to enable the use of tensor core on nvidia hardware with compute capability 7 5 volta or on tpus which benefit from having sequence lengths be a multiple of 128 return_attention_mask bool optional set to false to avoid returning attention mask default set to model specifics truncate inputs to predefined length or max length in the batch args processed_features union dict str np ndarray batchfeature dictionary of input values np ndarray float input vectors list np ndarray float or batch of inputs values list np ndarray int input vectors list np ndarray int max_length int optional maximum length of the returned list and optionally padding length see below pad_to_multiple_of int optional integer if set will pad the sequence to a multiple of the provided value this is especially useful to enable the use of tensor core on nvidia hardware with compute capability 7 5 volta or on tpus which benefit from having sequence lengths be a multiple of 128 truncation bool optional activates truncation to cut input sequences longer than max_length to max_length find max_length that fits pad_to_multiple_of find the correct padding strategy get padding strategy default to pad to the longest sequence in the batch set max length if needed test if we have a padding value
from typing import Dict, List, Optional, Union

import numpy as np

from .feature_extraction_utils import BatchFeature, FeatureExtractionMixin
from .utils import PaddingStrategy, TensorType, is_tf_tensor, is_torch_tensor, logging, to_numpy


logger = logging.get_logger(__name__)


class SequenceFeatureExtractor(FeatureExtractionMixin):
    

    def __init__(self, feature_size: int, sampling_rate: int, padding_value: float, **kwargs):
        self.feature_size = feature_size
        self.sampling_rate = sampling_rate
        self.padding_value = padding_value

        self.padding_side = kwargs.pop("padding_side", "right")
        self.return_attention_mask = kwargs.pop("return_attention_mask", True)

        super().__init__(**kwargs)

    def pad(
        self,
        processed_features: Union[
            BatchFeature,
            List[BatchFeature],
            Dict[str, BatchFeature],
            Dict[str, List[BatchFeature]],
            List[Dict[str, BatchFeature]],
        ],
        padding: Union[bool, str, PaddingStrategy] = True,
        max_length: Optional[int] = None,
        truncation: bool = False,
        pad_to_multiple_of: Optional[int] = None,
        return_attention_mask: Optional[bool] = None,
        return_tensors: Optional[Union[str, TensorType]] = None,
    ) -> BatchFeature:
        
        
        
        if isinstance(processed_features, (list, tuple)) and isinstance(processed_features[0], (dict, BatchFeature)):
            processed_features = {
                key: [example[key] for example in processed_features] for key in processed_features[0].keys()
            }

        
        if self.model_input_names[0] not in processed_features:
            raise ValueError(
                "You should supply an instance of `transformers.BatchFeature` or list of `transformers.BatchFeature`"
                f" to this method that includes {self.model_input_names[0]}, but you provided"
                f" {list(processed_features.keys())}"
            )

        required_input = processed_features[self.model_input_names[0]]
        return_attention_mask = (
            return_attention_mask if return_attention_mask is not None else self.return_attention_mask
        )

        if len(required_input) == 0:
            if return_attention_mask:
                processed_features["attention_mask"] = []
            return processed_features

        
        
        

        first_element = required_input[0]
        if isinstance(first_element, (list, tuple)):
            
            index = 0
            while len(required_input[index]) == 0:
                index += 1
            if index < len(required_input):
                first_element = required_input[index][0]

        if return_tensors is None:
            if is_tf_tensor(first_element):
                return_tensors = "tf"
            elif is_torch_tensor(first_element):
                return_tensors = "pt"
            elif isinstance(first_element, (int, float, list, tuple, np.ndarray)):
                return_tensors = "np"
            else:
                raise ValueError(
                    f"type of {first_element} unknown: {type(first_element)}. "
                    "Should be one of a python, numpy, pytorch or tensorflow object."
                )

        for key, value in processed_features.items():
            if isinstance(value[0], (int, float)):
                processed_features[key] = to_numpy(value)
            else:
                processed_features[key] = [to_numpy(v) for v in value]

        
        padding_strategy = self._get_padding_strategies(padding=padding, max_length=max_length)

        required_input = processed_features[self.model_input_names[0]]

        batch_size = len(required_input)
        if not all(len(v) == batch_size for v in processed_features.values()):
            raise ValueError("Some items in the output dictionary have a different batch size than others.")

        truncated_inputs = []
        for i in range(batch_size):
            inputs = {k: v[i] for k, v in processed_features.items()}
            
            inputs_slice = self._truncate(
                inputs,
                max_length=max_length,
                pad_to_multiple_of=pad_to_multiple_of,
                truncation=truncation,
            )
            truncated_inputs.append(inputs_slice)

        if padding_strategy == PaddingStrategy.LONGEST:
            
            max_length = max(len(input_slice[self.model_input_names[0]]) for input_slice in truncated_inputs)
            padding_strategy = PaddingStrategy.MAX_LENGTH

        batch_outputs = {}
        for i in range(batch_size):
            
            outputs = self._pad(
                truncated_inputs[i],
                max_length=max_length,
                padding_strategy=padding_strategy,
                pad_to_multiple_of=pad_to_multiple_of,
                return_attention_mask=return_attention_mask,
            )

            for key, value in outputs.items():
                if key not in batch_outputs:
                    batch_outputs[key] = []
                if value.dtype is np.dtype(np.float64):
                    value = value.astype(np.float32)
                batch_outputs[key].append(value)

        return BatchFeature(batch_outputs, tensor_type=return_tensors)

    def _pad(
        self,
        processed_features: Union[Dict[str, np.ndarray], BatchFeature],
        max_length: Optional[int] = None,
        padding_strategy: PaddingStrategy = PaddingStrategy.DO_NOT_PAD,
        pad_to_multiple_of: Optional[int] = None,
        return_attention_mask: Optional[bool] = None,
    ) -> dict:
        
        required_input = processed_features[self.model_input_names[0]]

        if padding_strategy == PaddingStrategy.LONGEST:
            max_length = len(required_input)

        if max_length is not None and pad_to_multiple_of is not None and (max_length % pad_to_multiple_of != 0):
            max_length = ((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

        needs_to_be_padded = padding_strategy != PaddingStrategy.DO_NOT_PAD and len(required_input) < max_length

        if return_attention_mask and "attention_mask" not in processed_features:
            processed_features["attention_mask"] = np.ones(len(required_input), dtype=np.int32)

        if needs_to_be_padded:
            difference = max_length - len(required_input)
            if self.padding_side == "right":
                if return_attention_mask:
                    processed_features["attention_mask"] = np.pad(
                        processed_features["attention_mask"], (0, difference)
                    )
                padding_shape = ((0, difference), (0, 0)) if self.feature_size > 1 else (0, difference)
                processed_features[self.model_input_names[0]] = np.pad(
                    required_input, padding_shape, "constant", constant_values=self.padding_value
                )
            elif self.padding_side == "left":
                if return_attention_mask:
                    processed_features["attention_mask"] = np.pad(
                        processed_features["attention_mask"], (difference, 0)
                    )
                padding_shape = ((difference, 0), (0, 0)) if self.feature_size > 1 else (difference, 0)
                processed_features[self.model_input_names[0]] = np.pad(
                    required_input, padding_shape, "constant", constant_values=self.padding_value
                )
            else:
                raise ValueError("Invalid padding strategy:" + str(self.padding_side))

        return processed_features

    def _truncate(
        self,
        processed_features: Union[Dict[str, np.ndarray], BatchFeature],
        max_length: Optional[int] = None,
        pad_to_multiple_of: Optional[int] = None,
        truncation: Optional[bool] = None,
    ):
        
        if not truncation:
            return processed_features
        elif truncation and max_length is None:
            raise ValueError("When setting ``truncation=True``, make sure that ``max_length`` is defined.")

        required_input = processed_features[self.model_input_names[0]]

        
        if max_length is not None and pad_to_multiple_of is not None and (max_length % pad_to_multiple_of != 0):
            max_length = ((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

        needs_to_be_truncated = len(required_input) > max_length

        if needs_to_be_truncated:
            processed_features[self.model_input_names[0]] = processed_features[self.model_input_names[0]][:max_length]
            if "attention_mask" in processed_features:
                processed_features["attention_mask"] = processed_features["attention_mask"][:max_length]

        return processed_features

    def _get_padding_strategies(self, padding=False, max_length=None):
        

        
        if padding is not False:
            if padding is True:
                padding_strategy = PaddingStrategy.LONGEST  
            elif not isinstance(padding, PaddingStrategy):
                padding_strategy = PaddingStrategy(padding)
            elif isinstance(padding, PaddingStrategy):
                padding_strategy = padding
        else:
            padding_strategy = PaddingStrategy.DO_NOT_PAD

        
        if max_length is None:
            if padding_strategy == PaddingStrategy.MAX_LENGTH:
                raise ValueError(
                    f"When setting ``padding={PaddingStrategy.MAX_LENGTH}``, make sure that max_length is defined"
                )

        
        if padding_strategy != PaddingStrategy.DO_NOT_PAD and (self.padding_value is None):
            raise ValueError(
                "Asking to pad but the feature_extractor does not have a padding value. Please select a value to use"
                " as `padding_value`. For example: `feature_extractor.padding_value = 0.0`."
            )

        return padding_strategy