
"""simple docstring"""





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


_a		       =							get_tests_dir('fixtures/dummy_feature_extractor_config.json')
_a		       =							get_tests_dir('fixtures/vocab.json')
_a		       =							get_tests_dir('fixtures')


class A_	(unittest.TestCase      ):




  '''simple docstring'''
  SCREAMING_SNAKE_CASE__			:						int       =		["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]







  def        UpperCamelCase__					(		self  ):

     """simple docstring"""
     UpperCAmelCase_					:					Optional[int]			     =    0







  def        UpperCamelCase__					(		self  ):

     """simple docstring"""
     UpperCAmelCase_					:					List[str]			     =    AutoProcessor.from_pretrained("facebook/wav2vec2-base-960h"  )
     self.assertIsInstance(lowercase_				,   lowercase_  )







  def        UpperCamelCase__					(		self  ):

     """simple docstring"""
     with tempfile.TemporaryDirectory() as tmpdirname:
        UpperCAmelCase_					:					Dict			     =    WavaVecaConfig()
        UpperCAmelCase_					:					Union[str, Any]			     =    AutoProcessor.from_pretrained("facebook/wav2vec2-base-960h"  )

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

        UpperCAmelCase_					:					Optional[Any]			     =    AutoProcessor.from_pretrained(lowercase_  )

     self.assertIsInstance(lowercase_				,   lowercase_  )







  def        UpperCamelCase__					(		self  ):

     """simple docstring"""
     with tempfile.TemporaryDirectory() as tmpdirname:
        # copy relevant files
        copyfile(lowercase_				,   os.path.join(lowercase_				,   lowercase_  )  )
        copyfile(lowercase_				,   os.path.join(lowercase_				,   "vocab.json"  )  )

        UpperCAmelCase_					:					Dict			     =    AutoProcessor.from_pretrained(lowercase_  )

     self.assertIsInstance(lowercase_				,   lowercase_  )







  def        UpperCamelCase__					(		self  ):

     """simple docstring"""
     with tempfile.TemporaryDirectory() as tmpdirname:
        UpperCAmelCase_					:					str			     =    WavaVecaFeatureExtractor()
        UpperCAmelCase_					:					int			     =    AutoTokenizer.from_pretrained("facebook/wav2vec2-base-960h"  )

        UpperCAmelCase_					:					List[str]			     =    WavaVecaProcessor(lowercase_				,   lowercase_  )

        # save in new folder
        processor.save_pretrained(lowercase_  )

        # drop `processor_class` in tokenizer
        with open(os.path.join(lowercase_				,   lowercase_  )				,   "r"  ) as f:
           UpperCAmelCase_					:					Optional[int]			     =    json.load(lowercase_  )
           config_dict.pop("processor_class"  )

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

        UpperCAmelCase_					:					List[Any]			     =    AutoProcessor.from_pretrained(lowercase_  )

     self.assertIsInstance(lowercase_				,   lowercase_  )







  def        UpperCamelCase__					(		self  ):

     """simple docstring"""
     with tempfile.TemporaryDirectory() as tmpdirname:
        UpperCAmelCase_					:					str			     =    WavaVecaFeatureExtractor()
        UpperCAmelCase_					:					Dict			     =    AutoTokenizer.from_pretrained("facebook/wav2vec2-base-960h"  )

        UpperCAmelCase_					:					List[Any]			     =    WavaVecaProcessor(lowercase_				,   lowercase_  )

        # save in new folder
        processor.save_pretrained(lowercase_  )

        # drop `processor_class` in feature extractor
        with open(os.path.join(lowercase_				,   lowercase_  )				,   "r"  ) as f:
           UpperCAmelCase_					:					Dict			     =    json.load(lowercase_  )
           config_dict.pop("processor_class"  )

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

        UpperCAmelCase_					:					List[Any]			     =    AutoProcessor.from_pretrained(lowercase_  )

     self.assertIsInstance(lowercase_				,   lowercase_  )







  def        UpperCamelCase__					(		self  ):

     """simple docstring"""
     with tempfile.TemporaryDirectory() as tmpdirname:
        UpperCAmelCase_					:					Union[str, Any]			     =    WavaVecaConfig(processor_class="Wav2Vec2Processor"  )
        model_config.save_pretrained(lowercase_  )
        # copy relevant files
        copyfile(lowercase_				,   os.path.join(lowercase_				,   "vocab.json"  )  )
        # create emtpy sample processor
        with open(os.path.join(lowercase_				,   lowercase_  )				,   "w"  ) as f:
           f.write("{}"  )

        UpperCAmelCase_					:					List[str]			     =    AutoProcessor.from_pretrained(lowercase_  )

     self.assertIsInstance(lowercase_				,   lowercase_  )







  def        UpperCamelCase__					(		self  ):

     """simple docstring"""
     # If remote code is not set, we will time out when asking whether to load the model.
     with self.assertRaises(lowercase_  ):
        UpperCAmelCase_					:					List[str]			     =    AutoProcessor.from_pretrained("hf-internal-testing/test_dynamic_processor"  )
     # If remote code is disabled, we can't load this config.
     with self.assertRaises(lowercase_  ):
        UpperCAmelCase_					:					Dict			     =    AutoProcessor.from_pretrained(
            "hf-internal-testing/test_dynamic_processor"				,   trust_remote_code=lowercase_  )

     UpperCAmelCase_					:					Optional[int]			     =    AutoProcessor.from_pretrained("hf-internal-testing/test_dynamic_processor"				,   trust_remote_code=lowercase_  )
     self.assertTrue(processor.special_attribute_present  )
     self.assertEqual(processor.__class__.__name__				,   "NewProcessor"  )

     UpperCAmelCase_					:					List[Any]			     =    processor.feature_extractor
     self.assertTrue(feature_extractor.special_attribute_present  )
     self.assertEqual(feature_extractor.__class__.__name__				,   "NewFeatureExtractor"  )

     UpperCAmelCase_					:					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
        UpperCAmelCase_					:					Optional[Any]			     =    AutoProcessor.from_pretrained(
            "hf-internal-testing/test_dynamic_processor"				,   trust_remote_code=lowercase_				,   use_fast=lowercase_  )
        UpperCAmelCase_					:					Dict			     =    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        UpperCamelCase__					(		self  ):

     """simple docstring"""
     try:
        AutoConfig.register("custom"				,   lowercase_  )
        AutoFeatureExtractor.register(lowercase_				,   lowercase_  )
        AutoTokenizer.register(lowercase_				,   slow_tokenizer_class=lowercase_  )
        AutoProcessor.register(lowercase_				,   lowercase_  )
        # Trying to register something existing in the Transformers library will raise an error
        with self.assertRaises(lowercase_  ):
           AutoProcessor.register(lowercase_				,   lowercase_  )

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

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

        UpperCAmelCase_					:					Any			     =    CustomProcessor(lowercase_				,   lowercase_  )

        with tempfile.TemporaryDirectory() as tmp_dir:
           processor.save_pretrained(lowercase_  )
           UpperCAmelCase_					:					Union[str, Any]			     =    AutoProcessor.from_pretrained(lowercase_  )
           self.assertIsInstance(lowercase_				,   lowercase_  )

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

     """simple docstring"""


     class A_	(lowercase__      ):




       '''simple docstring'''
       SCREAMING_SNAKE_CASE__			:						Dict       =		False




     class A_	(lowercase__      ):




       '''simple docstring'''
       SCREAMING_SNAKE_CASE__			:						Optional[Any]       =		False




     class A_	(lowercase__      ):




       '''simple docstring'''
       SCREAMING_SNAKE_CASE__			:						List[str]       =		"""AutoFeatureExtractor"""
       SCREAMING_SNAKE_CASE__			:						List[str]       =		"""AutoTokenizer"""
       SCREAMING_SNAKE_CASE__			:						Any       =		False

     try:
        AutoConfig.register("custom"				,   lowercase_  )
        AutoFeatureExtractor.register(lowercase_				,   lowercase_  )
        AutoTokenizer.register(lowercase_				,   slow_tokenizer_class=lowercase_  )
        AutoProcessor.register(lowercase_				,   lowercase_  )
        # If remote code is not set, the default is to use local classes.
        UpperCAmelCase_					:					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.
        UpperCAmelCase_					:					Any			     =    AutoProcessor.from_pretrained(
            "hf-internal-testing/test_dynamic_processor"				,   trust_remote_code=lowercase_  )
        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.
        UpperCAmelCase_					:					Optional[Any]			     =    AutoProcessor.from_pretrained(
            "hf-internal-testing/test_dynamic_processor"				,   trust_remote_code=lowercase_  )
        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        UpperCamelCase__					(		self  ):

     """simple docstring"""
     UpperCAmelCase_					:					int			     =    AutoProcessor.from_pretrained("hf-internal-testing/tiny-random-bert"  )
     self.assertEqual(processor.__class__.__name__				,   "BertTokenizerFast"  )







  def        UpperCamelCase__					(		self  ):

     """simple docstring"""
     UpperCAmelCase_					:					Dict			     =    AutoProcessor.from_pretrained("hf-internal-testing/tiny-random-convnext"  )
     self.assertEqual(processor.__class__.__name__				,   "ConvNextImageProcessor"  )




@is_staging_test
class A_	(unittest.TestCase      ):




  '''simple docstring'''
  SCREAMING_SNAKE_CASE__			:						Optional[int]       =		["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]







  @classmethod
  def        UpperCamelCase__					(		cls  ):

     """simple docstring"""
     UpperCAmelCase_					:					Union[str, Any]			     =    TOKEN
     HfFolder.save_token(lowercase_  )







  @classmethod
  def        UpperCamelCase__					(		cls  ):

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

     """simple docstring"""
     UpperCAmelCase_					:					Dict			     =    WavaVecaProcessor.from_pretrained(lowercase_  )
     with tempfile.TemporaryDirectory() as tmp_dir:
        processor.save_pretrained(
            os.path.join(lowercase_				,   "test-processor"  )				,   push_to_hub=lowercase_				,   use_auth_token=self._token  )

        UpperCAmelCase_					:					Optional[Any]			     =    WavaVecaProcessor.from_pretrained(F"""{USER}/test-processor"""  )
        for k, v in processor.feature_extractor.__dict__.items():
           self.assertEqual(lowercase_				,   getattr(new_processor.feature_extractor				,   lowercase_  )  )
        self.assertDictEqual(new_processor.tokenizer.get_vocab()				,   processor.tokenizer.get_vocab()  )







  def        UpperCamelCase__					(		self  ):

     """simple docstring"""
     UpperCAmelCase_					:					Tuple			     =    WavaVecaProcessor.from_pretrained(lowercase_  )

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

        UpperCAmelCase_					:					Union[str, Any]			     =    WavaVecaProcessor.from_pretrained("valid_org/test-processor-org"  )
        for k, v in processor.feature_extractor.__dict__.items():
           self.assertEqual(lowercase_				,   getattr(new_processor.feature_extractor				,   lowercase_  )  )
        self.assertDictEqual(new_processor.tokenizer.get_vocab()				,   processor.tokenizer.get_vocab()  )







  def        UpperCamelCase__					(		self  ):

     """simple docstring"""
     CustomFeatureExtractor.register_for_auto_class()
     CustomTokenizer.register_for_auto_class()
     CustomProcessor.register_for_auto_class()

     UpperCAmelCase_					:					Union[str, Any]			     =    CustomFeatureExtractor.from_pretrained(lowercase_  )

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

     UpperCAmelCase_					:					Any			     =    CustomProcessor(lowercase_				,   lowercase_  )

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

        # 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(lowercase_				,   "tokenizer_config.json"  )  ) as f:
           UpperCAmelCase_					:					Dict			     =    json.load(lowercase_  )
        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(lowercase_				,   "custom_feature_extraction.py"  )  )  )
        self.assertTrue(os.path.isfile(os.path.join(lowercase_				,   "custom_tokenization.py"  )  )  )
        self.assertTrue(os.path.isfile(os.path.join(lowercase_				,   "custom_processing.py"  )  )  )

        repo.push_to_hub()

     UpperCAmelCase_					:					List[Any]			     =    AutoProcessor.from_pretrained(F"""{USER}/test-dynamic-processor"""				,   trust_remote_code=lowercase_  )
     # 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 TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tf_available, is_torch_available


