





import json
import os
import sys
import tempfile
import unittest
from pathlib import Path
from shutil import copyfile

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

import transformers
from transformers import (
    CONFIG_MAPPING,
    FEATURE_EXTRACTOR_MAPPING,
    PROCESSOR_MAPPING,
    TOKENIZER_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    AutoProcessor,
    AutoTokenizer,
    BertTokenizer,
    ProcessorMixin,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
    WavaVecaProcessor,
)
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test
from transformers.tokenization_utils import TOKENIZER_CONFIG_FILE
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_tokenizers_available


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
from test_module.custom_processing import CustomProcessor  # noqa E402
from test_module.custom_tokenization import CustomTokenizer  # noqa E402


_UpperCAmelCase							:			Optional[Any]        =							get_tests_dir("""fixtures/dummy_feature_extractor_config.json""")
_UpperCAmelCase							:			List[str]        =							get_tests_dir("""fixtures/vocab.json""")
_UpperCAmelCase							:			Optional[Any]        =							get_tests_dir("""fixtures""")

class   lowerCAmelCase				(   unittest.TestCase     ):
 UpperCAmelCase__						=						["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]
 def 	A_      ( self    : Tuple      )     ->      List[Any]:
     lowerCamelCase__       : Tuple   					=						0
 def 	A_      ( self    : int      )     ->      Union[str, Any]:
     lowerCamelCase__       : str   					=						AutoProcessor.from_pretrained('facebook/wav2vec2-base-960h'      )
     self.assertIsInstance(UpperCAmelCase     ,	UpperCAmelCase      )
 def 	A_      ( self    : Optional[int]      )     ->      int:
     with tempfile.TemporaryDirectory() as tmpdirname:
         lowerCamelCase__       : Union[str, Any]   					=						WavaVecaConfig()
         lowerCamelCase__       : List[Any]   					=						AutoProcessor.from_pretrained('facebook/wav2vec2-base-960h'      )

         # save in new folder
         model_config.save_pretrained(UpperCAmelCase      )
         processor.save_pretrained(UpperCAmelCase      )

         lowerCamelCase__       : str   					=						AutoProcessor.from_pretrained(UpperCAmelCase      )

     self.assertIsInstance(UpperCAmelCase     ,	UpperCAmelCase      )
 def 	A_      ( self    : Dict      )     ->      Dict:
     with tempfile.TemporaryDirectory() as tmpdirname:
         # copy relevant files
         copyfile(UpperCAmelCase     ,	os.path.join(UpperCAmelCase     ,	UpperCAmelCase      )      )
         copyfile(UpperCAmelCase     ,	os.path.join(UpperCAmelCase     ,	'vocab.json'      )      )

         lowerCamelCase__       : Optional[int]   					=						AutoProcessor.from_pretrained(UpperCAmelCase      )

     self.assertIsInstance(UpperCAmelCase     ,	UpperCAmelCase      )
 def 	A_      ( self    : Dict      )     ->      Optional[int]:
     with tempfile.TemporaryDirectory() as tmpdirname:
         lowerCamelCase__       : Dict   					=						WavaVecaFeatureExtractor()
         lowerCamelCase__       : List[str]   					=						AutoTokenizer.from_pretrained('facebook/wav2vec2-base-960h'      )

         lowerCamelCase__       : List[str]   					=						WavaVecaProcessor(UpperCAmelCase     ,	UpperCAmelCase      )

         # save in new folder
         processor.save_pretrained(UpperCAmelCase      )

         # drop `processor_class` in tokenizer
         with open(os.path.join(UpperCAmelCase     ,	UpperCAmelCase      )     ,	'r'      ) as f:
             lowerCamelCase__       : Any   					=						json.load(UpperCAmelCase      )
             config_dict.pop('processor_class'      )

         with open(os.path.join(UpperCAmelCase     ,	UpperCAmelCase      )     ,	'w'      ) as f:
             f.write(json.dumps(UpperCAmelCase      )      )

         lowerCamelCase__       : int   					=						AutoProcessor.from_pretrained(UpperCAmelCase      )

     self.assertIsInstance(UpperCAmelCase     ,	UpperCAmelCase      )
 def 	A_      ( self    : str      )     ->      Tuple:
     with tempfile.TemporaryDirectory() as tmpdirname:
         lowerCamelCase__       : Tuple   					=						WavaVecaFeatureExtractor()
         lowerCamelCase__       : Any   					=						AutoTokenizer.from_pretrained('facebook/wav2vec2-base-960h'      )

         lowerCamelCase__       : Optional[int]   					=						WavaVecaProcessor(UpperCAmelCase     ,	UpperCAmelCase      )

         # save in new folder
         processor.save_pretrained(UpperCAmelCase      )

         # drop `processor_class` in feature extractor
         with open(os.path.join(UpperCAmelCase     ,	UpperCAmelCase      )     ,	'r'      ) as f:
             lowerCamelCase__       : List[Any]   					=						json.load(UpperCAmelCase      )
             config_dict.pop('processor_class'      )

         with open(os.path.join(UpperCAmelCase     ,	UpperCAmelCase      )     ,	'w'      ) as f:
             f.write(json.dumps(UpperCAmelCase      )      )

         lowerCamelCase__       : Union[str, Any]   					=						AutoProcessor.from_pretrained(UpperCAmelCase      )

     self.assertIsInstance(UpperCAmelCase     ,	UpperCAmelCase      )
 def 	A_      ( self    : Optional[int]      )     ->      Any:
     with tempfile.TemporaryDirectory() as tmpdirname:
         lowerCamelCase__       : int   					=						WavaVecaConfig(processor_class='Wav2Vec2Processor'      )
         model_config.save_pretrained(UpperCAmelCase      )
         # copy relevant files
         copyfile(UpperCAmelCase     ,	os.path.join(UpperCAmelCase     ,	'vocab.json'      )      )
         # create emtpy sample processor
         with open(os.path.join(UpperCAmelCase     ,	UpperCAmelCase      )     ,	'w'      ) as f:
             f.write('{}'      )

         lowerCamelCase__       : Optional[int]   					=						AutoProcessor.from_pretrained(UpperCAmelCase      )

     self.assertIsInstance(UpperCAmelCase     ,	UpperCAmelCase      )
 def 	A_      ( self    : int      )     ->      Union[str, Any]:
     # If remote code is not set, we will time out when asking whether to load the model.
     with self.assertRaises(UpperCAmelCase      ):
         lowerCamelCase__       : Any   					=						AutoProcessor.from_pretrained('hf-internal-testing/test_dynamic_processor'      )
     # If remote code is disabled, we can't load this config.
     with self.assertRaises(UpperCAmelCase      ):
         lowerCamelCase__       : Union[str, Any]   					=						AutoProcessor.from_pretrained(
             'hf-internal-testing/test_dynamic_processor'     ,	trust_remote_code=UpperCAmelCase      )

     lowerCamelCase__       : Dict   					=						AutoProcessor.from_pretrained('hf-internal-testing/test_dynamic_processor'     ,	trust_remote_code=UpperCAmelCase      )
     self.assertTrue(processor.special_attribute_present      )
     self.assertEqual(processor.__class__.__name__     ,	'NewProcessor'      )

     lowerCamelCase__       : int   					=						processor.feature_extractor
     self.assertTrue(feature_extractor.special_attribute_present      )
     self.assertEqual(feature_extractor.__class__.__name__     ,	'NewFeatureExtractor'      )

     lowerCamelCase__       : str   					=						processor.tokenizer
     self.assertTrue(tokenizer.special_attribute_present      )
     if is_tokenizers_available():
         self.assertEqual(tokenizer.__class__.__name__     ,	'NewTokenizerFast'      )

         # Test we can also load the slow version
         lowerCamelCase__       : List[Any]   					=						AutoProcessor.from_pretrained(
             'hf-internal-testing/test_dynamic_processor'     ,	trust_remote_code=UpperCAmelCase     ,	use_fast=UpperCAmelCase      )
         lowerCamelCase__       : Any   					=						new_processor.tokenizer
         self.assertTrue(new_tokenizer.special_attribute_present      )
         self.assertEqual(new_tokenizer.__class__.__name__     ,	'NewTokenizer'      )
     else:
         self.assertEqual(tokenizer.__class__.__name__     ,	'NewTokenizer'      )
 def 	A_      ( self    : Optional[int]      )     ->      Dict:
     try:
         AutoConfig.register('custom'     ,	UpperCAmelCase      )
         AutoFeatureExtractor.register(UpperCAmelCase     ,	UpperCAmelCase      )
         AutoTokenizer.register(UpperCAmelCase     ,	slow_tokenizer_class=UpperCAmelCase      )
         AutoProcessor.register(UpperCAmelCase     ,	UpperCAmelCase      )
         # Trying to register something existing in the Transformers library will raise an error
         with self.assertRaises(UpperCAmelCase      ):
             AutoProcessor.register(UpperCAmelCase     ,	UpperCAmelCase      )

         # Now that the config is registered, it can be used as any other config with the auto-API
         lowerCamelCase__       : Dict   					=						CustomFeatureExtractor.from_pretrained(UpperCAmelCase      )

         with tempfile.TemporaryDirectory() as tmp_dir:
             lowerCamelCase__       : List[Any]   					=						os.path.join(UpperCAmelCase     ,	'vocab.txt'      )
             with open(UpperCAmelCase     ,	'w'     ,	encoding='utf-8'      ) as vocab_writer:
                 vocab_writer.write(''.join([x + '\n' for x in self.vocab_tokens]      )      )
             lowerCamelCase__       : Any   					=						CustomTokenizer(UpperCAmelCase      )

         lowerCamelCase__       : Optional[Any]   					=						CustomProcessor(UpperCAmelCase     ,	UpperCAmelCase      )

         with tempfile.TemporaryDirectory() as tmp_dir:
             processor.save_pretrained(UpperCAmelCase      )
             lowerCamelCase__       : str   					=						AutoProcessor.from_pretrained(UpperCAmelCase      )
             self.assertIsInstance(UpperCAmelCase     ,	UpperCAmelCase      )

     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]
         if CustomConfig in TOKENIZER_MAPPING._extra_content:
             del TOKENIZER_MAPPING._extra_content[CustomConfig]
         if CustomConfig in PROCESSOR_MAPPING._extra_content:
             del PROCESSOR_MAPPING._extra_content[CustomConfig]
 def 	A_      ( self    : int      )     ->      List[str]:

     class   lowerCAmelCase				(   __UpperCamelCase     ):
      UpperCAmelCase__						=						False


     class   lowerCAmelCase				(   __UpperCamelCase     ):
      UpperCAmelCase__						=						False


     class   lowerCAmelCase				(   __UpperCamelCase     ):
      UpperCAmelCase__						=						"""AutoFeatureExtractor"""
      UpperCAmelCase__						=						"""AutoTokenizer"""
      UpperCAmelCase__						=						False

     try:
         AutoConfig.register('custom'     ,	UpperCAmelCase      )
         AutoFeatureExtractor.register(UpperCAmelCase     ,	UpperCAmelCase      )
         AutoTokenizer.register(UpperCAmelCase     ,	slow_tokenizer_class=UpperCAmelCase      )
         AutoProcessor.register(UpperCAmelCase     ,	UpperCAmelCase      )
         # If remote code is not set, the default is to use local classes.
         lowerCamelCase__       : Optional[int]   					=						AutoProcessor.from_pretrained('hf-internal-testing/test_dynamic_processor'      )
         self.assertEqual(processor.__class__.__name__     ,	'NewProcessor'      )
         self.assertFalse(processor.special_attribute_present      )
         self.assertFalse(processor.feature_extractor.special_attribute_present      )
         self.assertFalse(processor.tokenizer.special_attribute_present      )

         # If remote code is disabled, we load the local ones.
         lowerCamelCase__       : Any   					=						AutoProcessor.from_pretrained(
             'hf-internal-testing/test_dynamic_processor'     ,	trust_remote_code=UpperCAmelCase      )
         self.assertEqual(processor.__class__.__name__     ,	'NewProcessor'      )
         self.assertFalse(processor.special_attribute_present      )
         self.assertFalse(processor.feature_extractor.special_attribute_present      )
         self.assertFalse(processor.tokenizer.special_attribute_present      )

         # If remote is enabled, we load from the Hub.
         lowerCamelCase__       : Optional[int]   					=						AutoProcessor.from_pretrained(
             'hf-internal-testing/test_dynamic_processor'     ,	trust_remote_code=UpperCAmelCase      )
         self.assertEqual(processor.__class__.__name__     ,	'NewProcessor'      )
         self.assertTrue(processor.special_attribute_present      )
         self.assertTrue(processor.feature_extractor.special_attribute_present      )
         self.assertTrue(processor.tokenizer.special_attribute_present      )

     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]
         if CustomConfig in TOKENIZER_MAPPING._extra_content:
             del TOKENIZER_MAPPING._extra_content[CustomConfig]
         if CustomConfig in PROCESSOR_MAPPING._extra_content:
             del PROCESSOR_MAPPING._extra_content[CustomConfig]
 def 	A_      ( self    : str      )     ->      Tuple:
     lowerCamelCase__       : str   					=						AutoProcessor.from_pretrained('hf-internal-testing/tiny-random-bert'      )
     self.assertEqual(processor.__class__.__name__     ,	'BertTokenizerFast'      )







 def 	A_      ( self    : Tuple      )     ->      List[Any]:
     lowerCamelCase__       : List[Any]   					=						AutoProcessor.from_pretrained('hf-internal-testing/tiny-random-convnext'      )
     self.assertEqual(processor.__class__.__name__     ,	'ConvNextImageProcessor'      )


