"""simple docstring"""


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


lowerCAmelCase_:					Optional[Any]     =					logging.get_logger(__name__)






class    a__    (   __UpperCAmelCase	):


   def __init__(     self,				_UpperCAmelCase,				_UpperCAmelCase,				_UpperCAmelCase,				**_UpperCAmelCase  ):
        '''simple docstring'''



        lowercase__				 =							feature_size
        lowercase__				 =							sampling_rate
        lowercase__				 =							padding_value

        lowercase__				 =							kwargs.pop("padding_side",				"right"  )
        lowercase__				 =							kwargs.pop("return_attention_mask",				lowerCAmelCase_  )

        super().__init__(**lowerCAmelCase_  )


   def 	snake_case__	(     self,				_UpperCAmelCase,				_UpperCAmelCase = True,				_UpperCAmelCase = None,				_UpperCAmelCase = False,				_UpperCAmelCase = None,				_UpperCAmelCase = None,				_UpperCAmelCase = None,				):
        '''simple docstring'''



        if isinstance(lowerCAmelCase_,				(list, tuple)  ) and isinstance(processed_features[0],				(dict, BatchFeature)  ):
             lowercase__				 =							{
                 key: [example[key] for example in processed_features] for key in processed_features[0].keys()
             }

        # The model's main input name, usually `input_values`, has be passed for padding
        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()  )}'''  )

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

        if len(lowerCAmelCase_  ) == 0:
             if return_attention_mask:
                  lowercase__				 =							[]
             return processed_features

        # 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

        lowercase__				 =							required_input[0]
        if isinstance(lowerCAmelCase_,				(list, tuple)  ):
             # first_element might be an empty list/tuple in some edge cases so we grab the first non empty element.
             lowercase__				 =							0
             while len(required_input[index]  ) == 0:
                  index += 1
             if index < len(lowerCAmelCase_  ):
                  lowercase__				 =							required_input[index][0]

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

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

             # Convert padding_strategy in PaddingStrategy
        lowercase__				 =							self._get_padding_strategies(padding=lowerCAmelCase_,				max_length=lowerCAmelCase_  )

        lowercase__				 =							processed_features[self.model_input_names[0]]

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

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

        if padding_strategy == PaddingStrategy.LONGEST:
             # make sure that `max_length` cannot be longer than the longest truncated length
             lowercase__				 =							max(len(input_slice[self.model_input_names[0]]  ) for input_slice in truncated_inputs  )
             lowercase__				 =							PaddingStrategy.MAX_LENGTH

        lowercase__				 =							{}
        for i in range(lowerCAmelCase_  ):
             # padding
             lowercase__				 =							self._pad(
                 truncated_inputs[i],				max_length=lowerCAmelCase_,				padding_strategy=lowerCAmelCase_,				pad_to_multiple_of=lowerCAmelCase_,				return_attention_mask=lowerCAmelCase_,				)

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

        return BatchFeature(lowerCAmelCase_,				tensor_type=lowerCAmelCase_  )


   def 	snake_case__	(     self,				_UpperCAmelCase,				_UpperCAmelCase = None,				_UpperCAmelCase = PaddingStrategy.DO_NOT_PAD,				_UpperCAmelCase = None,				_UpperCAmelCase = None,				):
        '''simple docstring'''



        lowercase__				 =							processed_features[self.model_input_names[0]]

        if padding_strategy == PaddingStrategy.LONGEST:
             lowercase__				 =							len(lowerCAmelCase_  )

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

        lowercase__				 =							padding_strategy != PaddingStrategy.DO_NOT_PAD and len(lowerCAmelCase_  ) < max_length

        if return_attention_mask and "attention_mask" not in processed_features:
             lowercase__				 =							np.ones(len(lowerCAmelCase_  ),				dtype=np.intaa  )

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

        return processed_features


   def 	snake_case__	(     self,				_UpperCAmelCase,				_UpperCAmelCase = None,				_UpperCAmelCase = None,				_UpperCAmelCase = None,				):
        '''simple docstring'''



        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."  )

        lowercase__				 =							processed_features[self.model_input_names[0]]

        # find `max_length` that fits `pad_to_multiple_of`
        if max_length is not None and pad_to_multiple_of is not None and (max_length % pad_to_multiple_of != 0):
             lowercase__				 =							((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

        lowercase__				 =							len(lowerCAmelCase_  ) > max_length

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

        return processed_features


   def 	snake_case__	(     self,				_UpperCAmelCase=False,				_UpperCAmelCase=None  ):
        '''simple docstring'''



        if padding is not False:
             if padding is True:
                  lowercase__				 =							PaddingStrategy.LONGEST  # Default to pad to the longest sequence in the batch
             elif not isinstance(lowerCAmelCase_,				lowerCAmelCase_  ):
                  lowercase__				 =							PaddingStrategy(lowerCAmelCase_  )
             elif isinstance(lowerCAmelCase_,				lowerCAmelCase_  ):
                  lowercase__				 =							padding
        else:
             lowercase__				 =							PaddingStrategy.DO_NOT_PAD

        # Set max length if needed
        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'''  )

             # Test if we have a padding value
        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






"""simple docstring"""


from math import ceil
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 BatchFeature, SequenceFeatureExtractor
from ...utils import TensorType, logging


lowerCAmelCase_:					Union[str, Any]     =					logging.get_logger(__name__)






