



import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoFeatureExtractor, WavaVecaFeatureExtractor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


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

from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


a_  :Optional[Any]     = get_tests_dir("fixtures")




class    snake_case__		(			unittest.TestCase	):






     """simple docstring"""


     def   lowercase_  (			self    :    Union[str, Any]   )  ->Dict:
            snake_case__				:      Optional[Any]        =      mock.Mock()
            snake_case__				:      Tuple        =      5_0_0
            snake_case__				:      int        =      {}
            snake_case__				:      Union[str, Any]        =      HTTPError
            snake_case__				:      Tuple        =      {}

            # Download this model to make sure it's in the cache.
            snake_case__				:      Optional[Any]        =      WavaVecaFeatureExtractor.from_pretrained('hf-internal-testing/tiny-random-wav2vec2'   )
            # Under the mock environment we get a 500 error when trying to reach the model.
            with mock.patch('requests.Session.request',					return_value=_snake_case   ) as mock_head:
                   snake_case__				:      int        =      WavaVecaFeatureExtractor.from_pretrained('hf-internal-testing/tiny-random-wav2vec2'   )
                   # This check we did call the fake head request
                   mock_head.assert_called()






     def   lowercase_  (			self    :    str   )  ->List[str]:
            snake_case__				:      Optional[int]        =      WavaVecaFeatureExtractor.from_pretrained(
                'https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json'   )




@is_staging_test
class    snake_case__		(			unittest.TestCase	):






     """simple docstring"""


     @classmethod
     def   lowercase_  (			cls    :    List[Any]   )  ->Tuple:
            snake_case__				:      int        =      TOKEN
            HfFolder.save_token(_snake_case   )






     @classmethod
     def   lowercase_  (			cls    :    str   )  ->Any:







            try:
                   delete_repo(token=cls._token,					repo_id='test-feature-extractor'   )
            except HTTPError:
                   pass

            try:
                   delete_repo(token=cls._token,					repo_id='valid_org/test-feature-extractor-org'   )
            except HTTPError:
                   pass

            try:
                   delete_repo(token=cls._token,					repo_id='test-dynamic-feature-extractor'   )
            except HTTPError:
                   pass






     def   lowercase_  (			self    :    int   )  ->Any:
            snake_case__				:      Union[str, Any]        =      WavaVecaFeatureExtractor.from_pretrained(_snake_case   )
            feature_extractor.push_to_hub('test-feature-extractor',					use_auth_token=self._token   )

            snake_case__				:      Union[str, Any]        =      WavaVecaFeatureExtractor.from_pretrained(F'''{USER}/test-feature-extractor'''   )
            for k, v in feature_extractor.__dict__.items():
                   self.assertEqual(_snake_case,					getattr(_snake_case,					_snake_case   )   )

            # Reset repo
            delete_repo(token=self._token,					repo_id='test-feature-extractor'   )

            # Push to hub via save_pretrained
            with tempfile.TemporaryDirectory() as tmp_dir:
                   feature_extractor.save_pretrained(
                       _snake_case,					repo_id='test-feature-extractor',					push_to_hub=_snake_case,					use_auth_token=self._token   )

            snake_case__				:      List[Any]        =      WavaVecaFeatureExtractor.from_pretrained(F'''{USER}/test-feature-extractor'''   )
            for k, v in feature_extractor.__dict__.items():
                   self.assertEqual(_snake_case,					getattr(_snake_case,					_snake_case   )   )






     def   lowercase_  (			self    :    Any   )  ->Optional[Any]:
            snake_case__				:      Dict        =      WavaVecaFeatureExtractor.from_pretrained(_snake_case   )
            feature_extractor.push_to_hub('valid_org/test-feature-extractor',					use_auth_token=self._token   )

            snake_case__				:      int        =      WavaVecaFeatureExtractor.from_pretrained('valid_org/test-feature-extractor'   )
            for k, v in feature_extractor.__dict__.items():
                   self.assertEqual(_snake_case,					getattr(_snake_case,					_snake_case   )   )

            # Reset repo
            delete_repo(token=self._token,					repo_id='valid_org/test-feature-extractor'   )

            # Push to hub via save_pretrained
            with tempfile.TemporaryDirectory() as tmp_dir:
                   feature_extractor.save_pretrained(
                       _snake_case,					repo_id='valid_org/test-feature-extractor-org',					push_to_hub=_snake_case,					use_auth_token=self._token   )

            snake_case__				:      str        =      WavaVecaFeatureExtractor.from_pretrained('valid_org/test-feature-extractor-org'   )
            for k, v in feature_extractor.__dict__.items():
                   self.assertEqual(_snake_case,					getattr(_snake_case,					_snake_case   )   )






     def   lowercase_  (			self    :    Any   )  ->Optional[int]:
            CustomFeatureExtractor.register_for_auto_class()
            snake_case__				:      str        =      CustomFeatureExtractor.from_pretrained(_snake_case   )

            feature_extractor.push_to_hub('test-dynamic-feature-extractor',					use_auth_token=self._token   )

            # This has added the proper auto_map field to the config
            self.assertDictEqual(
                feature_extractor.auto_map,					{'AutoFeatureExtractor': 'custom_feature_extraction.CustomFeatureExtractor'},					)

            snake_case__				:      Any        =      AutoFeatureExtractor.from_pretrained(
                F'''{USER}/test-dynamic-feature-extractor''',					trust_remote_code=_snake_case   )
            # Can't make an isinstance check because the new_feature_extractor is from the CustomFeatureExtractor class of a dynamic module
            self.assertEqual(new_feature_extractor.__class__.__name__,					'CustomFeatureExtractor'   )












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


