



"""simple docstring"""






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


snake_case   					=				logging.get_logger(__name__)



class     UpperCAmelCase    (	__SCREAMING_SNAKE_CASE  ):
   A__		:	Union[str, Any]           =  ['''input_features''']



   def __init__(       self    :	Dict							,       __lowerCamelCase    :	int=8_0							,       __lowerCamelCase    :	List[Any]=1_6_0_0_0							,       __lowerCamelCase    :	List[str]=1_6_0							,       __lowerCamelCase    :	int=3_0							,       __lowerCamelCase    :	Union[str, Any]=4_0_0							,       __lowerCamelCase    :	Dict=0.0							,       __lowerCamelCase    :	Optional[Any]=False							,       **__lowerCamelCase    :	Tuple							,       ):



     """simple docstring"""
     super().__init__(
         feature_size=__lowerCamelCase							,       sampling_rate=__lowerCamelCase							,       padding_value=__lowerCamelCase							,       return_attention_mask=__lowerCamelCase							,       **__lowerCamelCase							,       )
     _snake_case   	=      n_fft
     _snake_case   	=      hop_length
     _snake_case   	=      chunk_length
     _snake_case   	=      chunk_length * sampling_rate
     _snake_case   	=      self.n_samples // hop_length
     _snake_case   	=      sampling_rate
     _snake_case   	=      mel_filter_bank(
         num_frequency_bins=1 + n_fft // 2							,       num_mel_filters=__lowerCamelCase							,       min_frequency=0.0							,       max_frequency=8_0_0_0.0							,       sampling_rate=__lowerCamelCase							,       norm='''slaney'''							,       mel_scale='''slaney'''							,       )



   def    __UpperCAmelCase    (       self    :	List[str]							,       __lowerCamelCase    :	np.array				):



     """simple docstring"""
     _snake_case   	=      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'''							,       )
     _snake_case   	=      log_spec[:, :-1]
     _snake_case   	=      np.maximum(__lowerCamelCase							,       log_spec.max() - 8.0				)
     _snake_case   	=      (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    __UpperCAmelCase    (       __lowerCamelCase    :	List[np.ndarray]							,       __lowerCamelCase    :	List[np.ndarray]							,       __lowerCamelCase    :	float = 0.0				):



     """simple docstring"""
     if attention_mask is not None:
       _snake_case   	=      np.array(__lowerCamelCase							,       np.intaa				)
       _snake_case   	=      []

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

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

     return normed_input_values



   def __call__(       self    :	Union[str, Any]							,       __lowerCamelCase    :	Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]							,       __lowerCamelCase    :	bool = True							,       __lowerCamelCase    :	Optional[int] = None							,       __lowerCamelCase    :	Optional[Union[str, TensorType]] = None							,       __lowerCamelCase    :	Optional[bool] = None							,       __lowerCamelCase    :	Optional[str] = "max_length"							,       __lowerCamelCase    :	Optional[int] = None							,       __lowerCamelCase    :	Optional[int] = None							,       __lowerCamelCase    :	Optional[bool] = None							,       **__lowerCamelCase    :	Dict							,       ):



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

     _snake_case   	=      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}"""				)
     _snake_case   	=      is_batched_numpy or (
         isinstance(__lowerCamelCase							,       (list, tuple)				) and (isinstance(raw_speech[0]							,       (np.ndarray, tuple, list)				))
     )

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

     # always return batch
     if not is_batched:
       _snake_case   	=      [np.asarray([raw_speech]				).T]

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

     # convert into correct format for padding

     _snake_case   	=      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:
       _snake_case   	=      self.zero_mean_unit_var_norm(
           padded_inputs['''input_features''']							,       attention_mask=padded_inputs['''attention_mask''']							,       padding_value=self.padding_value							,       )
       _snake_case   	=      np.stack(padded_inputs['''input_features''']							,       axis=0				)

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

     _snake_case   	=      [self._np_extract_fbank_features(__lowerCamelCase				) for waveform in input_features[0]]

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

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

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

     return padded_inputs





   def    __UpperCAmelCase    (       self    :	Union[str, Any]				):



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










"""simple docstring"""






import itertools
import os
import random
import tempfile
import unittest

import numpy as np

from transformers import TvltFeatureExtractor, is_datasets_available
from transformers.testing_utils import check_json_file_has_correct_format, require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_torch_available():
     import torch

if is_datasets_available():
     from datasets import load_dataset

snake_case   					=				random.Random()




def 			snake_case		(		lowerCAmelCase_  ,						lowerCAmelCase_=1.0  ,						lowerCAmelCase_=None  ,						lowerCAmelCase_=None	)						->				List[str]:
  if rng is None:
    _snake_case   	=      global_rng

  _snake_case   	=      []
  for batch_idx in range(shape[0]	):
    values.append([]	)
    for _ in range(shape[1]	):
      values[-1].append(rng.random() * scale	)

  return values



class     UpperCAmelCase    (	unittest.TestCase  ):



   def __init__(       self    :	Tuple							,       __lowerCamelCase    :	List[str]							,       __lowerCamelCase    :	Optional[Any]=7							,       __lowerCamelCase    :	List[Any]=4_0_0							,       __lowerCamelCase    :	Any=2_0_0_0							,       __lowerCamelCase    :	Any=2_0_4_8							,       __lowerCamelCase    :	Any=1_2_8							,       __lowerCamelCase    :	Any=1							,       __lowerCamelCase    :	Optional[int]=5_1_2							,       __lowerCamelCase    :	Tuple=3_0							,       __lowerCamelCase    :	List[Any]=4_4_1_0_0							,       ):



     """simple docstring"""
     _snake_case   	=      parent
     _snake_case   	=      batch_size
     _snake_case   	=      min_seq_length
     _snake_case   	=      max_seq_length
     _snake_case   	=      (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
     _snake_case   	=      spectrogram_length
     _snake_case   	=      feature_size
     _snake_case   	=      num_audio_channels
     _snake_case   	=      hop_length
     _snake_case   	=      chunk_length
     _snake_case   	=      sampling_rate



   def    __UpperCAmelCase    (       self    :	List[Any]				):



     """simple docstring"""
     return {
         "spectrogram_length": self.spectrogram_length,
         "feature_size": self.feature_size,
         "num_audio_channels": self.num_audio_channels,
         "hop_length": self.hop_length,
         "chunk_length": self.chunk_length,
         "sampling_rate": self.sampling_rate,
     }





   def    __UpperCAmelCase    (       self    :	Union[str, Any]							,       __lowerCamelCase    :	Any=False							,       __lowerCamelCase    :	int=False				):



     """simple docstring"""
     def _flatten(__lowerCamelCase    :	List[str]				):
       return list(itertools.chain(*__lowerCamelCase				)				)

     if equal_length:
       _snake_case   	=      [floats_list((self.max_seq_length, self.feature_size)				) for _ in range(self.batch_size				)]
     else:
       # make sure that inputs increase in size
       _snake_case   	=      [
           floats_list((x, self.feature_size)				)
           for x in range(self.min_seq_length							,       self.max_seq_length							,       self.seq_length_diff				)
       ]
     if numpify:
       _snake_case   	=      [np.asarray(__lowerCamelCase				) for x in speech_inputs]
     return speech_inputs



@require_torch
@require_torchaudio
class     UpperCAmelCase    (	__SCREAMING_SNAKE_CASE,unittest.TestCase  ):
   A__		:	Tuple           =  TvltFeatureExtractor



   def    __UpperCAmelCase    (       self    :	int				):



     """simple docstring"""
     _snake_case   	=      TvltFeatureExtractionTester(self				)



   def    __UpperCAmelCase    (       self    :	Any				):



     """simple docstring"""
     _snake_case   	=      self.feature_extraction_class(**self.feat_extract_dict				)
     self.assertTrue(hasattr(__lowerCamelCase							,       '''spectrogram_length'''				)				)
     self.assertTrue(hasattr(__lowerCamelCase							,       '''feature_size'''				)				)
     self.assertTrue(hasattr(__lowerCamelCase							,       '''num_audio_channels'''				)				)
     self.assertTrue(hasattr(__lowerCamelCase							,       '''hop_length'''				)				)
     self.assertTrue(hasattr(__lowerCamelCase							,       '''chunk_length'''				)				)
     self.assertTrue(hasattr(__lowerCamelCase							,       '''sampling_rate'''				)				)



   def    __UpperCAmelCase    (       self    :	int				):



     """simple docstring"""
     _snake_case   	=      self.feature_extraction_class(**self.feat_extract_dict				)

     with tempfile.TemporaryDirectory() as tmpdirname:
       _snake_case   	=      feat_extract_first.save_pretrained(__lowerCamelCase				)[0]
       check_json_file_has_correct_format(__lowerCamelCase				)
       _snake_case   	=      self.feature_extraction_class.from_pretrained(__lowerCamelCase				)

     _snake_case   	=      feat_extract_first.to_dict()
     _snake_case   	=      feat_extract_second.to_dict()
     _snake_case   	=      dict_first.pop('''mel_filters'''				)
     _snake_case   	=      dict_second.pop('''mel_filters'''				)
     self.assertTrue(np.allclose(__lowerCamelCase							,       __lowerCamelCase				)				)
     self.assertEqual(__lowerCamelCase							,       __lowerCamelCase				)



   def    __UpperCAmelCase    (       self    :	int				):



     """simple docstring"""
     _snake_case   	=      self.feature_extraction_class(**self.feat_extract_dict				)

     with tempfile.TemporaryDirectory() as tmpdirname:
       _snake_case   	=      os.path.join(__lowerCamelCase							,       '''feat_extract.json'''				)
       feat_extract_first.to_json_file(__lowerCamelCase				)
       _snake_case   	=      self.feature_extraction_class.from_json_file(__lowerCamelCase				)

     _snake_case   	=      feat_extract_first.to_dict()
     _snake_case   	=      feat_extract_second.to_dict()
     _snake_case   	=      dict_first.pop('''mel_filters'''				)
     _snake_case   	=      dict_second.pop('''mel_filters'''				)
     self.assertTrue(np.allclose(__lowerCamelCase							,       __lowerCamelCase				)				)
     self.assertEqual(__lowerCamelCase							,       __lowerCamelCase				)



   def    __UpperCAmelCase    (       self    :	Dict				):



     """simple docstring"""
     # Initialize feature_extractor
     _snake_case   	=      self.feature_extraction_class(**self.feat_extract_dict				)

     # create three inputs of length 800, 1000, and 1200
     _snake_case   	=      [floats_list((1, x)				)[0] for x in range(8_0_0							,       1_4_0_0							,       2_0_0				)]
     _snake_case   	=      [np.asarray(__lowerCamelCase				) for speech_input in speech_inputs]

     # Test not batched input
     _snake_case   	=      feature_extractor(np_speech_inputs[0]							,       return_tensors='''np'''							,       sampling_rate=4_4_1_0_0				).audio_values

     self.assertTrue(encoded_audios.ndim == 4				)
     self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size				)
     self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length				)
     self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels				)

     # Test batched
     _snake_case   	=      feature_extractor(__lowerCamelCase							,       return_tensors='''np'''							,       sampling_rate=4_4_1_0_0				).audio_values

     self.assertTrue(encoded_audios.ndim == 4				)
     self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size				)
     self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length				)
     self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels				)

     # Test audio masking
     _snake_case   	=      feature_extractor(
         __lowerCamelCase							,       return_tensors='''np'''							,       sampling_rate=4_4_1_0_0							,       mask_audio=__lowerCamelCase				).audio_values

     self.assertTrue(encoded_audios.ndim == 4				)
     self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size				)
     self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length				)
     self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels				)

     # Test 2-D numpy arrays are batched.
     _snake_case   	=      [floats_list((1, x)				)[0] for x in (8_0_0, 8_0_0, 8_0_0)]
     _snake_case   	=      np.asarray(__lowerCamelCase				)
     _snake_case   	=      feature_extractor(__lowerCamelCase							,       return_tensors='''np'''							,       sampling_rate=4_4_1_0_0				).audio_values
     self.assertTrue(encoded_audios.ndim == 4				)
     self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size				)
     self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length				)
     self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels				)



   def    __UpperCAmelCase    (       self    :	Union[str, Any]							,       __lowerCamelCase    :	Optional[int]				):



     """simple docstring"""
     _snake_case   	=      load_dataset('''hf-internal-testing/librispeech_asr_dummy'''							,       '''clean'''							,       split='''validation'''				)
     # automatic decoding with librispeech
     _snake_case   	=      ds.sort('''id'''				).select(range(__lowerCamelCase				)				)[:num_samples]['''audio''']

     return [x["array"] for x in speech_samples]





   def    __UpperCAmelCase    (       self    :	Dict				):



     """simple docstring"""
     _snake_case   	=      self._load_datasamples(1				)
     _snake_case   	=      TvltFeatureExtractor()
     _snake_case   	=      feature_extractor(__lowerCamelCase							,       return_tensors='''pt'''				).audio_values

     self.assertEquals(audio_values.shape							,       (1, 1, 1_9_2, 1_2_8)				)

     _snake_case   	=      torch.tensor([[-0.3_0_3_2, -0.2_7_0_8], [-0.4_4_3_4, -0.4_0_0_7]]				)
     self.assertTrue(torch.allclose(audio_values[0, 0, :2, :2]							,       __lowerCamelCase							,       atol=1E-4				)				)