UpperCAmelCase						:   Optional[int]			  =     {
    '''configuration_xlm''': ['''XLM_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''XLMConfig''', '''XLMOnnxConfig'''],
    '''tokenization_xlm''': ['''XLMTokenizer'''],
}

try:
				if not is_torch_available():
								raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
				pass
else:
				UpperCAmelCase						:   Union[str, Any]			  =     [
				    '''XLM_PRETRAINED_MODEL_ARCHIVE_LIST''',
				    '''XLMForMultipleChoice''',
				    '''XLMForQuestionAnswering''',
				    '''XLMForQuestionAnsweringSimple''',
				    '''XLMForSequenceClassification''',
				    '''XLMForTokenClassification''',
				    '''XLMModel''',
				    '''XLMPreTrainedModel''',
				    '''XLMWithLMHeadModel''',
				]

try:
				if not is_tf_available():
								raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
				pass
else:
				UpperCAmelCase						:   Optional[Any]			  =     [
				    '''TF_XLM_PRETRAINED_MODEL_ARCHIVE_LIST''',
				    '''TFXLMForMultipleChoice''',
				    '''TFXLMForQuestionAnsweringSimple''',
				    '''TFXLMForSequenceClassification''',
				    '''TFXLMForTokenClassification''',
				    '''TFXLMMainLayer''',
				    '''TFXLMModel''',
				    '''TFXLMPreTrainedModel''',
				    '''TFXLMWithLMHeadModel''',
				]