UpperCamelCase_    	=				logging.get_logger(__name__)






class 							_a				(			SCREAMING_SNAKE_CASE ):





       '''simple docstring'''






       A						:				str        =						['''audio_values''', '''audio_mask''']







       def __init__(      self,							A=2_048,							A=1,							A=[16, 16],							A=128,							A=44_100,							A=86,							A=2_048,							A=0.0,							**A,							):







             '''simple docstring'''







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

             SCREAMING_SNAKE_CASE   :      str    					=						spectrogram_length
             SCREAMING_SNAKE_CASE   :      Optional[Any]    					=						num_channels
             SCREAMING_SNAKE_CASE   :      List[str]    					=						patch_size
             SCREAMING_SNAKE_CASE   :      Optional[int]    					=						feature_size // self.patch_size[1]
             SCREAMING_SNAKE_CASE   :      Dict    					=						n_fft
             SCREAMING_SNAKE_CASE   :      Tuple    					=						sampling_rate // hop_length_to_sampling_rate
             SCREAMING_SNAKE_CASE   :      str    					=						sampling_rate
             SCREAMING_SNAKE_CASE   :      int    					=						padding_value
             SCREAMING_SNAKE_CASE   :      Any    					=						mel_filter_bank(
                 num_frequency_bins=1 + n_fft // 2,							num_mel_filters=A,							min_frequency=0.0,							max_frequency=2_20_50.0,							sampling_rate=A,							norm='slaney',							mel_scale='slaney',							).T







       def        UpperCamelCase_							(      self,							A  ):







             '''simple docstring'''







             SCREAMING_SNAKE_CASE   :      Optional[int]    					=						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.T,							log_mel='dB',							db_range=80.0,							)
             SCREAMING_SNAKE_CASE   :      Union[str, Any]    					=						log_spec[:, :-1]
             SCREAMING_SNAKE_CASE   :      List[Any]    					=						log_spec - 20.0
             SCREAMING_SNAKE_CASE   :      Optional[Any]    					=						np.clip(log_spec / 40.0,							-2.0,							0.0  ) + 1.0
             return log_spec







       def __call__(      self,							A,							A = None,							A = True,							A = None,							A = False,							A = False,							**A,							):







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

             SCREAMING_SNAKE_CASE   :      List[Any]    					=						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}"  )
             SCREAMING_SNAKE_CASE   :      int    					=						is_batched_numpy or (
                 isinstance(A,							(list, tuple)  ) and (isinstance(raw_speech[0],							(np.ndarray, tuple, list)  ))
             )
             if is_batched:
                   SCREAMING_SNAKE_CASE   :      Union[str, Any]    					=						[np.asarray([speech],							dtype=np.floataa  ).T for speech in raw_speech]
             elif not is_batched and not isinstance(A,							np.ndarray  ):
                   SCREAMING_SNAKE_CASE   :      Any    					=						np.asarray(A,							dtype=np.floataa  )
             elif isinstance(A,							np.ndarray  ) and raw_speech.dtype is np.dtype(np.floataa  ):
                   SCREAMING_SNAKE_CASE   :      Optional[Any]    					=						raw_speech.astype(np.floataa  )
             # always return batch
             if not is_batched:
                   SCREAMING_SNAKE_CASE   :      Union[str, Any]    					=						[np.asarray([raw_speech]  ).T]

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

             # Create audio attention mask
             SCREAMING_SNAKE_CASE   :      Tuple    					=						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:
                   SCREAMING_SNAKE_CASE   :      List[Any]    					=						[
                       (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
                   ]
                   SCREAMING_SNAKE_CASE   :      Tuple    					=						np.array(A  ).astype(np.floataa  )

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

             # return as BatchFeature
             if return_attention_mask:
                   SCREAMING_SNAKE_CASE   :      Any    					=						{'audio_values': padded_audio_features, 'audio_mask': audio_mask}
             else:
                   SCREAMING_SNAKE_CASE   :      Dict    					=						{'audio_values': padded_audio_features}

             SCREAMING_SNAKE_CASE   :      str    					=						BatchFeature(data=A,							tensor_type=A  )
             return encoded_inputs


