


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


UpperCAmelCase    =					get_tests_dir("""fixtures/dummy_feature_extractor_config.json""")
UpperCAmelCase    =					get_tests_dir("""fixtures/vocab.json""")
UpperCAmelCase    =					get_tests_dir("""fixtures""")







class lowerCAmelCase_		(				unittest.TestCase ):

 '''simple docstring'''

 __snake_case				=						["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "bla", "blou"]



 def 		UpperCamelCase__  (			self      ):
       snake_case_ =   0



 def 		UpperCamelCase__  (			self      ):
       snake_case_ =   AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h'''      )
       self.assertIsInstance(_UpperCAmelCase	,   _UpperCAmelCase      )



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

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

             snake_case_ =   AutoProcessor.from_pretrained(_UpperCAmelCase      )

       self.assertIsInstance(_UpperCAmelCase	,   _UpperCAmelCase      )



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

             snake_case_ =   AutoProcessor.from_pretrained(_UpperCAmelCase      )

       self.assertIsInstance(_UpperCAmelCase	,   _UpperCAmelCase      )



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

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

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

             snake_case_ =   AutoProcessor.from_pretrained(_UpperCAmelCase      )

       self.assertIsInstance(_UpperCAmelCase	,   _UpperCAmelCase      )



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

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

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

             snake_case_ =   AutoProcessor.from_pretrained(_UpperCAmelCase      )

       self.assertIsInstance(_UpperCAmelCase	,   _UpperCAmelCase      )



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

             snake_case_ =   AutoProcessor.from_pretrained(_UpperCAmelCase      )

       self.assertIsInstance(_UpperCAmelCase	,   _UpperCAmelCase      )



 def 		UpperCamelCase__  (			self      ):
       # If remote code is not set, we will time out when asking whether to load the model.
       with self.assertRaises(_UpperCAmelCase      ):
             snake_case_ =   AutoProcessor.from_pretrained('''hf-internal-testing/test_dynamic_processor'''      )
       # If remote code is disabled, we can't load this config.
       with self.assertRaises(_UpperCAmelCase      ):
             snake_case_ =   AutoProcessor.from_pretrained(
                 '''hf-internal-testing/test_dynamic_processor'''	,   trust_remote_code=_UpperCAmelCase      )

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

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

       snake_case_ =   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
             snake_case_ =   AutoProcessor.from_pretrained(
                 '''hf-internal-testing/test_dynamic_processor'''	,   trust_remote_code=_UpperCAmelCase	,   use_fast=_UpperCAmelCase      )
             snake_case_ =   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      ):
       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
             snake_case_ =   CustomFeatureExtractor.from_pretrained(_UpperCAmelCase      )

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

             snake_case_ =   CustomProcessor(_UpperCAmelCase	,   _UpperCAmelCase      )

             with tempfile.TemporaryDirectory() as tmp_dir:
                   processor.save_pretrained(_UpperCAmelCase      )
                   snake_case_ =   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 		UpperCamelCase__  (			self      ):







       class lowerCAmelCase_		(				lowerCamelCase__ ):

        '''simple docstring'''

        __snake_case				=						False







       class lowerCAmelCase_		(				lowerCamelCase__ ):

        '''simple docstring'''

        __snake_case				=						False







       class lowerCAmelCase_		(				lowerCamelCase__ ):

        '''simple docstring'''

        __snake_case				=						"AutoFeatureExtractor"
        __snake_case				=						"AutoTokenizer"
        __snake_case				=						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.
             snake_case_ =   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.
             snake_case_ =   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.
             snake_case_ =   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 		UpperCamelCase__  (			self      ):
       snake_case_ =   AutoProcessor.from_pretrained('''hf-internal-testing/tiny-random-bert'''      )
       self.assertEqual(processor.__class__.__name__	,   '''BertTokenizerFast'''      )





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







@is_staging_test
class lowerCAmelCase_		(				unittest.TestCase ):

 '''simple docstring'''

 __snake_case				=						["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "bla", "blou"]



 @classmethod
 def 		UpperCamelCase__  (			cls      ):
       snake_case_ =   TOKEN
       HfFolder.save_token(_UpperCAmelCase      )



 @classmethod
 def 		UpperCamelCase__  (			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 		UpperCamelCase__  (			self      ):
       snake_case_ =   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      )

             snake_case_ =   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 		UpperCamelCase__  (			self      ):
       snake_case_ =   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'''	,   )

             snake_case_ =   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 		UpperCamelCase__  (			self      ):
       CustomFeatureExtractor.register_for_auto_class()
       CustomTokenizer.register_for_auto_class()
       CustomProcessor.register_for_auto_class()

       snake_case_ =   CustomFeatureExtractor.from_pretrained(_UpperCAmelCase      )

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

       snake_case_ =   CustomProcessor(_UpperCAmelCase	,   _UpperCAmelCase      )

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

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