if TYPE_CHECKING:
				from .configuration_xlm import XLM_PRETRAINED_CONFIG_ARCHIVE_MAP, XLMConfig, XLMOnnxConfig
				from .tokenization_xlm import XLMTokenizer

				try:
								if not is_torch_available():
												raise OptionalDependencyNotAvailable()
				except OptionalDependencyNotAvailable:
								pass
				else:
								from .modeling_xlm import (
								    XLM_PRETRAINED_MODEL_ARCHIVE_LIST,
								    XLMForMultipleChoice,
								    XLMForQuestionAnswering,
								    XLMForQuestionAnsweringSimple,
								    XLMForSequenceClassification,
								    XLMForTokenClassification,
								    XLMModel,
								    XLMPreTrainedModel,
								    XLMWithLMHeadModel,
								)

				try:
								if not is_tf_available():
												raise OptionalDependencyNotAvailable()
				except OptionalDependencyNotAvailable:
								pass
				else:
								from .modeling_tf_xlm import (
								    TF_XLM_PRETRAINED_MODEL_ARCHIVE_LIST,
								    TFXLMForMultipleChoice,
								    TFXLMForQuestionAnsweringSimple,
								    TFXLMForSequenceClassification,
								    TFXLMForTokenClassification,
								    TFXLMMainLayer,
								    TFXLMModel,
								    TFXLMPreTrainedModel,
								    TFXLMWithLMHeadModel,
								)

else:
				import sys

				UpperCAmelCase						:   str			  =     _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)




