


"""simple docstring"""







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


lowerCAmelCase__      				=   get_tests_dir('''fixtures''')


class       SCREAMING_SNAKE_CASE__				(		unittest.TestCase				):
       """simple docstring"""


       def        lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase     :     Tuple								=							mock.Mock()
         lowerCAmelCase     :     List[str]								=							500
         lowerCAmelCase     :     List[str]								=							{}
         lowerCAmelCase     :     Optional[int]								=							HTTPError
         lowerCAmelCase     :     Tuple								=							{}

         # Download this model to make sure it's in the cache.
         lowerCAmelCase     :     str								=							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:
           lowerCAmelCase     :     Any								=							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						):






         """simple docstring"""





         lowerCAmelCase     :     List[str]								=							WavaVecaFeatureExtractor.from_pretrained(
             "https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json"						)


@is_staging_test
class       SCREAMING_SNAKE_CASE__				(		unittest.TestCase				):
       """simple docstring"""


       @classmethod
       def        lowercase__							(						cls						):






         """simple docstring"""





         lowerCAmelCase     :     Optional[Any]								=							TOKEN
         HfFolder.save_token(snake_case__						)

       @classmethod
       def        lowercase__							(						cls						):






         """simple docstring"""





         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						):






         """simple docstring"""





         lowerCAmelCase     :     str								=							WavaVecaFeatureExtractor.from_pretrained(snake_case__						)
         feature_extractor.push_to_hub("test-feature-extractor"		,     use_auth_token=self._token						)

         lowerCAmelCase     :     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__						)						)

         # 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						)

         lowerCAmelCase     :     int								=							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						):






         """simple docstring"""





         lowerCAmelCase     :     int								=							WavaVecaFeatureExtractor.from_pretrained(snake_case__						)
         feature_extractor.push_to_hub("valid_org/test-feature-extractor"		,     use_auth_token=self._token						)

         lowerCAmelCase     :     Optional[Any]								=							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						)

         lowerCAmelCase     :     Dict								=							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						):






         """simple docstring"""





         CustomFeatureExtractor.register_for_auto_class()
         lowerCAmelCase     :     Tuple								=							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"}		,     )

         lowerCAmelCase     :     str								=							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



def     a__ (				snake_case__       , snake_case__				)						->					Optional[int]:
			lowerCamelCase			      =					list(range(0       , snake_case__				)				)

			lowerCamelCase			      =					[item for sublist in list(device_map.values()				) for item in sublist]

			# Duplicate check
			lowerCamelCase			      =					[]
			for i in device_map_blocks:
						if device_map_blocks.count(snake_case__				) > 1 and i not in duplicate_blocks:
									duplicate_blocks.append(snake_case__				)
    # Missing blocks
			lowerCamelCase			      =					[i for i in blocks if i not in device_map_blocks]
			lowerCamelCase			      =					[i for i in device_map_blocks if i not in blocks]

			if len(snake_case__				) != 0:
						raise ValueError(
						    """Duplicate attention blocks specified in device_map. Attention blocks must be specified to one device."""
						    """ These attention blocks were specified more than once: """ + str(snake_case__				)				)
			if len(snake_case__				) != 0:
						raise ValueError(
						    """There are attention blocks for this model that are not specified in the device_map. Add these attention """
						    """blocks to a device on the device_map: """ + str(snake_case__				)				)
			if len(snake_case__				) != 0:
						raise ValueError(
						    """The device_map contains more attention blocks than this model has. Remove these from the device_map:"""
						    + str(snake_case__				)				)



def     a__ (				snake_case__       , snake_case__				)						->					List[Any]:
			lowerCamelCase			      =					list(range(snake_case__				)				)
			lowerCamelCase			      =					int(ceil(n_layers / len(snake_case__				)				)				)
			lowerCamelCase			      =					[layers[i : i + n_blocks] for i in range(0       , snake_case__       , snake_case__				)]

			return dict(zip(snake_case__       , snake_case__				)				)



