

import copy
from typing import Any, Dict, 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 TensorType, logging


lowercase_									=   logging.get_logger(__name__)





class  __lowerCAmelCase     (					SCREAMING_SNAKE_CASE							):
  _a									=  ["""input_features"""]




  def __init__(				self							,   lowerCAmelCase=80							,   lowerCAmelCase=16_000							,   lowerCAmelCase=160							,   lowerCAmelCase=30							,   lowerCAmelCase=400							,   lowerCAmelCase=0.0							,   lowerCAmelCase=False							,   **lowerCAmelCase							,   )  -> Any:
        '''simple docstring'''






        super().__init__(
            feature_size=lowerCAmelCase							,   sampling_rate=lowerCAmelCase							,   padding_value=lowerCAmelCase							,   return_attention_mask=lowerCAmelCase							,   **lowerCAmelCase							,   )
        _lowercase	       =n_fft
        _lowercase	       =hop_length
        _lowercase	       =chunk_length
        _lowercase	       =chunk_length * sampling_rate
        _lowercase	       =self.n_samples // hop_length
        _lowercase	       =sampling_rate
        _lowercase	       =mel_filter_bank(
            num_frequency_bins=1 + n_fft // 2							,   num_mel_filters=lowerCAmelCase							,   min_frequency=0.0							,   max_frequency=8000.0							,   sampling_rate=lowerCAmelCase							,   norm='slaney'							,   mel_scale='slaney'							,   )




  def   A__     (				self							,   lowerCAmelCase					)  -> np.ndarray:
        '''simple docstring'''






        _lowercase	       =spectrogram(
            lowerCAmelCase							,   window_function(self.n_fft							,   'hann'					)							,   frame_length=self.n_fft							,   hop_length=self.hop_length							,   power=2.0							,   mel_filters=self.mel_filters							,   log_mel='log10'							,   )
        _lowercase	       =log_spec[:, :-1]
        _lowercase	       =np.maximum(lowerCAmelCase							,   log_spec.max() - 8.0					)
        _lowercase	       =(log_spec + 4.0) / 4.0
        return log_spec




  @staticmethod
  # Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
  def   A__     (				lowerCAmelCase							,   lowerCAmelCase							,   lowerCAmelCase = 0.0					)  -> List[np.ndarray]:
        '''simple docstring'''






        if attention_mask is not None:
              _lowercase	       =np.array(lowerCAmelCase							,   np.intaa					)
              _lowercase	       =[]

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

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

        return normed_input_values




  def __call__(				self							,   lowerCAmelCase							,   lowerCAmelCase = True							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   lowerCAmelCase = "max_length"							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   **lowerCAmelCase							,   )  -> BatchFeature:
        '''simple docstring'''






        if sampling_rate is not None:
              if sampling_rate != self.sampling_rate:
                    raise ValueError(
                        F'''The model corresponding to this feature extractor: {self.__class__.__name__} was trained using a'''
                        F''' sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input'''
                        F''' was sampled 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(lowerCAmelCase							,   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(lowerCAmelCase							,   (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(lowerCAmelCase							,   np.ndarray					):
              _lowercase	       =np.asarray(lowerCAmelCase							,   dtype=np.floataa					)
        elif isinstance(lowerCAmelCase							,   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]

        _lowercase	       =BatchFeature({'input_features': raw_speech}					)

        # convert into correct format for padding

        _lowercase	       =self.pad(
            lowerCAmelCase							,   padding=lowerCAmelCase							,   max_length=max_length if max_length else self.n_samples							,   truncation=lowerCAmelCase							,   pad_to_multiple_of=lowerCAmelCase							,   return_attention_mask=return_attention_mask or do_normalize							,   )

        # zero-mean and unit-variance normalization
        if do_normalize:
              _lowercase	       =self.zero_mean_unit_var_norm(
                  padded_inputs['input_features']							,   attention_mask=padded_inputs['attention_mask']							,   padding_value=self.padding_value							,   )
              _lowercase	       =np.stack(padded_inputs['input_features']							,   axis=0					)

        # make sure list is in array format
        _lowercase	       =padded_inputs.get('input_features'					).transpose(2							,   0							,   1					)

        _lowercase	       =[self._np_extract_fbank_features(lowerCAmelCase					) for waveform in input_features[0]]

        if isinstance(input_features[0]							,   lowerCAmelCase					):
              _lowercase	       =[np.asarray(lowerCAmelCase							,   dtype=np.floataa					) for feature in input_features]
        else:
              _lowercase	       =input_features

        if return_attention_mask:
              # rescale from sample (48000) to feature (3000)
              _lowercase	       =padded_inputs['attention_mask'][:, :: self.hop_length]

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

        return padded_inputs




  def   A__     (				self					)  -> Dict[str, Any]:
        '''simple docstring'''






        _lowercase	       =copy.deepcopy(self.__dict__					)
        _lowercase	       =self.__class__.__name__
        if "mel_filters" in output:
              del output["mel_filters"]
        return output





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


lowercase_									=   logging.get_logger(__name__)





class  __lowerCAmelCase     (					SCREAMING_SNAKE_CASE							):




  def __init__(				self							,   lowerCAmelCase							,   lowerCAmelCase							,   lowerCAmelCase							,   **lowerCAmelCase					)  -> Dict:
        '''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   A__     (				self							,   lowerCAmelCase							,   lowerCAmelCase = True							,   lowerCAmelCase = None							,   lowerCAmelCase = False							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   )  -> BatchFeature:
        '''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   A__     (				self							,   lowerCAmelCase							,   lowerCAmelCase = None							,   lowerCAmelCase = PaddingStrategy.DO_NOT_PAD							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   )  -> dict:
        '''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   A__     (				self							,   lowerCAmelCase							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   )  -> Any:
        '''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   A__     (				self							,   lowerCAmelCase=False							,   lowerCAmelCase=None					)  -> Optional[int]:
        '''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


