




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


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


class 							UpperCamelCase_       (					unittest.TestCase	):

    '''simple docstring'''






    UpperCAmelCase__				    = ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']


    def 			SCREAMING_SNAKE_CASE				(     self :      Optional[Any])		->List[str]:





           '''simple docstring'''

           A__ 					=       0


    def 			SCREAMING_SNAKE_CASE				(     self :      Optional[int])		->int:





           '''simple docstring'''

           A__ 					=       AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h''')
           self.assertIsInstance(UpperCAmelCase__   ,					UpperCAmelCase__)


    def 			SCREAMING_SNAKE_CASE				(     self :      Optional[int])		->Any:





           '''simple docstring'''

           with tempfile.TemporaryDirectory() as tmpdirname:
                  A__ 					=       WavaVecaConfig()
                  A__ 					=       AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h''')

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

                  A__ 					=       AutoProcessor.from_pretrained(UpperCAmelCase__)

           self.assertIsInstance(UpperCAmelCase__   ,					UpperCAmelCase__)


    def 			SCREAMING_SNAKE_CASE				(     self :      str)		->Any:





           '''simple docstring'''

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

                  A__ 					=       AutoProcessor.from_pretrained(UpperCAmelCase__)

           self.assertIsInstance(UpperCAmelCase__   ,					UpperCAmelCase__)


    def 			SCREAMING_SNAKE_CASE				(     self :      List[str])		->Any:





           '''simple docstring'''

           with tempfile.TemporaryDirectory() as tmpdirname:
                  A__ 					=       WavaVecaFeatureExtractor()
                  A__ 					=       AutoTokenizer.from_pretrained('''facebook/wav2vec2-base-960h''')

                  A__ 					=       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:
                         A__ 					=       json.load(UpperCAmelCase__)
                         config_dict.pop('''processor_class''')

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

                  A__ 					=       AutoProcessor.from_pretrained(UpperCAmelCase__)

           self.assertIsInstance(UpperCAmelCase__   ,					UpperCAmelCase__)


    def 			SCREAMING_SNAKE_CASE				(     self :      str)		->Optional[int]:





           '''simple docstring'''

           with tempfile.TemporaryDirectory() as tmpdirname:
                  A__ 					=       WavaVecaFeatureExtractor()
                  A__ 					=       AutoTokenizer.from_pretrained('''facebook/wav2vec2-base-960h''')

                  A__ 					=       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:
                         A__ 					=       json.load(UpperCAmelCase__)
                         config_dict.pop('''processor_class''')

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

                  A__ 					=       AutoProcessor.from_pretrained(UpperCAmelCase__)

           self.assertIsInstance(UpperCAmelCase__   ,					UpperCAmelCase__)


    def 			SCREAMING_SNAKE_CASE				(     self :      str)		->Optional[int]:





           '''simple docstring'''

           with tempfile.TemporaryDirectory() as tmpdirname:
                  A__ 					=       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('''{}''')

                  A__ 					=       AutoProcessor.from_pretrained(UpperCAmelCase__)

           self.assertIsInstance(UpperCAmelCase__   ,					UpperCAmelCase__)


    def 			SCREAMING_SNAKE_CASE				(     self :      Optional[Any])		->Union[str, Any]:





           '''simple docstring'''

           with self.assertRaises(UpperCAmelCase__):
                  A__ 					=       AutoProcessor.from_pretrained('''hf-internal-testing/test_dynamic_processor''')
           # If remote code is disabled, we can't load this config.
           with self.assertRaises(UpperCAmelCase__):
                  A__ 					=       AutoProcessor.from_pretrained(
                      '''hf-internal-testing/test_dynamic_processor'''   ,					trust_remote_code=UpperCAmelCase__)

           A__ 					=       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''')

           A__ 					=       processor.feature_extractor
           self.assertTrue(feature_extractor.special_attribute_present)
           self.assertEqual(feature_extractor.__class__.__name__   ,					'''NewFeatureExtractor''')

           A__ 					=       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__ 					=       AutoProcessor.from_pretrained(
                      '''hf-internal-testing/test_dynamic_processor'''   ,					trust_remote_code=UpperCAmelCase__   ,					use_fast=UpperCAmelCase__)
                  A__ 					=       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 :      Optional[Any])		->Union[str, Any]:





           '''simple docstring'''

           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
                  A__ 					=       CustomFeatureExtractor.from_pretrained(UpperCAmelCase__)

                  with tempfile.TemporaryDirectory() as tmp_dir:
                         A__ 					=       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]))
                         A__ 					=       CustomTokenizer(UpperCAmelCase__)

                  A__ 					=       CustomProcessor(UpperCAmelCase__   ,					UpperCAmelCase__)

                  with tempfile.TemporaryDirectory() as tmp_dir:
                         processor.save_pretrained(UpperCAmelCase__)
                         A__ 					=       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 			SCREAMING_SNAKE_CASE				(     self :      Union[str, Any])		->Tuple:





           '''simple docstring'''

           class 							UpperCamelCase_       (					UpperCAmelCase__	):

               '''simple docstring'''






               UpperCAmelCase__				    = False


           class 							UpperCamelCase_       (					UpperCAmelCase__	):

               '''simple docstring'''






               UpperCAmelCase__				    = False


           class 							UpperCamelCase_       (					UpperCAmelCase__	):

               '''simple docstring'''






               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.
                  A__ 					=       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__ 					=       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.
                  A__ 					=       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 			SCREAMING_SNAKE_CASE				(     self :      Optional[int])		->Dict:





           '''simple docstring'''

           A__ 					=       AutoProcessor.from_pretrained('''hf-internal-testing/tiny-random-bert''')
           self.assertEqual(processor.__class__.__name__   ,					'''BertTokenizerFast''')





    def 			SCREAMING_SNAKE_CASE				(     self :      Any)		->Optional[Any]:





           '''simple docstring'''

           A__ 					=       AutoProcessor.from_pretrained('''hf-internal-testing/tiny-random-convnext''')
           self.assertEqual(processor.__class__.__name__   ,					'''ConvNextImageProcessor''')


@is_staging_test
class 							UpperCamelCase_       (					unittest.TestCase	):

    '''simple docstring'''






    UpperCAmelCase__				    = ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']


    @classmethod
    def 			SCREAMING_SNAKE_CASE				(     cls :      Any)		->List[Any]:





           '''simple docstring'''

           A__ 					=       TOKEN
           HfFolder.save_token(UpperCAmelCase__)


    @classmethod
    def 			SCREAMING_SNAKE_CASE				(     cls :      Union[str, Any])		->Any:





           '''simple docstring'''

           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 :      List[str])		->Any:





           '''simple docstring'''

           A__ 					=       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)

                  A__ 					=       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 			SCREAMING_SNAKE_CASE				(     self :      List[Any])		->str:





           '''simple docstring'''

           A__ 					=       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'''   ,					)

                  A__ 					=       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 			SCREAMING_SNAKE_CASE				(     self :      Union[str, Any])		->Union[str, Any]:





           '''simple docstring'''

           CustomFeatureExtractor.register_for_auto_class()
           CustomTokenizer.register_for_auto_class()
           CustomProcessor.register_for_auto_class()

           A__ 					=       CustomFeatureExtractor.from_pretrained(UpperCAmelCase__)

           with tempfile.TemporaryDirectory() as tmp_dir:
                  A__ 					=       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]))
                  A__ 					=       CustomTokenizer(UpperCAmelCase__)

           A__ 					=       CustomProcessor(UpperCAmelCase__   ,					UpperCAmelCase__)

           with tempfile.TemporaryDirectory() as tmp_dir:
                  create_repo(f"""{USER}/test-dynamic-processor"""   ,					token=self._token)
                  A__ 					=       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:
                         A__ 					=       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()

           A__ 					=       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''')




'''simple docstring'''
from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_torch_available,
)


lowerCAmelCase					:    Optional[Any]           = {
    'configuration_falcon': ['FALCON_PRETRAINED_CONFIG_ARCHIVE_MAP', 'FalconConfig'],
}

try:
						if not is_torch_available():
												raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
						pass
else:
						lowerCAmelCase					:    str           = [
						    'FALCON_PRETRAINED_MODEL_ARCHIVE_LIST',
						    'FalconForCausalLM',
						    'FalconModel',
						    'FalconPreTrainedModel',
						    'FalconForSequenceClassification',
						    'FalconForTokenClassification',
						    'FalconForQuestionAnswering',
						]


if TYPE_CHECKING:
						from .configuration_falcon import FALCON_PRETRAINED_CONFIG_ARCHIVE_MAP, FalconConfig

						try:
												if not is_torch_available():
																		raise OptionalDependencyNotAvailable()
						except OptionalDependencyNotAvailable:
												pass
						else:
												from .modeling_falcon import (
												    FALCON_PRETRAINED_MODEL_ARCHIVE_LIST,
												    FalconForCausalLM,
												    FalconForQuestionAnswering,
												    FalconForSequenceClassification,
												    FalconForTokenClassification,
												    FalconModel,
												    FalconPreTrainedModel,
												)


else:
						import sys

						lowerCAmelCase					:    Tuple           = _LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)




