

'''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


_lowerCAmelCase         =						logging.get_logger(__name__)






class   _SCREAMING_SNAKE_CASE   (   __a     ):
 __SCREAMING_SNAKE_CASE					:str      	=					["""input_features"""]



 def __init__(   self		:   Union[str, Any]			,				a__		:   Optional[int]=80			,				a__		:   Union[str, Any]=1_6000			,				a__		:   int=160			,				a__		:   List[Any]=30			,				a__		:   int=400			,				a__		:   Optional[Any]=0.0			,				a__		:   List[Any]=False			,				**a__		:   Dict			,				):
      super().__init__(
          feature_size=a__			,				sampling_rate=a__			,				padding_value=a__			,				return_attention_mask=a__			,				**a__			,				)
      __magic_name__						       =    n_fft
      __magic_name__						       =    hop_length
      __magic_name__						       =    chunk_length
      __magic_name__						       =    chunk_length * sampling_rate
      __magic_name__						       =    self.n_samples // hop_length
      __magic_name__						       =    sampling_rate
      __magic_name__						       =    mel_filter_bank(
          num_frequency_bins=1 + n_fft // 2			,				num_mel_filters=a__			,				min_frequency=0.0			,				max_frequency=8_000.0			,				sampling_rate=a__			,				norm='''slaney'''			,				mel_scale='''slaney'''			,				)



 def 			snake_case__  (   self		:   List[Any]			,				a__		:   np.array      ):
      __magic_name__						       =    spectrogram(
          a__			,				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'''			,				)
      __magic_name__						       =    log_spec[:, :-1]
      __magic_name__						       =    np.maximum(a__			,				log_spec.max() - 8.0      )
      __magic_name__						       =    (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 			snake_case__  (   a__		:   List[np.ndarray]			,				a__		:   List[np.ndarray]			,				a__		:   float = 0.0      ):
      if attention_mask is not None:
           __magic_name__						       =    np.array(a__			,				np.intaa      )
           __magic_name__						       =    []

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

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

      return normed_input_values



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

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

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

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

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

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

      # convert into correct format for padding

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

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

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

      __magic_name__						       =    [self._np_extract_fbank_features(a__      ) for waveform in input_features[0]]

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

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

      if return_tensors is not None:
           __magic_name__						       =    padded_inputs.convert_to_tensors(a__      )

      return padded_inputs



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









'''simple docstring'''




import json
import sys
import tempfile
import unittest
from pathlib import Path

import transformers
from transformers import (
    CONFIG_MAPPING,
    FEATURE_EXTRACTOR_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
)
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER, get_tests_dir


sys.path.append(str(Path(__file__).parent.parent.parent.parent / "utils"))

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


_lowerCAmelCase         =						get_tests_dir("fixtures")
_lowerCAmelCase         =						get_tests_dir("fixtures/dummy_feature_extractor_config.json")
_lowerCAmelCase         =						get_tests_dir("fixtures/dummy-config.json")






class   _SCREAMING_SNAKE_CASE   (   unittest.TestCase     ):



 def 			snake_case__  (   self		:   Union[str, Any]      ):
      __magic_name__						       =    0



 def 			snake_case__  (   self		:   Optional[int]      ):
      __magic_name__						       =    AutoFeatureExtractor.from_pretrained('''facebook/wav2vec2-base-960h'''      )
      self.assertIsInstance(a__			,				a__      )



 def 			snake_case__  (   self		:   Optional[int]      ):
      __magic_name__						       =    AutoFeatureExtractor.from_pretrained(a__      )
      self.assertIsInstance(a__			,				a__      )



 def 			snake_case__  (   self		:   Optional[Any]      ):
      with tempfile.TemporaryDirectory() as tmpdirname:
           __magic_name__						       =    WavaVecaConfig()

           # remove feature_extractor_type to make sure config.json alone is enough to load feature processor locally
           __magic_name__						       =    AutoFeatureExtractor.from_pretrained(a__      ).to_dict()

           config_dict.pop('''feature_extractor_type'''      )
           __magic_name__						       =    WavaVecaFeatureExtractor(**a__      )

           # save in new folder
           model_config.save_pretrained(a__      )
           config.save_pretrained(a__      )

           __magic_name__						       =    AutoFeatureExtractor.from_pretrained(a__      )

           # make sure private variable is not incorrectly saved
           __magic_name__						       =    json.loads(config.to_json_string()      )
           self.assertTrue('''_processor_class''' not in dict_as_saved      )

      self.assertIsInstance(a__			,				a__      )



 def 			snake_case__  (   self		:   Optional[Any]      ):
      __magic_name__						       =    AutoFeatureExtractor.from_pretrained(a__      )
      self.assertIsInstance(a__			,				a__      )



 def 			snake_case__  (   self		:   str      ):
      with self.assertRaisesRegex(
          a__			,				'''bert-base is not a local folder and is not a valid model identifier'''      ):
           __magic_name__						       =    AutoFeatureExtractor.from_pretrained('''bert-base'''      )



 def 			snake_case__  (   self		:   str      ):
      with self.assertRaisesRegex(
          a__			,				r'''aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)'''      ):
           __magic_name__						       =    AutoFeatureExtractor.from_pretrained(a__			,				revision='''aaaaaa'''      )



 def 			snake_case__  (   self		:   Union[str, Any]      ):
      with self.assertRaisesRegex(
          a__			,				'''hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json.'''			,				):
           __magic_name__						       =    AutoFeatureExtractor.from_pretrained('''hf-internal-testing/config-no-model'''      )



 def 			snake_case__  (   self		:   Dict      ):
      # If remote code is not set, we will time out when asking whether to load the model.
      with self.assertRaises(a__      ):
           __magic_name__						       =    AutoFeatureExtractor.from_pretrained(
               '''hf-internal-testing/test_dynamic_feature_extractor'''      )
      # If remote code is disabled, we can't load this config.
      with self.assertRaises(a__      ):
           __magic_name__						       =    AutoFeatureExtractor.from_pretrained(
               '''hf-internal-testing/test_dynamic_feature_extractor'''			,				trust_remote_code=a__      )

      __magic_name__						       =    AutoFeatureExtractor.from_pretrained(
          '''hf-internal-testing/test_dynamic_feature_extractor'''			,				trust_remote_code=a__      )
      self.assertEqual(feature_extractor.__class__.__name__			,				'''NewFeatureExtractor'''      )

      # Test feature extractor can be reloaded.
      with tempfile.TemporaryDirectory() as tmp_dir:
           feature_extractor.save_pretrained(a__      )
           __magic_name__						       =    AutoFeatureExtractor.from_pretrained(a__			,				trust_remote_code=a__      )
      self.assertEqual(reloaded_feature_extractor.__class__.__name__			,				'''NewFeatureExtractor'''      )



 def 			snake_case__  (   self		:   int      ):
      try:
           AutoConfig.register('''custom'''			,				a__      )
           AutoFeatureExtractor.register(a__			,				a__      )
           # Trying to register something existing in the Transformers library will raise an error
           with self.assertRaises(a__      ):
                AutoFeatureExtractor.register(a__			,				a__      )

           # Now that the config is registered, it can be used as any other config with the auto-API
           __magic_name__						       =    CustomFeatureExtractor.from_pretrained(a__      )
           with tempfile.TemporaryDirectory() as tmp_dir:
                feature_extractor.save_pretrained(a__      )
                __magic_name__						       =    AutoFeatureExtractor.from_pretrained(a__      )
                self.assertIsInstance(a__			,				a__      )

      finally:
           if "custom" in CONFIG_MAPPING._extra_content:
                del CONFIG_MAPPING._extra_content["custom"]
           if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
                del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]



 def 			snake_case__  (   self		:   int      ):






      class   _SCREAMING_SNAKE_CASE   (   __a     ):
       __SCREAMING_SNAKE_CASE					:Optional[int]      	=					True

      try:
           AutoConfig.register('''custom'''			,				a__      )
           AutoFeatureExtractor.register(a__			,				a__      )
           # If remote code is not set, the default is to use local
           __magic_name__						       =    AutoFeatureExtractor.from_pretrained(
               '''hf-internal-testing/test_dynamic_feature_extractor'''      )
           self.assertEqual(feature_extractor.__class__.__name__			,				'''NewFeatureExtractor'''      )
           self.assertTrue(feature_extractor.is_local      )

           # If remote code is disabled, we load the local one.
           __magic_name__						       =    AutoFeatureExtractor.from_pretrained(
               '''hf-internal-testing/test_dynamic_feature_extractor'''			,				trust_remote_code=a__      )
           self.assertEqual(feature_extractor.__class__.__name__			,				'''NewFeatureExtractor'''      )
           self.assertTrue(feature_extractor.is_local      )

           # If remote is enabled, we load from the Hub
           __magic_name__						       =    AutoFeatureExtractor.from_pretrained(
               '''hf-internal-testing/test_dynamic_feature_extractor'''			,				trust_remote_code=a__      )
           self.assertEqual(feature_extractor.__class__.__name__			,				'''NewFeatureExtractor'''      )
           self.assertTrue(not hasattr(a__			,				'''is_local'''      )      )

      finally:
           if "custom" in CONFIG_MAPPING._extra_content:
                del CONFIG_MAPPING._extra_content["custom"]
           if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
                del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]






