




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       a					(       unittest.TestCase      ):

     '''simple docstring'''





     lowerCAmelCase    :	List[Any]       		=  ['[UNK]', '[CLS]', '[SEP]', '[PAD]', '[MASK]', 'bla', 'blou']
     def   lowerCamelCase_       (     self       :      str    ):
         UpperCAmelCase_										=					0
     def   lowerCamelCase_       (     self       :      int    ):
         UpperCAmelCase_										=					AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h'''    )
         self.assertIsInstance(__snake_case ,   __snake_case    )
     def   lowerCamelCase_       (     self       :      Dict    ):
         with tempfile.TemporaryDirectory() as tmpdirname:
             UpperCAmelCase_										=					WavaVecaConfig()
             UpperCAmelCase_										=					AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h'''    )

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

             UpperCAmelCase_										=					AutoProcessor.from_pretrained(__snake_case    )

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

             UpperCAmelCase_										=					AutoProcessor.from_pretrained(__snake_case    )

         self.assertIsInstance(__snake_case ,   __snake_case    )
     def   lowerCamelCase_       (     self       :      List[str]    ):
         with tempfile.TemporaryDirectory() as tmpdirname:
             UpperCAmelCase_										=					WavaVecaFeatureExtractor()
             UpperCAmelCase_										=					AutoTokenizer.from_pretrained('''facebook/wav2vec2-base-960h'''    )

             UpperCAmelCase_										=					WavaVecaProcessor(__snake_case ,   __snake_case    )

             # save in new folder
             processor.save_pretrained(__snake_case    )

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

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

             UpperCAmelCase_										=					AutoProcessor.from_pretrained(__snake_case    )

         self.assertIsInstance(__snake_case ,   __snake_case    )
     def   lowerCamelCase_       (     self       :      List[str]    ):
         with tempfile.TemporaryDirectory() as tmpdirname:
             UpperCAmelCase_										=					WavaVecaFeatureExtractor()
             UpperCAmelCase_										=					AutoTokenizer.from_pretrained('''facebook/wav2vec2-base-960h'''    )

             UpperCAmelCase_										=					WavaVecaProcessor(__snake_case ,   __snake_case    )

             # save in new folder
             processor.save_pretrained(__snake_case    )

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

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

             UpperCAmelCase_										=					AutoProcessor.from_pretrained(__snake_case    )

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

             UpperCAmelCase_										=					AutoProcessor.from_pretrained(__snake_case    )

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

         UpperCAmelCase_										=					AutoProcessor.from_pretrained('''hf-internal-testing/test_dynamic_processor''' ,   trust_remote_code=__snake_case    )
         self.assertTrue(processor.special_attribute_present    )
         self.assertEqual(processor.__class__.__name__ ,   '''NewProcessor'''    )

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

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

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

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

             UpperCAmelCase_										=					CustomProcessor(__snake_case ,   __snake_case    )

             with tempfile.TemporaryDirectory() as tmp_dir:
                 processor.save_pretrained(__snake_case    )
                 UpperCAmelCase_										=					AutoProcessor.from_pretrained(__snake_case    )
                 self.assertIsInstance(__snake_case ,   __snake_case    )

         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       a					(       _A      ):

              '''simple docstring'''





              lowerCAmelCase    :	Dict       		=  False



         class       a					(       _A      ):

              '''simple docstring'''





              lowerCAmelCase    :	str       		=  False



         class       a					(       _A      ):

              '''simple docstring'''





              lowerCAmelCase    :	Optional[int]       		=  'AutoFeatureExtractor'
              lowerCAmelCase    :	Optional[Any]       		=  'AutoTokenizer'
              lowerCAmelCase    :	str       		=  False

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




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