@is_staging_test
class   lowerCAmelCase				(   unittest.TestCase     ):
 UpperCAmelCase__						=						["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]
 @classmethod
 def 	A_      ( cls    : Dict      )     ->      Tuple:
     lowerCamelCase__       : Union[str, Any]   					=						TOKEN
     HfFolder.save_token(UpperCAmelCase      )
 @classmethod
 def 	A_      ( cls    : int      )     ->      List[Any]:
     try:
         delete_repo(token=cls._token     ,	repo_id='test-processor'      )
     except HTTPError:
         pass

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

     try:
         delete_repo(token=cls._token     ,	repo_id='test-dynamic-processor'      )
     except HTTPError:
         pass
 def 	A_      ( self    : List[str]      )     ->      Optional[int]:
     lowerCamelCase__       : str   					=						WavaVecaProcessor.from_pretrained(UpperCAmelCase      )
     with tempfile.TemporaryDirectory() as tmp_dir:
         processor.save_pretrained(
             os.path.join(UpperCAmelCase     ,	'test-processor'      )     ,	push_to_hub=UpperCAmelCase     ,	use_auth_token=self._token      )

         lowerCamelCase__       : Tuple   					=						WavaVecaProcessor.from_pretrained(F"""{USER}/test-processor"""      )
         for k, v in processor.feature_extractor.__dict__.items():
             self.assertEqual(UpperCAmelCase     ,	getattr(new_processor.feature_extractor     ,	UpperCAmelCase      )      )
         self.assertDictEqual(new_processor.tokenizer.get_vocab()     ,	processor.tokenizer.get_vocab()      )
 def 	A_      ( self    : Union[str, Any]      )     ->      List[Any]:
     lowerCamelCase__       : List[Any]   					=						WavaVecaProcessor.from_pretrained(UpperCAmelCase      )

     with tempfile.TemporaryDirectory() as tmp_dir:
         processor.save_pretrained(
             os.path.join(UpperCAmelCase     ,	'test-processor-org'      )     ,	push_to_hub=UpperCAmelCase     ,	use_auth_token=self._token     ,	organization='valid_org'     ,	)

         lowerCamelCase__       : Any   					=						WavaVecaProcessor.from_pretrained('valid_org/test-processor-org'      )
         for k, v in processor.feature_extractor.__dict__.items():
             self.assertEqual(UpperCAmelCase     ,	getattr(new_processor.feature_extractor     ,	UpperCAmelCase      )      )
         self.assertDictEqual(new_processor.tokenizer.get_vocab()     ,	processor.tokenizer.get_vocab()      )







 def 	A_      ( self    : Union[str, Any]      )     ->      str:
     CustomFeatureExtractor.register_for_auto_class()
     CustomTokenizer.register_for_auto_class()
     CustomProcessor.register_for_auto_class()

     lowerCamelCase__       : Union[str, Any]   					=						CustomFeatureExtractor.from_pretrained(UpperCAmelCase      )

     with tempfile.TemporaryDirectory() as tmp_dir:
         lowerCamelCase__       : Optional[Any]   					=						os.path.join(UpperCAmelCase     ,	'vocab.txt'      )
         with open(UpperCAmelCase     ,	'w'     ,	encoding='utf-8'      ) as vocab_writer:
             vocab_writer.write(''.join([x + '\n' for x in self.vocab_tokens]      )      )
         lowerCamelCase__       : Any   					=						CustomTokenizer(UpperCAmelCase      )

     lowerCamelCase__       : Optional[int]   					=						CustomProcessor(UpperCAmelCase     ,	UpperCAmelCase      )

     with tempfile.TemporaryDirectory() as tmp_dir:
         create_repo(F"""{USER}/test-dynamic-processor"""     ,	token=self._token      )
         lowerCamelCase__       : str   					=						Repository(UpperCAmelCase     ,	clone_from=F"""{USER}/test-dynamic-processor"""     ,	token=self._token      )
         processor.save_pretrained(UpperCAmelCase      )

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

         # This has added the proper auto_map field to the tokenizer config
         with open(os.path.join(UpperCAmelCase     ,	'tokenizer_config.json'      )      ) as f:
             lowerCamelCase__       : List[str]   					=						json.load(UpperCAmelCase      )
         self.assertDictEqual(
             tokenizer_config['auto_map']     ,	{
                 'AutoTokenizer': ['custom_tokenization.CustomTokenizer', None],
                 'AutoProcessor': 'custom_processing.CustomProcessor',
             }     ,	)

         # The code has been copied from fixtures
         self.assertTrue(os.path.isfile(os.path.join(UpperCAmelCase     ,	'custom_feature_extraction.py'      )      )      )
         self.assertTrue(os.path.isfile(os.path.join(UpperCAmelCase     ,	'custom_tokenization.py'      )      )      )
         self.assertTrue(os.path.isfile(os.path.join(UpperCAmelCase     ,	'custom_processing.py'      )      )      )

         repo.push_to_hub()

     lowerCamelCase__       : Tuple   					=						AutoProcessor.from_pretrained(F"""{USER}/test-dynamic-processor"""     ,	trust_remote_code=UpperCAmelCase      )
     # Can't make an isinstance check because the new_processor is from the CustomProcessor class of a dynamic module
     self.assertEqual(new_processor.__class__.__name__     ,	'CustomProcessor'      )












from typing import List

from ...configuration_utils import PretrainedConfig
from ...utils import logging


_UpperCAmelCase							:			Any        =							logging.get_logger(__name__)

_UpperCAmelCase							:			Union[str, Any]        =							{
    """snap-research/efficientformer-l1-300""": (
        """https://huggingface.co/snap-research/efficientformer-l1-300/resolve/main/config.json"""
    ),
}

class   lowerCAmelCase				(   __UpperCamelCase     ):
 UpperCAmelCase__						=						"""efficientformer"""
 def __init__( self    : Dict     ,	UpperCAmelCase    : List[int] = [3, 2, 6, 4]     ,	UpperCAmelCase    : List[int] = [48, 96, 224, 448]     ,	UpperCAmelCase    : List[bool] = [True, True, True, True]     ,	UpperCAmelCase    : int = 448     ,	UpperCAmelCase    : int = 32     ,	UpperCAmelCase    : int = 4     ,	UpperCAmelCase    : int = 7     ,	UpperCAmelCase    : int = 5     ,	UpperCAmelCase    : int = 8     ,	UpperCAmelCase    : int = 4     ,	UpperCAmelCase    : float = 0.0     ,	UpperCAmelCase    : int = 16     ,	UpperCAmelCase    : int = 3     ,	UpperCAmelCase    : int = 3     ,	UpperCAmelCase    : int = 3     ,	UpperCAmelCase    : int = 2     ,	UpperCAmelCase    : int = 1     ,	UpperCAmelCase    : float = 0.0     ,	UpperCAmelCase    : int = 1     ,	UpperCAmelCase    : bool = True     ,	UpperCAmelCase    : bool = True     ,	UpperCAmelCase    : float = 1e-5     ,	UpperCAmelCase    : str = "gelu"     ,	UpperCAmelCase    : float = 0.0_2     ,	UpperCAmelCase    : float = 1e-12     ,	UpperCAmelCase    : int = 224     ,	UpperCAmelCase    : float = 1e-05     ,	**UpperCAmelCase    : Tuple     ,	)     ->      None:
     super().__init__(**UpperCAmelCase      )

     lowerCamelCase__       : Optional[Any]   					=						hidden_act
     lowerCamelCase__       : int   					=						hidden_dropout_prob
     lowerCamelCase__       : List[Any]   					=						hidden_sizes
     lowerCamelCase__       : int   					=						num_hidden_layers
     lowerCamelCase__       : int   					=						num_attention_heads
     lowerCamelCase__       : Dict   					=						initializer_range
     lowerCamelCase__       : Optional[Any]   					=						layer_norm_eps
     lowerCamelCase__       : int   					=						patch_size
     lowerCamelCase__       : List[Any]   					=						num_channels
     lowerCamelCase__       : List[str]   					=						depths
     lowerCamelCase__       : Any   					=						mlp_expansion_ratio
     lowerCamelCase__       : Any   					=						downsamples
     lowerCamelCase__       : str   					=						dim
     lowerCamelCase__       : Tuple   					=						key_dim
     lowerCamelCase__       : int   					=						attention_ratio
     lowerCamelCase__       : int   					=						resolution
     lowerCamelCase__       : Dict   					=						pool_size
     lowerCamelCase__       : List[str]   					=						downsample_patch_size
     lowerCamelCase__       : Tuple   					=						downsample_stride
     lowerCamelCase__       : int   					=						downsample_pad
     lowerCamelCase__       : Optional[int]   					=						drop_path_rate
     lowerCamelCase__       : Optional[Any]   					=						num_metaad_blocks
     lowerCamelCase__       : Any   					=						distillation
     lowerCamelCase__       : Optional[int]   					=						use_layer_scale
     lowerCamelCase__       : Union[str, Any]   					=						layer_scale_init_value
     lowerCamelCase__       : Optional[int]   					=						image_size
     lowerCamelCase__       : Optional[Any]   					=						batch_norm_eps





