

'''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


_UpperCAmelCase  :		List[Any]  =		get_tests_dir('''fixtures/dummy_feature_extractor_config.json''')
_UpperCAmelCase  :		List[str]  =		get_tests_dir('''fixtures/vocab.json''')
_UpperCAmelCase  :		Optional[int]  =		get_tests_dir('''fixtures''')

class 						__magic_name__   (							unittest.TestCase ):
 UpperCamelCase__     =       ['[UNK]', '[CLS]', '[SEP]', '[PAD]', '[MASK]', 'bla', 'blou']







 def 						_A(       self   ):
   lowercase									=0







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







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

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

     lowercase									=AutoProcessor.from_pretrained(snake_case_   )

   self.assertIsInstance(snake_case_  , snake_case_   )







 def 						_A(       self   ):
   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'''   )   )

     lowercase									=AutoProcessor.from_pretrained(snake_case_   )

   self.assertIsInstance(snake_case_  , snake_case_   )







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

     lowercase									=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:
       lowercase									=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_   )   )

     lowercase									=AutoProcessor.from_pretrained(snake_case_   )

   self.assertIsInstance(snake_case_  , snake_case_   )







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

     lowercase									=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:
       lowercase									=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_   )   )

     lowercase									=AutoProcessor.from_pretrained(snake_case_   )

   self.assertIsInstance(snake_case_  , snake_case_   )







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

     lowercase									=AutoProcessor.from_pretrained(snake_case_   )

   self.assertIsInstance(snake_case_  , snake_case_   )







 def 						_A(       self   ):
   # If remote code is not set, we will time out when asking whether to load the model.
   with self.assertRaises(snake_case_   ):
     lowercase									=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_   ):
     lowercase									=AutoProcessor.from_pretrained(
         '''hf-internal-testing/test_dynamic_processor'''  , trust_remote_code=snake_case_   )

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

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

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

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

     lowercase									=CustomProcessor(snake_case_  , snake_case_   )

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

   class 						__magic_name__   (							__SCREAMING_SNAKE_CASE ):
    UpperCamelCase__     =       False

   class 						__magic_name__   (							__SCREAMING_SNAKE_CASE ):
    UpperCamelCase__     =       False

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







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

@is_staging_test
class 						__magic_name__   (							unittest.TestCase ):
 UpperCamelCase__     =       ['[UNK]', '[CLS]', '[SEP]', '[PAD]', '[MASK]', 'bla', 'blou']







 @classmethod
 def 						_A(       cls   ):
   lowercase									=TOKEN
   HfFolder.save_token(snake_case_   )







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

     lowercase									=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 						_A(       self   ):
   lowercase									=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'''  , )

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

   lowercase									=CustomFeatureExtractor.from_pretrained(snake_case_   )

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

   lowercase									=CustomProcessor(snake_case_  , snake_case_   )

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

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






import torch
import torch.nn as nn
from transformers.modeling_utils import ModuleUtilsMixin
from transformers.models.ta.modeling_ta import TaBlock, TaConfig, TaLayerNorm

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin
class 						UpperCAmelCase  (							UpperCAmelCase__			,  UpperCAmelCase__			,  UpperCAmelCase__     ):







						'''simple docstring'''





						@register_to_config
						def __init__(		self				:	Tuple			,      __lowercase				:	int			,      __lowercase				:	int			,      __lowercase				:	int			,      __lowercase				:	float			,      __lowercase				:	int			,      __lowercase				:	int			,      __lowercase				:	int			,      __lowercase				:	int			,      __lowercase				:	str			,      __lowercase				:	bool = False			,      ):






							"""simple docstring"""






							super().__init__()

							snake_case_						       =	nn.Embedding(__lowercase			,      __lowercase		)

							snake_case_						       =	nn.Embedding(__lowercase			,      __lowercase		)
							snake_case_						       =	False

							snake_case_						       =	nn.Dropout(p=__lowercase		)

							snake_case_						       =	TaConfig(
							    vocab_size=__lowercase			,      d_model=__lowercase			,      num_heads=__lowercase			,      d_kv=__lowercase			,      d_ff=__lowercase			,      dropout_rate=__lowercase			,      feed_forward_proj=__lowercase			,      is_decoder=__lowercase			,      is_encoder_decoder=__lowercase			,      )

							snake_case_						       =	nn.ModuleList()
							for lyr_num in range(__lowercase		):
								snake_case_						       =	TaBlock(__lowercase		)
								self.encoders.append(__lowercase		)

							snake_case_						       =	TaLayerNorm(__lowercase		)
							snake_case_						       =	nn.Dropout(p=__lowercase		)






						def 			snake_case__		(		self				:	str			,      __lowercase				:	Dict			,      __lowercase				:	str		):






							"""simple docstring"""






							snake_case_						       =	self.token_embedder(__lowercase		)

							snake_case_						       =	encoder_input_tokens.shape[1]
							snake_case_						       =	torch.arange(__lowercase			,      device=encoder_input_tokens.device		)
							x += self.position_encoding(__lowercase		)

							snake_case_						       =	self.dropout_pre(__lowercase		)

							# inverted the attention mask
							snake_case_						       =	encoder_input_tokens.size()
							snake_case_						       =	self.get_extended_attention_mask(__lowercase			,      __lowercase		)

							for lyr in self.encoders:
								snake_case_						       =	lyr(__lowercase			,      __lowercase		)[0]
							snake_case_						       =	self.layer_norm(__lowercase		)

							return self.dropout_post(__lowercase		), encoder_inputs_mask
