


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


snake_case    :   Dict	    =					get_tests_dir('''fixtures/dummy_feature_extractor_config.json''')
snake_case    :   List[str]	    =					get_tests_dir('''fixtures/vocab.json''')
snake_case    :   Dict	    =					get_tests_dir('''fixtures''')







class 		_snake_case  (				unittest.TestCase							):
 SCREAMING_SNAKE_CASE__				=	['[UNK]', '[CLS]', '[SEP]', '[PAD]', '[MASK]', 'bla', 'blou']




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       a						:List[Any]										=	0




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       a						:Tuple										=	AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h'''   )
       self.assertIsInstance(_lowerCamelCase ,  _lowerCamelCase   )




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       with tempfile.TemporaryDirectory() as tmpdirname:
             a						:Optional[int]										=	WavaVecaConfig()
             a						:Optional[Any]										=	AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h'''   )

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

             a						:str										=	AutoProcessor.from_pretrained(_lowerCamelCase   )

       self.assertIsInstance(_lowerCamelCase ,  _lowerCamelCase   )




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       with tempfile.TemporaryDirectory() as tmpdirname:
             # copy relevant files
             copyfile(_lowerCamelCase ,  os.path.join(_lowerCamelCase ,  _lowerCamelCase   )   )
             copyfile(_lowerCamelCase ,  os.path.join(_lowerCamelCase ,  '''vocab.json'''   )   )

             a						:str										=	AutoProcessor.from_pretrained(_lowerCamelCase   )

       self.assertIsInstance(_lowerCamelCase ,  _lowerCamelCase   )




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       with tempfile.TemporaryDirectory() as tmpdirname:
             a						:Tuple										=	WavaVecaFeatureExtractor()
             a						:str										=	AutoTokenizer.from_pretrained('''facebook/wav2vec2-base-960h'''   )

             a						:List[str]										=	WavaVecaProcessor(_lowerCamelCase ,  _lowerCamelCase   )

             # save in new folder
             processor.save_pretrained(_lowerCamelCase   )

             # drop `processor_class` in tokenizer
             with open(os.path.join(_lowerCamelCase ,  _lowerCamelCase   ) ,  '''r'''   ) as f:
                   a						:List[str]										=	json.load(_lowerCamelCase   )
                   config_dict.pop('''processor_class'''   )

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

             a						:Optional[Any]										=	AutoProcessor.from_pretrained(_lowerCamelCase   )

       self.assertIsInstance(_lowerCamelCase ,  _lowerCamelCase   )




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       with tempfile.TemporaryDirectory() as tmpdirname:
             a						:int										=	WavaVecaFeatureExtractor()
             a						:Tuple										=	AutoTokenizer.from_pretrained('''facebook/wav2vec2-base-960h'''   )

             a						:Union[str, Any]										=	WavaVecaProcessor(_lowerCamelCase ,  _lowerCamelCase   )

             # save in new folder
             processor.save_pretrained(_lowerCamelCase   )

             # drop `processor_class` in feature extractor
             with open(os.path.join(_lowerCamelCase ,  _lowerCamelCase   ) ,  '''r'''   ) as f:
                   a						:int										=	json.load(_lowerCamelCase   )
                   config_dict.pop('''processor_class'''   )

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

             a						:List[str]										=	AutoProcessor.from_pretrained(_lowerCamelCase   )

       self.assertIsInstance(_lowerCamelCase ,  _lowerCamelCase   )




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       with tempfile.TemporaryDirectory() as tmpdirname:
             a						:Any										=	WavaVecaConfig(processor_class='''Wav2Vec2Processor'''   )
             model_config.save_pretrained(_lowerCamelCase   )
             # copy relevant files
             copyfile(_lowerCamelCase ,  os.path.join(_lowerCamelCase ,  '''vocab.json'''   )   )
             # create emtpy sample processor
             with open(os.path.join(_lowerCamelCase ,  _lowerCamelCase   ) ,  '''w'''   ) as f:
                   f.write('''{}'''   )

             a						:int										=	AutoProcessor.from_pretrained(_lowerCamelCase   )

       self.assertIsInstance(_lowerCamelCase ,  _lowerCamelCase   )




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

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

       a						:Union[str, Any]										=	processor.feature_extractor
       self.assertTrue(feature_extractor.special_attribute_present   )
       self.assertEqual(feature_extractor.__class__.__name__ ,  '''NewFeatureExtractor'''   )

       a						:Union[str, Any]										=	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
             a						:Any										=	AutoProcessor.from_pretrained(
                 '''hf-internal-testing/test_dynamic_processor''' ,  trust_remote_code=_lowerCamelCase ,  use_fast=_lowerCamelCase   )
             a						:List[str]										=	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 	SCREAMING_SNAKE_CASE__    (		self   ):
       try:
             AutoConfig.register('''custom''' ,  _lowerCamelCase   )
             AutoFeatureExtractor.register(_lowerCamelCase ,  _lowerCamelCase   )
             AutoTokenizer.register(_lowerCamelCase ,  slow_tokenizer_class=_lowerCamelCase   )
             AutoProcessor.register(_lowerCamelCase ,  _lowerCamelCase   )
             # Trying to register something existing in the Transformers library will raise an error
             with self.assertRaises(_lowerCamelCase   ):
                   AutoProcessor.register(_lowerCamelCase ,  _lowerCamelCase   )

             # Now that the config is registered, it can be used as any other config with the auto-API
             a						:Union[str, Any]										=	CustomFeatureExtractor.from_pretrained(_lowerCamelCase   )

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

             a						:List[Any]										=	CustomProcessor(_lowerCamelCase ,  _lowerCamelCase   )

             with tempfile.TemporaryDirectory() as tmp_dir:
                   processor.save_pretrained(_lowerCamelCase   )
                   a						:Dict										=	AutoProcessor.from_pretrained(_lowerCamelCase   )
                   self.assertIsInstance(_lowerCamelCase ,  _lowerCamelCase   )

       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 	SCREAMING_SNAKE_CASE__    (		self   ):







       class 		_snake_case  (				_snake_case							):
        SCREAMING_SNAKE_CASE__				=	False







       class 		_snake_case  (				_snake_case							):
        SCREAMING_SNAKE_CASE__				=	False







       class 		_snake_case  (				_snake_case							):
        SCREAMING_SNAKE_CASE__				=	'AutoFeatureExtractor'
        SCREAMING_SNAKE_CASE__				=	'AutoTokenizer'
        SCREAMING_SNAKE_CASE__				=	False

       try:
             AutoConfig.register('''custom''' ,  _lowerCamelCase   )
             AutoFeatureExtractor.register(_lowerCamelCase ,  _lowerCamelCase   )
             AutoTokenizer.register(_lowerCamelCase ,  slow_tokenizer_class=_lowerCamelCase   )
             AutoProcessor.register(_lowerCamelCase ,  _lowerCamelCase   )
             # If remote code is not set, the default is to use local classes.
             a						:Union[str, Any]										=	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.
             a						:str										=	AutoProcessor.from_pretrained(
                 '''hf-internal-testing/test_dynamic_processor''' ,  trust_remote_code=_lowerCamelCase   )
             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.
             a						:str										=	AutoProcessor.from_pretrained(
                 '''hf-internal-testing/test_dynamic_processor''' ,  trust_remote_code=_lowerCamelCase   )
             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 	SCREAMING_SNAKE_CASE__    (		self   ):
       a						:List[str]										=	AutoProcessor.from_pretrained('''hf-internal-testing/tiny-random-bert'''   )
       self.assertEqual(processor.__class__.__name__ ,  '''BertTokenizerFast'''   )




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       a						:Any										=	AutoProcessor.from_pretrained('''hf-internal-testing/tiny-random-convnext'''   )
       self.assertEqual(processor.__class__.__name__ ,  '''ConvNextImageProcessor'''   )