@is_staging_test
class       a					(       unittest.TestCase      ):

     '''simple docstring'''





     lowerCAmelCase    :	Any       		=  ['[UNK]', '[CLS]', '[SEP]', '[PAD]', '[MASK]', 'bla', 'blou']
     @classmethod
     def   lowerCamelCase_       (     cls       :      Optional[Any]    ):
         UpperCAmelCase_										=					TOKEN
         HfFolder.save_token(__snake_case    )
     @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       :      str    ):
         UpperCAmelCase_										=					WavaVecaProcessor.from_pretrained(__snake_case    )
         with tempfile.TemporaryDirectory() as tmp_dir:
             processor.save_pretrained(
                 os.path.join(__snake_case ,   '''test-processor'''    ) ,   push_to_hub=__snake_case ,   use_auth_token=self._token    )

             UpperCAmelCase_										=					WavaVecaProcessor.from_pretrained(F'{USER}/test-processor'    )
             for k, v in processor.feature_extractor.__dict__.items():
                 self.assertEqual(__snake_case ,   getattr(new_processor.feature_extractor ,   __snake_case    )    )
             self.assertDictEqual(new_processor.tokenizer.get_vocab() ,   processor.tokenizer.get_vocab()    )
     def   lowerCamelCase_       (     self       :      List[str]    ):
         UpperCAmelCase_										=					WavaVecaProcessor.from_pretrained(__snake_case    )

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

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




     def   lowerCamelCase_       (     self       :      Any    ):
         CustomFeatureExtractor.register_for_auto_class()
         CustomTokenizer.register_for_auto_class()
         CustomProcessor.register_for_auto_class()

         UpperCAmelCase_										=					CustomFeatureExtractor.from_pretrained(__snake_case    )

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

         UpperCAmelCase_										=					CustomProcessor(__snake_case ,   __snake_case    )

         with tempfile.TemporaryDirectory() as tmp_dir:
             create_repo(F'{USER}/test-dynamic-processor' ,   token=self._token    )
             UpperCAmelCase_										=					Repository(__snake_case ,   clone_from=F'{USER}/test-dynamic-processor' ,   token=self._token    )
             processor.save_pretrained(__snake_case    )

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

             repo.push_to_hub()

         UpperCAmelCase_										=					AutoProcessor.from_pretrained(F'{USER}/test-dynamic-processor' ,   trust_remote_code=__snake_case    )
         # 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_flax_available,
    is_tf_available,
    is_tokenizers_available,
    is_torch_available,
)


_lowerCamelCase      			=						{
    'configuration_roberta': ['ROBERTA_PRETRAINED_CONFIG_ARCHIVE_MAP', 'RobertaConfig', 'RobertaOnnxConfig'],
    'tokenization_roberta': ['RobertaTokenizer'],
}

try:
 if not is_tokenizers_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 _lowerCamelCase      			=						['RobertaTokenizerFast']

try:
 if not is_torch_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 _lowerCamelCase      			=						[
     'ROBERTA_PRETRAINED_MODEL_ARCHIVE_LIST',
     'RobertaForCausalLM',
     'RobertaForMaskedLM',
     'RobertaForMultipleChoice',
     'RobertaForQuestionAnswering',
     'RobertaForSequenceClassification',
     'RobertaForTokenClassification',
     'RobertaModel',
     'RobertaPreTrainedModel',
 ]

try:
 if not is_tf_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 _lowerCamelCase      			=						[
     'TF_ROBERTA_PRETRAINED_MODEL_ARCHIVE_LIST',
     'TFRobertaForCausalLM',
     'TFRobertaForMaskedLM',
     'TFRobertaForMultipleChoice',
     'TFRobertaForQuestionAnswering',
     'TFRobertaForSequenceClassification',
     'TFRobertaForTokenClassification',
     'TFRobertaMainLayer',
     'TFRobertaModel',
     'TFRobertaPreTrainedModel',
 ]

try:
 if not is_flax_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 _lowerCamelCase      			=						[
     'FlaxRobertaForCausalLM',
     'FlaxRobertaForMaskedLM',
     'FlaxRobertaForMultipleChoice',
     'FlaxRobertaForQuestionAnswering',
     'FlaxRobertaForSequenceClassification',
     'FlaxRobertaForTokenClassification',
     'FlaxRobertaModel',
     'FlaxRobertaPreTrainedModel',
 ]


if TYPE_CHECKING:
 from .configuration_roberta import ROBERTA_PRETRAINED_CONFIG_ARCHIVE_MAP, RobertaConfig, RobertaOnnxConfig
 from .tokenization_roberta import RobertaTokenizer

 try:
  if not is_tokenizers_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .tokenization_roberta_fast import RobertaTokenizerFast

 try:
  if not is_torch_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_roberta import (
      ROBERTA_PRETRAINED_MODEL_ARCHIVE_LIST,
      RobertaForCausalLM,
      RobertaForMaskedLM,
      RobertaForMultipleChoice,
      RobertaForQuestionAnswering,
      RobertaForSequenceClassification,
      RobertaForTokenClassification,
      RobertaModel,
      RobertaPreTrainedModel,
  )

 try:
  if not is_tf_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_tf_roberta import (
      TF_ROBERTA_PRETRAINED_MODEL_ARCHIVE_LIST,
      TFRobertaForCausalLM,
      TFRobertaForMaskedLM,
      TFRobertaForMultipleChoice,
      TFRobertaForQuestionAnswering,
      TFRobertaForSequenceClassification,
      TFRobertaForTokenClassification,
      TFRobertaMainLayer,
      TFRobertaModel,
      TFRobertaPreTrainedModel,
  )

 try:
  if not is_flax_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_flax_roberta import (
      FlaxRobertaForCausalLM,
      FlaxRobertaForMaskedLM,
      FlaxRobertaForMultipleChoice,
      FlaxRobertaForQuestionAnswering,
      FlaxRobertaForSequenceClassification,
      FlaxRobertaForTokenClassification,
      FlaxRobertaModel,
      FlaxRobertaPreTrainedModel,
  )

else:
 import sys

 _lowerCamelCase      			=						_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)





