
'''simple docstring'''
import warnings
from typing import Any, Dict, List, Optional, Union

import numpy as np

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


__lowerCAmelCase					=			logging.get_logger(__name__)


class      _lowerCAmelCase       (       __snake_case ):







							'''simple docstring'''

							lowerCAmelCase_   		=		["input_values", "attention_mask"]

							def __init__(self    ,    UpperCAmelCase = 1    ,    UpperCAmelCase = 16000    ,    UpperCAmelCase = 0.0    ,    UpperCAmelCase = False    ,    UpperCAmelCase = 80    ,    UpperCAmelCase = 16    ,    UpperCAmelCase = 64    ,    UpperCAmelCase = "hann_window"    ,    UpperCAmelCase = 1.0    ,    UpperCAmelCase = 80    ,    UpperCAmelCase = 7600    ,    UpperCAmelCase = 1e-1_0    ,    UpperCAmelCase = 2    ,    UpperCAmelCase = True    ,    **UpperCAmelCase    ,    )	->  Optional[Any]:
														super().__init__(feature_size=SCREAMING_SNAKE_CASE__    ,    sampling_rate=SCREAMING_SNAKE_CASE__    ,    padding_value=SCREAMING_SNAKE_CASE__    ,    **SCREAMING_SNAKE_CASE__    )
														_snake_case  =    do_normalize
														_snake_case  =    return_attention_mask

														_snake_case  =    num_mel_bins
														_snake_case  =    hop_length
														_snake_case  =    win_length
														_snake_case  =    win_function
														_snake_case  =    frame_signal_scale
														_snake_case  =    fmin
														_snake_case  =    fmax
														_snake_case  =    mel_floor
														_snake_case  =    reduction_factor

														_snake_case  =    win_length * sampling_rate // 1000
														_snake_case  =    hop_length * sampling_rate // 1000
														_snake_case  =    optimal_fft_length(self.sample_size    )
														_snake_case  =    (self.n_fft // 2) + 1

														_snake_case  =    window_function(window_length=self.sample_size    ,    name=self.win_function    ,    periodic=SCREAMING_SNAKE_CASE__    )

														_snake_case  =    mel_filter_bank(
														    num_frequency_bins=self.n_freqs    ,    num_mel_filters=self.num_mel_bins    ,    min_frequency=self.fmin    ,    max_frequency=self.fmax    ,    sampling_rate=self.sampling_rate    ,    norm="""slaney"""    ,    mel_scale="""slaney"""    ,    )

														if frame_signal_scale != 1.0:
																					warnings.warn(
																					    """The argument `frame_signal_scale` is deprecated and will be removed in version 4.30.0 of Transformers"""    ,    SCREAMING_SNAKE_CASE__    ,    )
														if reduction_factor != 2.0:
																					warnings.warn(
																					    """The argument `reduction_factor` is deprecated and will be removed in version 4.30.0 of Transformers"""    ,    SCREAMING_SNAKE_CASE__    ,    )

							@staticmethod
							# Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
							def 			lowercase				(UpperCAmelCase    ,    UpperCAmelCase    ,    UpperCAmelCase = 0.0    )	->  Any:



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

																					for vector, length in zip(SCREAMING_SNAKE_CASE__    ,    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(SCREAMING_SNAKE_CASE__    )
														else:
																					_snake_case  =    [(x - x.mean()) / np.sqrt(x.var() + 1e-7    ) for x in input_values]

														return normed_input_values

							def 			lowercase				(self    ,    UpperCAmelCase    ,    )	->  int:
														_snake_case  =    spectrogram(
														    SCREAMING_SNAKE_CASE__    ,    window=self.window    ,    frame_length=self.sample_size    ,    hop_length=self.sample_stride    ,    fft_length=self.n_fft    ,    mel_filters=self.mel_filters    ,    mel_floor=self.mel_floor    ,    log_mel="""log10"""    ,    )
														return log_mel_spec.T

							def __call__(self    ,    UpperCAmelCase = None    ,    UpperCAmelCase = None    ,    UpperCAmelCase = False    ,    UpperCAmelCase = None    ,    UpperCAmelCase = False    ,    UpperCAmelCase = None    ,    UpperCAmelCase = None    ,    UpperCAmelCase = None    ,    UpperCAmelCase = None    ,    **UpperCAmelCase    ,    )	->  Dict:



														if audio is None and audio_target is None:
																					raise ValueError("""You must provide either `audio` or `audio_target` values."""    )

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

														if audio is not None:
																					_snake_case  =    self._process_audio(
																					    SCREAMING_SNAKE_CASE__    ,    SCREAMING_SNAKE_CASE__    ,    SCREAMING_SNAKE_CASE__    ,    SCREAMING_SNAKE_CASE__    ,    SCREAMING_SNAKE_CASE__    ,    SCREAMING_SNAKE_CASE__    ,    SCREAMING_SNAKE_CASE__    ,    SCREAMING_SNAKE_CASE__    ,    **SCREAMING_SNAKE_CASE__    ,    )
														else:
																					_snake_case  =    None

														if audio_target is not None:
																					_snake_case  =    self._process_audio(
																					    SCREAMING_SNAKE_CASE__    ,    SCREAMING_SNAKE_CASE__    ,    SCREAMING_SNAKE_CASE__    ,    SCREAMING_SNAKE_CASE__    ,    SCREAMING_SNAKE_CASE__    ,    SCREAMING_SNAKE_CASE__    ,    SCREAMING_SNAKE_CASE__    ,    SCREAMING_SNAKE_CASE__    ,    **SCREAMING_SNAKE_CASE__    ,    )

																					if inputs is None:
																												return inputs_target
																					else:
																												_snake_case  =    inputs_target["""input_values"""]
																												_snake_case  =    inputs_target.get("""attention_mask"""    )
																												if decoder_attention_mask is not None:
																																			_snake_case  =    decoder_attention_mask

														return inputs

							def 			lowercase				(self    ,    UpperCAmelCase    ,    UpperCAmelCase = False    ,    UpperCAmelCase = False    ,    UpperCAmelCase = None    ,    UpperCAmelCase = False    ,    UpperCAmelCase = None    ,    UpperCAmelCase = None    ,    UpperCAmelCase = None    ,    **UpperCAmelCase    ,    )	->  List[str]:
														_snake_case  =    isinstance(SCREAMING_SNAKE_CASE__    ,    np.ndarray    ) and len(speech.shape    ) > 1
														if is_batched_numpy and len(speech.shape    ) > 2:
																					raise ValueError(f"""Only mono-channel audio is supported for input to {self}"""    )
														_snake_case  =    is_batched_numpy or (
														    isinstance(SCREAMING_SNAKE_CASE__    ,    (list, tuple)    ) and (isinstance(speech[0]    ,    (np.ndarray, tuple, list)    ))
														)

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

														# always return batch
														if not is_batched:
																					_snake_case  =    [speech]

														# needed to make pad() work on spectrogram inputs
														_snake_case  =    self.feature_size

														# convert into correct format for padding
														if is_target:
																					_snake_case  =    [self._extract_mel_features(SCREAMING_SNAKE_CASE__    ) for waveform in speech]
																					_snake_case  =    BatchFeature({"""input_values""": features}    )
																					_snake_case  =    self.num_mel_bins
														else:
																					_snake_case  =    BatchFeature({"""input_values""": speech}    )

														_snake_case  =    self.pad(
														    SCREAMING_SNAKE_CASE__    ,    padding=SCREAMING_SNAKE_CASE__    ,    max_length=SCREAMING_SNAKE_CASE__    ,    truncation=SCREAMING_SNAKE_CASE__    ,    pad_to_multiple_of=SCREAMING_SNAKE_CASE__    ,    return_attention_mask=SCREAMING_SNAKE_CASE__    ,    **SCREAMING_SNAKE_CASE__    ,    )

														_snake_case  =    feature_size_hack

														# convert input values to correct format
														_snake_case  =    padded_inputs["""input_values"""]
														if not isinstance(input_values[0]    ,    np.ndarray    ):
																					_snake_case  =    [np.asarray(SCREAMING_SNAKE_CASE__    ,    dtype=np.floataa    ) for array in input_values]
														elif (
														    not isinstance(SCREAMING_SNAKE_CASE__    ,    np.ndarray    )
														    and isinstance(input_values[0]    ,    np.ndarray    )
														    and input_values[0].dtype is np.dtype(np.floataa    )
														):
																					_snake_case  =    [array.astype(np.floataa    ) for array in input_values]
														elif isinstance(SCREAMING_SNAKE_CASE__    ,    np.ndarray    ) and input_values.dtype is np.dtype(np.floataa    ):
																					_snake_case  =    input_values.astype(np.floataa    )

														# convert attention_mask to correct format
														_snake_case  =    padded_inputs.get("""attention_mask"""    )
														if attention_mask is not None:
																					_snake_case  =    [np.asarray(SCREAMING_SNAKE_CASE__    ,    dtype=np.intaa    ) for array in attention_mask]

														# zero-mean and unit-variance normalization
														if not is_target and self.do_normalize:
																					_snake_case  =    (
																					    attention_mask
																					    if self._get_padding_strategies(SCREAMING_SNAKE_CASE__    ,    max_length=SCREAMING_SNAKE_CASE__    ) is not PaddingStrategy.DO_NOT_PAD
																					    else None
																					)
																					_snake_case  =    self.zero_mean_unit_var_norm(
																					    padded_inputs["""input_values"""]    ,    attention_mask=SCREAMING_SNAKE_CASE__    ,    padding_value=self.padding_value    )

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

														return padded_inputs





							def 			lowercase				(self    )	->  Optional[int]:
														_snake_case  =    super().to_dict()

														# Don't serialize these as they are derived from the other properties.
														_snake_case  =    ["""window""", """mel_filters""", """sample_size""", """sample_stride""", """n_fft""", """n_freqs"""]
														for name in names:
																					if name in output:
																												del output[name]

														return output

import json
import os
import tempfile

from transformers.testing_utils import check_json_file_has_correct_format
class       _lowerCAmelCase  :

 """simple docstring"""
 SCREAMING_SNAKE_CASE_				:					Optional[Any]       						=None

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






      """simple docstring"""



      UpperCamelCase						    =					self.feature_extraction_class(**self.feat_extract_dict   )
      UpperCamelCase						    =					json.loads(feat_extract.to_json_string()   )
      for key, value in self.feat_extract_dict.items():
           self.assertEqual(obj[key]					, SCREAMING_SNAKE_CASE__   )

 def      __lowerCAmelCase  (						self			:			Tuple   ):






      """simple docstring"""



      UpperCamelCase						    =					self.feature_extraction_class(**self.feat_extract_dict   )

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

      self.assertEqual(feat_extract_second.to_dict()					, feat_extract_first.to_dict()   )

 def      __lowerCAmelCase  (						self			:			Tuple   ):






      """simple docstring"""



      UpperCamelCase						    =					self.feature_extraction_class(**self.feat_extract_dict   )

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

      self.assertEqual(feat_extract_second.to_dict()					, feat_extract_first.to_dict()   )






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






      """simple docstring"""



      UpperCamelCase						    =					self.feature_extraction_class()
      self.assertIsNotNone(SCREAMING_SNAKE_CASE__   )