@is_staging_test
class 		_snake_case  (				unittest.TestCase							):
 SCREAMING_SNAKE_CASE__				=	['[UNK]', '[CLS]', '[SEP]', '[PAD]', '[MASK]', 'bla', 'blou']




 @classmethod
 def 	SCREAMING_SNAKE_CASE__    (		cls   ):
       a						:List[str]										=	TOKEN
       HfFolder.save_token(_lowerCamelCase   )




 @classmethod
 def 	SCREAMING_SNAKE_CASE__    (		cls   ):
       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 	SCREAMING_SNAKE_CASE__    (		self   ):
       a						:int										=	WavaVecaProcessor.from_pretrained(_lowerCamelCase   )
       with tempfile.TemporaryDirectory() as tmp_dir:
             processor.save_pretrained(
                 os.path.join(_lowerCamelCase ,  '''test-processor'''   ) ,  push_to_hub=_lowerCamelCase ,  use_auth_token=self._token   )

             a						:Dict										=	WavaVecaProcessor.from_pretrained(F'''{USER}/test-processor'''   )
             for k, v in processor.feature_extractor.__dict__.items():
                   self.assertEqual(_lowerCamelCase ,  getattr(new_processor.feature_extractor ,  _lowerCamelCase   )   )
             self.assertDictEqual(new_processor.tokenizer.get_vocab() ,  processor.tokenizer.get_vocab()   )




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       a						:str										=	WavaVecaProcessor.from_pretrained(_lowerCamelCase   )

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

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




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       CustomFeatureExtractor.register_for_auto_class()
       CustomTokenizer.register_for_auto_class()
       CustomProcessor.register_for_auto_class()

       a						:Optional[int]										=	CustomFeatureExtractor.from_pretrained(_lowerCamelCase   )

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

       a						:Optional[int]										=	CustomProcessor(_lowerCamelCase ,  _lowerCamelCase   )

       with tempfile.TemporaryDirectory() as tmp_dir:
             create_repo(F'''{USER}/test-dynamic-processor''' ,  token=self._token   )
             a						:Optional[Any]										=	Repository(_lowerCamelCase ,  clone_from=F'''{USER}/test-dynamic-processor''' ,  token=self._token   )
             processor.save_pretrained(_lowerCamelCase   )

             # 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(_lowerCamelCase ,  '''tokenizer_config.json'''   )   ) as f:
                   a						:str										=	json.load(_lowerCamelCase   )
             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(_lowerCamelCase ,  '''custom_feature_extraction.py'''   )   )   )
             self.assertTrue(os.path.isfile(os.path.join(_lowerCamelCase ,  '''custom_tokenization.py'''   )   )   )
             self.assertTrue(os.path.isfile(os.path.join(_lowerCamelCase ,  '''custom_processing.py'''   )   )   )

             repo.push_to_hub()

       a						:int										=	AutoProcessor.from_pretrained(F'''{USER}/test-dynamic-processor''' ,  trust_remote_code=_lowerCamelCase   )
       # 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 ...configuration_utils import PretrainedConfig







