


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__      		= get_tests_dir('fixtures/dummy_feature_extractor_config.json')
lowerCAmelCase__      		= get_tests_dir('fixtures/vocab.json')
lowerCAmelCase__      		= get_tests_dir('fixtures')







class      lowerCAmelCase__       (				unittest.TestCase):




     '''simple docstring'''
     __SCREAMING_SNAKE_CASE  			= ["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "bla", "blou"]


     def  _lowerCamelCase (						self)			->				Any:
            _A      :       str				    = 0


     def  _lowerCamelCase (						self)			->				Optional[Any]:
            _A      :       Dict				    = AutoProcessor.from_pretrained("facebook/wav2vec2-base-960h")
            self.assertIsInstance(__lowerCamelCase ,						__lowerCamelCase)


     def  _lowerCamelCase (						self)			->				int:
            with tempfile.TemporaryDirectory() as tmpdirname:
                   _A      :       Union[str, Any]				    = WavaVecaConfig()
                   _A      :       List[Any]				    = AutoProcessor.from_pretrained("facebook/wav2vec2-base-960h")

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

                   _A      :       Dict				    = AutoProcessor.from_pretrained(__lowerCamelCase)

            self.assertIsInstance(__lowerCamelCase ,						__lowerCamelCase)


     def  _lowerCamelCase (						self)			->				Dict:
            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  _lowerCamelCase (						self)			->				int:
            with tempfile.TemporaryDirectory() as tmpdirname:
                   _A      :       Dict				    = WavaVecaFeatureExtractor()
                   _A      :       List[Any]				    = 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 tokenizer
                   with open(os.path.join(__lowerCamelCase ,						__lowerCamelCase) ,						"r") as f:
                          _A      :       Dict				    = json.load(__lowerCamelCase)
                          config_dict.pop("processor_class")

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

                   _A      :       int				    = AutoProcessor.from_pretrained(__lowerCamelCase)

            self.assertIsInstance(__lowerCamelCase ,						__lowerCamelCase)


     def  _lowerCamelCase (						self)			->				Dict:
            with tempfile.TemporaryDirectory() as tmpdirname:
                   _A      :       Tuple				    = WavaVecaFeatureExtractor()
                   _A      :       List[str]				    = 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      :       Optional[Any]				    = json.load(__lowerCamelCase)
                          config_dict.pop("processor_class")

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

                   _A      :       str				    = AutoProcessor.from_pretrained(__lowerCamelCase)

            self.assertIsInstance(__lowerCamelCase ,						__lowerCamelCase)


     def  _lowerCamelCase (						self)			->				List[str]:
            with tempfile.TemporaryDirectory() as tmpdirname:
                   _A      :       Tuple				    = 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      :       Optional[int]				    = AutoProcessor.from_pretrained(__lowerCamelCase)

            self.assertIsInstance(__lowerCamelCase ,						__lowerCamelCase)


     def  _lowerCamelCase (						self)			->				Union[str, Any]:
            # If remote code is not set, we will time out when asking whether to load the model.
            with self.assertRaises(__lowerCamelCase):
                   _A      :       Any				    = 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      :       Optional[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      :       Any				    = processor.feature_extractor
            self.assertTrue(feature_extractor.special_attribute_present)
            self.assertEqual(feature_extractor.__class__.__name__ ,						"NewFeatureExtractor")

            _A      :       Dict				    = 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      :       str				    = AutoProcessor.from_pretrained(
                       "hf-internal-testing/test_dynamic_processor" ,						trust_remote_code=__lowerCamelCase ,						use_fast=__lowerCamelCase)
                   _A      :       List[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  _lowerCamelCase (						self)			->				Dict:
            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      :       Any				    = CustomFeatureExtractor.from_pretrained(__lowerCamelCase)

                   with tempfile.TemporaryDirectory() as tmp_dir:
                          _A      :       Optional[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      :       List[str]				    = CustomTokenizer(__lowerCamelCase)

                   _A      :       Tuple				    = CustomProcessor(__lowerCamelCase ,						__lowerCamelCase)

                   with tempfile.TemporaryDirectory() as tmp_dir:
                          processor.save_pretrained(__lowerCamelCase)
                          _A      :       Tuple				    = 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  _lowerCamelCase (						self)			->				Any:







            class      lowerCAmelCase__       (				a):




                 '''simple docstring'''
                 __SCREAMING_SNAKE_CASE  			= False







            class      lowerCAmelCase__       (				a):




                 '''simple docstring'''
                 __SCREAMING_SNAKE_CASE  			= False







            class      lowerCAmelCase__       (				a):




                 '''simple docstring'''
                 __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      :       Dict				    = 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      :       Any				    = 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      :       Union[str, Any]				    = 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  _lowerCamelCase (						self)			->				Dict:
            _A      :       Tuple				    = AutoProcessor.from_pretrained("hf-internal-testing/tiny-random-bert")
            self.assertEqual(processor.__class__.__name__ ,						"BertTokenizerFast")


     def  _lowerCamelCase (						self)			->				List[Any]:
            _A      :       Optional[Any]				    = AutoProcessor.from_pretrained("hf-internal-testing/tiny-random-convnext")
            self.assertEqual(processor.__class__.__name__ ,						"ConvNextImageProcessor")







@is_staging_test
class      lowerCAmelCase__       (				unittest.TestCase):




     '''simple docstring'''
     __SCREAMING_SNAKE_CASE  			= ["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "bla", "blou"]


     @classmethod
     def  _lowerCamelCase (						cls)			->				Any:
            _A      :       Any				    = TOKEN
            HfFolder.save_token(__lowerCamelCase)


     @classmethod
     def  _lowerCamelCase (						cls)			->				List[str]:
            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  _lowerCamelCase (						self)			->				Tuple:
            _A      :       Optional[Any]				    = 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      :       str				    = 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  _lowerCamelCase (						self)			->				str:
            _A      :       Optional[Any]				    = 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      :       List[str]				    = 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  _lowerCamelCase (						self)			->				int:
            CustomFeatureExtractor.register_for_auto_class()
            CustomTokenizer.register_for_auto_class()
            CustomProcessor.register_for_auto_class()

            _A      :       Any				    = CustomFeatureExtractor.from_pretrained(__lowerCamelCase)

            with tempfile.TemporaryDirectory() as tmp_dir:
                   _A      :       str				    = 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      :       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      :       Any				    = 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      :       List[Any]				    = 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")




import warnings

from ...utils import logging
from .image_processing_beit import BeitImageProcessor


lowerCAmelCase__      		= logging.get_logger(__name__)







class      lowerCAmelCase__       (				a):




     '''simple docstring'''


     def __init__(						self ,						*__lowerCamelCase ,						**__lowerCamelCase)			->				None:
            warnings.warn(
                "The class BeitFeatureExtractor is deprecated and will be removed in version 5 of Transformers. Please"
                " use BeitImageProcessor instead." ,						__lowerCamelCase ,						)
            super().__init__(*__lowerCamelCase ,						**__lowerCamelCase)