class    a__    (   _a	):
   snake_case_							=      ["audio_values", "audio_mask"]


   def __init__(     self,				_UpperCAmelCase=2048,				_UpperCAmelCase=1,				_UpperCAmelCase=[16, 16],				_UpperCAmelCase=128,				_UpperCAmelCase=4_4100,				_UpperCAmelCase=86,				_UpperCAmelCase=2048,				_UpperCAmelCase=0.0,				**_UpperCAmelCase,				):
        '''simple docstring'''



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

        lowercase__				 =							spectrogram_length
        lowercase__				 =							num_channels
        lowercase__				 =							patch_size
        lowercase__				 =							feature_size // self.patch_size[1]
        lowercase__				 =							n_fft
        lowercase__				 =							sampling_rate // hop_length_to_sampling_rate
        lowercase__				 =							sampling_rate
        lowercase__				 =							padding_value
        lowercase__				 =							mel_filter_bank(
            num_frequency_bins=1 + n_fft // 2,				num_mel_filters=_UpperCAmelCase,				min_frequency=0.0,				max_frequency=22_050.0,				sampling_rate=_UpperCAmelCase,				norm="slaney",				mel_scale="slaney",				).T


   def 	snake_case__	(     self,				_UpperCAmelCase  ):
        '''simple docstring'''



        lowercase__				 =							spectrogram(
            _UpperCAmelCase,				window_function(self.n_fft,				"hann"  ),				frame_length=self.n_fft,				hop_length=self.hop_length,				power=2.0,				mel_filters=self.mel_filters.T,				log_mel="dB",				db_range=80.0,				)
        lowercase__				 =							log_spec[:, :-1]
        lowercase__				 =							log_spec - 20.0
        lowercase__				 =							np.clip(log_spec / 40.0,				-2.0,				0.0  ) + 1.0
        return log_spec


   def __call__(     self,				_UpperCAmelCase,				_UpperCAmelCase = None,				_UpperCAmelCase = True,				_UpperCAmelCase = None,				_UpperCAmelCase = False,				_UpperCAmelCase = False,				**_UpperCAmelCase,				):
        '''simple docstring'''



        if sampling_rate is not None:
             if sampling_rate != self.sampling_rate:
                  raise ValueError(
                      "This feature extractor is set to support sampling rate"
                      F''' of {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled'''
                      F''' with {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."  )

        lowercase__				 =							isinstance(_UpperCAmelCase,				np.ndarray  ) and len(raw_speech.shape  ) > 1
        if is_batched_numpy and len(raw_speech.shape  ) > 2:
             raise ValueError(F'''Only mono-channel audio is supported for input to {self}'''  )
        lowercase__				 =							is_batched_numpy or (
            isinstance(_UpperCAmelCase,				(list, tuple)  ) and (isinstance(raw_speech[0],				(np.ndarray, tuple, list)  ))
        )
        if is_batched:
             lowercase__				 =							[np.asarray([speech],				dtype=np.floataa  ).T for speech in raw_speech]
        elif not is_batched and not isinstance(_UpperCAmelCase,				np.ndarray  ):
             lowercase__				 =							np.asarray(_UpperCAmelCase,				dtype=np.floataa  )
        elif isinstance(_UpperCAmelCase,				np.ndarray  ) and raw_speech.dtype is np.dtype(np.floataa  ):
             lowercase__				 =							raw_speech.astype(np.floataa  )
        # always return batch
        if not is_batched:
             lowercase__				 =							[np.asarray([raw_speech]  ).T]

        # Convert audio signals to log mel spectrograms, truncate by time axis
        lowercase__				 =							[
            self._np_extract_fbank_features(waveform.squeeze()  ).T[: self.spectrogram_length] for waveform in raw_speech
        ]
        if isinstance(audio_features[0],				_UpperCAmelCase  ):
             lowercase__				 =							[np.asarray(_UpperCAmelCase,				dtype=np.floataa  ) for feature in audio_features]

        # Create audio attention mask
        lowercase__				 =							max(
            [ceil(feature.shape[0] / self.patch_size[0]  ) * self.freq_len for feature in audio_features]  )  # The maximum number of audio patches in a batch
        if return_attention_mask:
             lowercase__				 =							[
                 (ceil(feature.shape[0] / self.patch_size[0]  ) * self.freq_len) * [1]
                 + (max_patch_len - ceil(feature.shape[0] / self.patch_size[0]  ) * self.freq_len) * [0]
                 for feature in audio_features
             ]
             lowercase__				 =							np.array(_UpperCAmelCase  ).astype(np.floataa  )

        # convert into correct format for padding
        lowercase__				 =							max_patch_len // self.freq_len * self.patch_size[0]  # The maximum audio size in a batch
        lowercase__				 =							np.ones([len(_UpperCAmelCase  ), 1, max_time_len, self.feature_size]  ).astype(np.floataa  )
        lowercase__				 =							padded_audio_features * self.padding_value
        for i in range(len(_UpperCAmelCase  )  ):
             lowercase__				 =							audio_features[i]
             lowercase__				 =							feature

        # return as BatchFeature
        if return_attention_mask:
             lowercase__				 =							{"audio_values": padded_audio_features, "audio_mask": audio_mask}
        else:
             lowercase__				 =							{"audio_values": padded_audio_features}

        lowercase__				 =							BatchFeature(data=_UpperCAmelCase,				tensor_type=_UpperCAmelCase  )
        return encoded_inputs