class 		_snake_case  (				_snake_case							):
 SCREAMING_SNAKE_CASE__				=	'bert-generation'




 def __init__(		self ,  _lowerCamelCase=5_0358 ,  _lowerCamelCase=1024 ,  _lowerCamelCase=24 ,  _lowerCamelCase=16 ,  _lowerCamelCase=4096 ,  _lowerCamelCase="gelu" ,  _lowerCamelCase=0.1 ,  _lowerCamelCase=0.1 ,  _lowerCamelCase=512 ,  _lowerCamelCase=0.02 ,  _lowerCamelCase=1e-12 ,  _lowerCamelCase=0 ,  _lowerCamelCase=2 ,  _lowerCamelCase=1 ,  _lowerCamelCase="absolute" ,  _lowerCamelCase=True ,  **_lowerCamelCase ,  ):
       super().__init__(pad_token_id=_lowerCamelCase ,  bos_token_id=_lowerCamelCase ,  eos_token_id=_lowerCamelCase ,  **_lowerCamelCase   )

       a						:Optional[int]										=	vocab_size
       a						:Tuple										=	hidden_size
       a						:Any										=	num_hidden_layers
       a						:Any										=	num_attention_heads
       a						:List[Any]										=	hidden_act
       a						:Tuple										=	intermediate_size
       a						:Any										=	hidden_dropout_prob
       a						:int										=	attention_probs_dropout_prob
       a						:Dict										=	max_position_embeddings
       a						:int										=	initializer_range
       a						:Union[str, Any]										=	layer_norm_eps
       a						:str										=	position_embedding_type
       a						:int										=	use_cache