from dataclasses import dataclass
from typing import Optional, Tuple

import torch
from torch import nn
from transformers import RobertaPreTrainedModel, XLMRobertaConfig, XLMRobertaModel
from transformers.utils import ModelOutput







@dataclass
class lowerCAmelCase_		(				lowerCamelCase__ ):

 '''simple docstring'''

 __snake_case				=						None
 __snake_case				=						None
 __snake_case				=						None
 __snake_case				=						None







class lowerCAmelCase_		(				lowerCamelCase__ ):

 '''simple docstring'''

 def __init__(			self	,   _UpperCAmelCase=1	,   _UpperCAmelCase=0	,   _UpperCAmelCase=2	,   _UpperCAmelCase=5_12	,   _UpperCAmelCase="cls"	,   _UpperCAmelCase=False	,   _UpperCAmelCase=True	,   **_UpperCAmelCase	,   ):
       super().__init__(pad_token_id=_UpperCAmelCase	,   bos_token_id=_UpperCAmelCase	,   eos_token_id=_UpperCAmelCase	,   **_UpperCAmelCase      )
       snake_case_ =   project_dim
       snake_case_ =   pooler_fn
       snake_case_ =   learn_encoder
       snake_case_ =   use_attention_mask







class lowerCAmelCase_		(				lowerCamelCase__ ):

 '''simple docstring'''

 __snake_case				=						[R"pooler", R"logit_scale"]
 __snake_case				=						[R"position_ids", R"predictions.decoder.bias"]
 __snake_case				=						"roberta"
 __snake_case				=						RobertaSeriesConfig



 def __init__(			self	,   _UpperCAmelCase      ):
       super().__init__(_UpperCAmelCase      )
       snake_case_ =   XLMRobertaModel(_UpperCAmelCase      )
       snake_case_ =   nn.Linear(config.hidden_size	,   config.project_dim      )
       snake_case_ =   getattr(_UpperCAmelCase	,   '''has_pre_transformation'''	,   _UpperCAmelCase      )
       if self.has_pre_transformation:
             snake_case_ =   nn.Linear(config.hidden_size	,   config.project_dim      )
             snake_case_ =   nn.LayerNorm(config.hidden_size	,   eps=config.layer_norm_eps      )
       self.post_init()





 def 		UpperCamelCase__  (			self	,   _UpperCAmelCase = None	,   _UpperCAmelCase = None	,   _UpperCAmelCase = None	,   _UpperCAmelCase = None	,   _UpperCAmelCase = None	,   _UpperCAmelCase = None	,   _UpperCAmelCase = None	,   _UpperCAmelCase = None	,   _UpperCAmelCase = None	,   _UpperCAmelCase = None	,   _UpperCAmelCase = None	,   ):
       snake_case_ =   return_dict if return_dict is not None else self.config.use_return_dict

       snake_case_ =   self.base_model(
           input_ids=_UpperCAmelCase	,   attention_mask=_UpperCAmelCase	,   token_type_ids=_UpperCAmelCase	,   position_ids=_UpperCAmelCase	,   head_mask=_UpperCAmelCase	,   inputs_embeds=_UpperCAmelCase	,   encoder_hidden_states=_UpperCAmelCase	,   encoder_attention_mask=_UpperCAmelCase	,   output_attentions=_UpperCAmelCase	,   output_hidden_states=True if self.has_pre_transformation else output_hidden_states	,   return_dict=_UpperCAmelCase	,   )

       if self.has_pre_transformation:
             snake_case_ =   outputs['''hidden_states'''][-2]
             snake_case_ =   self.pre_LN(_UpperCAmelCase      )
             snake_case_ =   self.transformation_pre(_UpperCAmelCase      )

             return TransformationModelOutput(
                 projection_state=_UpperCAmelCase	,   last_hidden_state=outputs.last_hidden_state	,   hidden_states=outputs.hidden_states	,   attentions=outputs.attentions	,   )
       else:
             snake_case_ =   self.transformation(outputs.last_hidden_state      )
             return TransformationModelOutput(
                 projection_state=_UpperCAmelCase	,   last_hidden_state=outputs.last_hidden_state	,   hidden_states=outputs.hidden_states	,   attentions=outputs.attentions	,   )