



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


UpperCAmelCase_									=							get_tests_dir("fixtures")





class 		UpperCAmelCase						(						unittest.TestCase				):





   def 	__lowerCAmelCase							(	self			):
         _lowerCAmelCase 			=  mock.Mock()
         _lowerCAmelCase 			=  500
         _lowerCAmelCase 			=  {}
         _lowerCAmelCase 			=  HTTPError
         _lowerCAmelCase 			=  {}

         # Download this model to make sure it's in the cache.
         _lowerCAmelCase 			=  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=_lowerCamelCase			) as mock_head:
               _lowerCAmelCase 			=  WavaVecaFeatureExtractor.from_pretrained('''hf-internal-testing/tiny-random-wav2vec2'''			)
               # This check we did call the fake head request
               mock_head.assert_called()





   def 	__lowerCAmelCase							(	self			):
         _lowerCAmelCase 			=  WavaVecaFeatureExtractor.from_pretrained(
             '''https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json'''			)






@is_staging_test
class 		UpperCAmelCase						(						unittest.TestCase				):





   @classmethod
   def 	__lowerCAmelCase							(	cls			):
         _lowerCAmelCase 			=  TOKEN
         HfFolder.save_token(_lowerCamelCase			)





   @classmethod
   def 	__lowerCAmelCase							(	cls			):





         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 	__lowerCAmelCase							(	self			):
         _lowerCAmelCase 			=  WavaVecaFeatureExtractor.from_pretrained(_lowerCamelCase			)
         feature_extractor.push_to_hub('''test-feature-extractor'''       ,     use_auth_token=self._token			)

         _lowerCAmelCase 			=  WavaVecaFeatureExtractor.from_pretrained(F'''{USER}/test-feature-extractor'''			)
         for k, v in feature_extractor.__dict__.items():
               self.assertEqual(_lowerCamelCase       ,     getattr(_lowerCamelCase       ,     _lowerCamelCase			)			)

         # 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(
                   _lowerCamelCase       ,     repo_id='''test-feature-extractor'''       ,     push_to_hub=_lowerCamelCase       ,     use_auth_token=self._token			)

         _lowerCAmelCase 			=  WavaVecaFeatureExtractor.from_pretrained(F'''{USER}/test-feature-extractor'''			)
         for k, v in feature_extractor.__dict__.items():
               self.assertEqual(_lowerCamelCase       ,     getattr(_lowerCamelCase       ,     _lowerCamelCase			)			)





   def 	__lowerCAmelCase							(	self			):
         _lowerCAmelCase 			=  WavaVecaFeatureExtractor.from_pretrained(_lowerCamelCase			)
         feature_extractor.push_to_hub('''valid_org/test-feature-extractor'''       ,     use_auth_token=self._token			)

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

         # 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(
                   _lowerCamelCase       ,     repo_id='''valid_org/test-feature-extractor-org'''       ,     push_to_hub=_lowerCamelCase       ,     use_auth_token=self._token			)

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





   def 	__lowerCAmelCase							(	self			):
         CustomFeatureExtractor.register_for_auto_class()
         _lowerCAmelCase 			=  CustomFeatureExtractor.from_pretrained(_lowerCamelCase			)

         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'''}       ,     )

         _lowerCAmelCase 			=  AutoFeatureExtractor.from_pretrained(
             F'''{USER}/test-dynamic-feature-extractor'''       ,     trust_remote_code=_lowerCamelCase			)
         # 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'''			)











import functools
import gc
import inspect

import torch

from .imports import is_npu_available, is_xpu_available
def 							UpperCAmelCase__		(				*_SCREAMING_SNAKE_CASE   :	Tuple				)->List[Any]:
      if not isinstance(_SCREAMING_SNAKE_CASE     ,				_SCREAMING_SNAKE_CASE				):
            _lowerCAmelCase 			=  list(_SCREAMING_SNAKE_CASE				)
      for i in range(len(_SCREAMING_SNAKE_CASE				)				):
            _lowerCAmelCase 			=  None
      gc.collect()
      if is_xpu_available():
            torch.xpu.empty_cache()
      elif is_npu_available():
            torch.npu.empty_cache()
      else:
            torch.cuda.empty_cache()
      return objects
def 							UpperCAmelCase__		(				_SCREAMING_SNAKE_CASE   :	Exception				)->bool:
      _lowerCAmelCase 			=  [
          '''CUDA out of memory.''',  # CUDA OOM
          '''cuDNN error: CUDNN_STATUS_NOT_SUPPORTED.''',  # CUDNN SNAFU
          '''DefaultCPUAllocator: can\'t allocate memory''',  # CPU OOM
      ]
      if isinstance(_SCREAMING_SNAKE_CASE     ,				_SCREAMING_SNAKE_CASE				) and len(exception.args				) == 1:
            return any(err in exception.args[0] for err in _statements				)
      return False


def 							UpperCAmelCase__		(				_SCREAMING_SNAKE_CASE   :	callable = None     ,				_SCREAMING_SNAKE_CASE   :	int = 1_2_8				)->Optional[int]:
      if function is None:
            return functools.partial(_SCREAMING_SNAKE_CASE     ,				starting_batch_size=_SCREAMING_SNAKE_CASE				)

      _lowerCAmelCase 			=  starting_batch_size

      def decorator(*_SCREAMING_SNAKE_CASE   :	Optional[int]     ,				**_SCREAMING_SNAKE_CASE   :	Optional[Any]				):
            nonlocal batch_size
            gc.collect()
            if is_xpu_available():
                  torch.xpu.empty_cache()
            elif is_npu_available():
                  torch.npu.empty_cache()
            else:
                  torch.cuda.empty_cache()
            _lowerCAmelCase 			=  list(inspect.signature(_SCREAMING_SNAKE_CASE				).parameters.keys()				)
            # Guard against user error
            if len(_SCREAMING_SNAKE_CASE				) < (len(_SCREAMING_SNAKE_CASE				) + 1):
                  _lowerCAmelCase 			=  ''', '''.join([f'''{arg}={value}''' for arg, value in zip(params[1:]     ,				args[1:]				)]				)
                  raise TypeError(
                      f'''Batch size was passed into `{function.__name__}` as the first argument when called.'''
                      f'''Remove this as the decorator already does so: `{function.__name__}({arg_str})`'''				)
            while True:
                  if batch_size == 0:
                        raise RuntimeError('''No executable batch size found, reached zero.'''				)
                  try:
                        return function(_SCREAMING_SNAKE_CASE     ,				*_SCREAMING_SNAKE_CASE     ,				**_SCREAMING_SNAKE_CASE				)
                  except Exception as e:
                        if should_reduce_batch_size(_SCREAMING_SNAKE_CASE				):
                              gc.collect()
                              if is_xpu_available():
                                    torch.xpu.empty_cache()
                              elif is_npu_available():
                                    torch.npu.empty_cache()
                              else:
                                    torch.cuda.empty_cache()
                              batch_size //= 2
                        else:
                              raise

      return decorator