


"""simple docstring"""






import os
from shutil import copyfile
from typing import List, Optional, Tuple

from tokenizers import processors

from ...tokenization_utils import AddedToken, BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging


if is_sentencepiece_available():
  from .tokenization_mbart import MBartTokenizer
else:
  _lowerCAmelCase							:    Any					   =	None


_lowerCAmelCase							:    Union[str, Any]					   =	logging.get_logger(__name__)


_lowerCAmelCase							:    Dict					   =	{'''vocab_file''': '''sentencepiece.bpe.model''', '''tokenizer_file''': '''tokenizer.json'''}

_lowerCAmelCase							:    str					   =	{
    '''vocab_file''': {
        '''facebook/mbart-large-en-ro''': (
            '''https://huggingface.co/facebook/mbart-large-en-ro/resolve/main/sentencepiece.bpe.model'''
        ),
        '''facebook/mbart-large-cc25''': (
            '''https://huggingface.co/facebook/mbart-large-cc25/resolve/main/sentencepiece.bpe.model'''
        ),
    },
    '''tokenizer_file''': {
        '''facebook/mbart-large-en-ro''': '''https://huggingface.co/facebook/mbart-large-en-ro/resolve/main/tokenizer.json''',
        '''facebook/mbart-large-cc25''': '''https://huggingface.co/facebook/mbart-large-cc25/resolve/main/tokenizer.json''',
    },
}

_lowerCAmelCase							:    List[Any]					   =	{
    '''facebook/mbart-large-en-ro''': 1024,
    '''facebook/mbart-large-cc25''': 1024,
}

# fmt: off
_lowerCAmelCase							:    Union[str, Any]					   =	['''ar_AR''', '''cs_CZ''', '''de_DE''', '''en_XX''', '''es_XX''', '''et_EE''', '''fi_FI''', '''fr_XX''', '''gu_IN''', '''hi_IN''', '''it_IT''', '''ja_XX''', '''kk_KZ''', '''ko_KR''', '''lt_LT''', '''lv_LV''', '''my_MM''', '''ne_NP''', '''nl_XX''', '''ro_RO''', '''ru_RU''', '''si_LK''', '''tr_TR''', '''vi_VN''', '''zh_CN''']




class 							A_      (       _a ):
  lowerCAmelCase__      			=	VOCAB_FILES_NAMES
  lowerCAmelCase__      			=	PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  lowerCAmelCase__      			=	PRETRAINED_VOCAB_FILES_MAP
  lowerCAmelCase__      			=	['input_ids', 'attention_mask']
  lowerCAmelCase__      			=	MBartTokenizer

  lowerCAmelCase__      			=	[]
  lowerCAmelCase__      			=	[]


  def __init__(			self:						List[str]							,__lowerCAmelCase:						List[Any]=None							,__lowerCAmelCase:						Tuple=None							,__lowerCAmelCase:						Optional[int]="<s>"							,__lowerCAmelCase:						List[Any]="</s>"							,__lowerCAmelCase:						List[str]="</s>"							,__lowerCAmelCase:						Optional[Any]="<s>"							,__lowerCAmelCase:						List[str]="<unk>"							,__lowerCAmelCase:						int="<pad>"							,__lowerCAmelCase:						List[Any]="<mask>"							,__lowerCAmelCase:						List[Any]=None							,__lowerCAmelCase:						Union[str, Any]=None							,__lowerCAmelCase:						Tuple=None							,**__lowerCAmelCase:						str							,):







     '''simple docstring'''







     _lowerCamelCase   :				int							=     AddedToken(__lowerCAmelCase							,lstrip=__lowerCAmelCase							,rstrip=__lowerCAmelCase    ) if isinstance(__lowerCAmelCase							,__lowerCAmelCase    ) else mask_token

     super().__init__(
         vocab_file=__lowerCAmelCase							,tokenizer_file=__lowerCAmelCase							,bos_token=__lowerCAmelCase							,eos_token=__lowerCAmelCase							,sep_token=__lowerCAmelCase							,cls_token=__lowerCAmelCase							,unk_token=__lowerCAmelCase							,pad_token=__lowerCAmelCase							,mask_token=__lowerCAmelCase							,src_lang=__lowerCAmelCase							,tgt_lang=__lowerCAmelCase							,additional_special_tokens=__lowerCAmelCase							,**__lowerCAmelCase							,)

     _lowerCamelCase   :				Any							=     vocab_file
     _lowerCamelCase   :				Dict							=     False if not self.vocab_file else True

     _lowerCamelCase   :				Tuple							=     FAIRSEQ_LANGUAGE_CODES.copy()

     if additional_special_tokens is not None:
        # Only add those special tokens if they are not already there.
        _additional_special_tokens.extend(
            [t for t in additional_special_tokens if t not in _additional_special_tokens]    )

     self.add_special_tokens({"additional_special_tokens": _additional_special_tokens}    )
     _lowerCamelCase   :				Optional[int]							=     {
         lang_code: self.convert_tokens_to_ids(__lowerCAmelCase    ) for lang_code in FAIRSEQ_LANGUAGE_CODES
     }

     _lowerCamelCase   :				Any							=     src_lang if src_lang is not None else "en_XX"
     _lowerCamelCase   :				Optional[int]							=     self.convert_tokens_to_ids(self._src_lang    )
     _lowerCamelCase   :				Optional[Any]							=     tgt_lang
     self.set_src_lang_special_tokens(self._src_lang    )


  @property
  def 							_lowercase					(			self:						Tuple    ):







     '''simple docstring'''







     return self._src_lang


  @src_lang.setter
  def 							_lowercase					(			self:						Any							,__lowerCAmelCase:						str    ):







     '''simple docstring'''







     _lowerCamelCase   :				int							=     new_src_lang
     self.set_src_lang_special_tokens(self._src_lang    )


  def 							_lowercase					(			self:						Dict							,__lowerCAmelCase:						List[int]							,__lowerCAmelCase:						Optional[List[int]] = None    ):







     '''simple docstring'''







     if token_ids_a is None:
        return self.prefix_tokens + token_ids_a + self.suffix_tokens
     # We don't expect to process pairs, but leave the pair logic for API consistency
     return self.prefix_tokens + token_ids_a + token_ids_a + self.suffix_tokens


  def 							_lowercase					(			self:						Any							,__lowerCAmelCase:						List[int]							,__lowerCAmelCase:						Optional[List[int]] = None    ):







     '''simple docstring'''







     _lowerCamelCase   :				List[str]							=     [self.sep_token_id]
     _lowerCamelCase   :				Optional[Any]							=     [self.cls_token_id]

     if token_ids_a is None:
        return len(cls + token_ids_a + sep    ) * [0]
     return len(cls + token_ids_a + sep + sep + token_ids_a + sep    ) * [0]


  def 							_lowercase					(			self:						Optional[Any]							,__lowerCAmelCase:						Dict							,__lowerCAmelCase:						str							,__lowerCAmelCase:						Optional[str]							,__lowerCAmelCase:						Optional[str]							,**__lowerCAmelCase:						List[str]    ):







     '''simple docstring'''







     if src_lang is None or tgt_lang is None:
        raise ValueError("Translation requires a `src_lang` and a `tgt_lang` for this model"    )
     _lowerCamelCase   :				Tuple							=     src_lang
     _lowerCamelCase   :				Optional[int]							=     self(__lowerCAmelCase							,add_special_tokens=__lowerCAmelCase							,return_tensors=__lowerCAmelCase							,**__lowerCAmelCase    )
     _lowerCamelCase   :				List[str]							=     self.convert_tokens_to_ids(__lowerCAmelCase    )
     _lowerCamelCase   :				Optional[int]							=     tgt_lang_id
     return inputs


  def 							_lowercase					(			self:						List[str]							,__lowerCAmelCase:						List[str]							,__lowerCAmelCase:						str = "en_XX"							,__lowerCAmelCase:						Optional[List[str]] = None							,__lowerCAmelCase:						str = "ro_RO"							,**__lowerCAmelCase:						Dict							,):







     '''simple docstring'''







     _lowerCamelCase   :				List[Any]							=     src_lang
     _lowerCamelCase   :				List[str]							=     tgt_lang
     return super().prepare_seqaseq_batch(__lowerCAmelCase							,__lowerCAmelCase							,**__lowerCAmelCase    )


  def 							_lowercase					(			self:						Tuple    ):







     '''simple docstring'''







     return self.set_src_lang_special_tokens(self.src_lang    )


  def 							_lowercase					(			self:						List[str]    ):







     '''simple docstring'''







     return self.set_tgt_lang_special_tokens(self.tgt_lang    )


  def 							_lowercase					(			self:						List[str]							,__lowerCAmelCase:						Dict    ):







     '''simple docstring'''







     _lowerCamelCase   :				Any							=     self.convert_tokens_to_ids(__lowerCAmelCase    )
     _lowerCamelCase   :				Any							=     []
     _lowerCamelCase   :				Dict							=     [self.eos_token_id, self.cur_lang_code]

     _lowerCamelCase   :				Union[str, Any]							=     self.convert_ids_to_tokens(self.prefix_tokens    )
     _lowerCamelCase   :				Dict							=     self.convert_ids_to_tokens(self.suffix_tokens    )

     _lowerCamelCase   :				Tuple							=     processors.TemplateProcessing(
         single=prefix_tokens_str + ["$A"] + suffix_tokens_str							,pair=prefix_tokens_str + ["$A", "$B"] + suffix_tokens_str							,special_tokens=list(zip(prefix_tokens_str + suffix_tokens_str							,self.prefix_tokens + self.suffix_tokens    )    )							,)


  def 							_lowercase					(			self:						Optional[Any]							,__lowerCAmelCase:						str    ):







     '''simple docstring'''







     _lowerCamelCase   :				int							=     self.convert_tokens_to_ids(__lowerCAmelCase    )
     _lowerCamelCase   :				Tuple							=     []
     _lowerCamelCase   :				Tuple							=     [self.eos_token_id, self.cur_lang_code]

     _lowerCamelCase   :				str							=     self.convert_ids_to_tokens(self.prefix_tokens    )
     _lowerCamelCase   :				Any							=     self.convert_ids_to_tokens(self.suffix_tokens    )

     _lowerCamelCase   :				Optional[Any]							=     processors.TemplateProcessing(
         single=prefix_tokens_str + ["$A"] + suffix_tokens_str							,pair=prefix_tokens_str + ["$A", "$B"] + suffix_tokens_str							,special_tokens=list(zip(prefix_tokens_str + suffix_tokens_str							,self.prefix_tokens + self.suffix_tokens    )    )							,)






  def 							_lowercase					(			self:						Dict							,__lowerCAmelCase:						str							,__lowerCAmelCase:						Optional[str] = None    ):







     '''simple docstring'''







     if not self.can_save_slow_tokenizer:
        raise ValueError(
            "Your fast tokenizer does not have the necessary information to save the vocabulary for a slow "
            "tokenizer."    )

     if not os.path.isdir(__lowerCAmelCase    ):
        logger.error(F"""Vocabulary path ({save_directory}) should be a directory."""    )
        return
     _lowerCamelCase   :				List[str]							=     os.path.join(
         __lowerCAmelCase							,(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]    )

     if os.path.abspath(self.vocab_file    ) != os.path.abspath(__lowerCAmelCase    ):
        copyfile(self.vocab_file							,__lowerCAmelCase    )

     return (out_vocab_file,)






"""simple docstring"""




import unittest

from transformers import EsmConfig, is_torch_available
from transformers.testing_utils import TestCasePlus, require_torch, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
		import torch

		from transformers import EsmForMaskedLM, EsmForSequenceClassification, EsmForTokenClassification, EsmModel
		from transformers.models.esm.modeling_esm import (
		    ESM_PRETRAINED_MODEL_ARCHIVE_LIST,
		    EsmEmbeddings,
		    create_position_ids_from_input_ids,
		)


class 			__UpperCamelCase							:






							def __init__(  self ,   lowerCAmelCase__ ,   lowerCAmelCase__=13 ,   lowerCAmelCase__=7 ,   lowerCAmelCase__=False ,   lowerCAmelCase__=True ,   lowerCAmelCase__=False ,   lowerCAmelCase__=True ,   lowerCAmelCase__=33 ,   lowerCAmelCase__=32 ,   lowerCAmelCase__=5 ,   lowerCAmelCase__=4 ,   lowerCAmelCase__=37 ,   lowerCAmelCase__="gelu" ,   lowerCAmelCase__=0.1 ,   lowerCAmelCase__=0.1 ,   lowerCAmelCase__=512 ,   lowerCAmelCase__=16 ,   lowerCAmelCase__=2 ,   lowerCAmelCase__=0.02 ,   lowerCAmelCase__=3 ,   lowerCAmelCase__=4 ,   lowerCAmelCase__=None ,   )      ->					List[str]:
								a :						List[str]      =							parent
								a :						Optional[int]      =							batch_size
								a :						Optional[int]      =							seq_length
								a :						Tuple      =							is_training
								a :						Optional[int]      =							use_input_mask
								a :						List[Any]      =							use_token_type_ids
								a :						Tuple      =							use_labels
								a :						Union[str, Any]      =							vocab_size
								a :						str      =							hidden_size
								a :						List[Any]      =							num_hidden_layers
								a :						Dict      =							num_attention_heads
								a :						int      =							intermediate_size
								a :						str      =							hidden_act
								a :						Optional[int]      =							hidden_dropout_prob
								a :						str      =							attention_probs_dropout_prob
								a :						str      =							max_position_embeddings
								a :						List[str]      =							type_vocab_size
								a :						List[str]      =							type_sequence_label_size
								a :						Union[str, Any]      =							initializer_range
								a :						int      =							num_labels
								a :						Optional[int]      =							num_choices
								a :						Union[str, Any]      =							scope






							def        __a  (  self )      ->					Optional[Any]:
								a :						str      =							ids_tensor([self.batch_size, self.seq_length] ,   self.vocab_size )

								a :						Dict      =							None
								if self.use_input_mask:
									a :						str      =							random_attention_mask([self.batch_size, self.seq_length] )

								a :						Union[str, Any]      =							None
								a :						int      =							None
								a :						int      =							None
								if self.use_labels:
									a :						Optional[int]      =							ids_tensor([self.batch_size] ,   self.type_sequence_label_size )
									a :						Optional[int]      =							ids_tensor([self.batch_size, self.seq_length] ,   self.num_labels )
									a :						List[Any]      =							ids_tensor([self.batch_size] ,   self.num_choices )

								a :						str      =							self.get_config()

								return config, input_ids, input_mask, sequence_labels, token_labels, choice_labels






							def        __a  (  self )      ->					Optional[int]:
								return EsmConfig(
								    vocab_size=self.vocab_size ,   hidden_size=self.hidden_size ,   pad_token_id=1 ,   num_hidden_layers=self.num_hidden_layers ,   num_attention_heads=self.num_attention_heads ,   intermediate_size=self.intermediate_size ,   hidden_act=self.hidden_act ,   hidden_dropout_prob=self.hidden_dropout_prob ,   attention_probs_dropout_prob=self.attention_probs_dropout_prob ,   max_position_embeddings=self.max_position_embeddings ,   type_vocab_size=self.type_vocab_size ,   initializer_range=self.initializer_range ,   )






							def        __a  (  self ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ )      ->					str:
								a :						Union[str, Any]      =							EsmModel(config=lowerCAmelCase__ )
								model.to(lowerCAmelCase__ )
								model.eval()
								a :						Optional[int]      =							model(lowerCAmelCase__ ,   attention_mask=lowerCAmelCase__ )
								a :						List[str]      =							model(lowerCAmelCase__ )
								a :						Optional[Any]      =							model(lowerCAmelCase__ )

								self.parent.assertEqual(result.last_hidden_state.shape ,   (self.batch_size, self.seq_length, self.hidden_size) )
								self.parent.assertEqual(result.pooler_output.shape ,   (self.batch_size, self.hidden_size) )






							def        __a  (  self ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ )      ->					Tuple:
								a :						int      =							EsmForMaskedLM(config=lowerCAmelCase__ )
								model.to(lowerCAmelCase__ )
								model.eval()
								a :						Union[str, Any]      =							model(lowerCAmelCase__ ,   attention_mask=lowerCAmelCase__ ,   labels=lowerCAmelCase__ )
								self.parent.assertEqual(result.logits.shape ,   (self.batch_size, self.seq_length, self.vocab_size) )






							def        __a  (  self ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ )      ->					Optional[Any]:
								a :						str      =							self.num_labels
								a :						int      =							EsmForTokenClassification(config=lowerCAmelCase__ )
								model.to(lowerCAmelCase__ )
								model.eval()
								a :						int      =							model(lowerCAmelCase__ ,   attention_mask=lowerCAmelCase__ ,   labels=lowerCAmelCase__ )
								self.parent.assertEqual(result.logits.shape ,   (self.batch_size, self.seq_length, self.num_labels) )






							def        __a  (  self )      ->					Union[str, Any]:
								a :						str      =							self.prepare_config_and_inputs()
								(
								    (
								    a
								),			(
								    a
								),			(
								    a
								),			(
								    a
								),			(
								    a
								),			(
								    a
								),			
								) :						Optional[int]      =							config_and_inputs
								a :						int      =							{"input_ids": input_ids, "attention_mask": input_mask}
								return config, inputs_dict


@require_torch
class 			__UpperCamelCase							(		a__						,			a__						,			unittest.TestCase		):
							lowerCamelCase  :   Optional[Any]    =False

							lowerCamelCase  :   Tuple    =(
							    (
							        EsmForMaskedLM,
							        EsmModel,
							        EsmForSequenceClassification,
							        EsmForTokenClassification,
							    )
							    if is_torch_available()
							    else ()
							)
							lowerCamelCase  :   Tuple    =()
							lowerCamelCase  :   Optional[int]    =(
							    {
							        """feature-extraction""": EsmModel,
							        """fill-mask""": EsmForMaskedLM,
							        """text-classification""": EsmForSequenceClassification,
							        """token-classification""": EsmForTokenClassification,
							        """zero-shot""": EsmForSequenceClassification,
							    }
							    if is_torch_available()
							    else {}
							)
							lowerCamelCase  :   Any    =True






							def        __a  (  self )      ->					Tuple:
								a :						str      =							EsmModelTester(self )
								a :						Union[str, Any]      =							ConfigTester(self ,   config_class=lowerCAmelCase__ ,   hidden_size=37 )






							def        __a  (  self )      ->					Dict:
								self.config_tester.run_common_tests()






							def        __a  (  self )      ->					Union[str, Any]:
								a :						Tuple      =							self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_model(*lowerCAmelCase__ )






							def        __a  (  self )      ->					List[str]:
								a :						Dict      =							self.model_tester.prepare_config_and_inputs()
								for type in ["absolute", "relative_key", "relative_key_query"]:
									a :						List[str]      =							type
									self.model_tester.create_and_check_model(*lowerCAmelCase__ )






							def        __a  (  self )      ->					Optional[Any]:
								a :						Dict      =							self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_masked_lm(*lowerCAmelCase__ )






							def        __a  (  self )      ->					Dict:
								a :						int      =							self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_token_classification(*lowerCAmelCase__ )






							@slow
							def        __a  (  self )      ->					Optional[Any]:
								for model_name in ESM_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
									a :						Tuple      =							EsmModel.from_pretrained(lowerCAmelCase__ )
									self.assertIsNotNone(lowerCAmelCase__ )






							def        __a  (  self )      ->					Optional[int]:
								a :						List[str]      =							self.model_tester.prepare_config_and_inputs()[0]
								a :						str      =							EsmEmbeddings(config=lowerCAmelCase__ )

								a :						Optional[Any]      =							torch.as_tensor([[12, 31, 13, model.padding_idx]] )
								a :						Tuple      =							torch.as_tensor(
								    [
								        [
								            0 + model.padding_idx + 1,
								            1 + model.padding_idx + 1,
								            2 + model.padding_idx + 1,
								            model.padding_idx,
								        ]
								    ] )
								a :						str      =							create_position_ids_from_input_ids(lowerCAmelCase__ ,   model.padding_idx )
								self.assertEqual(position_ids.shape ,   expected_positions.shape )
								self.assertTrue(torch.all(torch.eq(lowerCAmelCase__ ,   lowerCAmelCase__ ) ) )






							def        __a  (  self )      ->					List[Any]:
								a :						int      =							self.model_tester.prepare_config_and_inputs()[0]
								a :						Dict      =							EsmEmbeddings(config=lowerCAmelCase__ )

								a :						Any      =							torch.empty(2 ,   4 ,   30 )
								a :						str      =							[
								    0 + embeddings.padding_idx + 1,
								    1 + embeddings.padding_idx + 1,
								    2 + embeddings.padding_idx + 1,
								    3 + embeddings.padding_idx + 1,
								]
								a :						Optional[Any]      =							torch.as_tensor([expected_single_positions, expected_single_positions] )
								a :						str      =							embeddings.create_position_ids_from_inputs_embeds(lowerCAmelCase__ )
								self.assertEqual(position_ids.shape ,   expected_positions.shape )
								self.assertTrue(torch.all(torch.eq(lowerCAmelCase__ ,   lowerCAmelCase__ ) ) )






							@unittest.skip("Esm does not support embedding resizing" )
							def        __a  (  self )      ->					Any:
								pass






							@unittest.skip("Esm does not support embedding resizing" )
							def        __a  (  self )      ->					Dict:
								pass






							@unittest.skip("Will be fixed soon by reducing the size of the model used for common tests." )
							def        __a  (  self )      ->					str:
								pass


@require_torch
class 			__UpperCamelCase							(		a__		):






							@slow
							def        __a  (  self )      ->					int:
								with torch.no_grad():
									a :						str      =							EsmForMaskedLM.from_pretrained("facebook/esm2_t6_8M_UR50D" )
									model.eval()
									a :						List[str]      =							torch.tensor([[0, 1, 2, 3, 4, 5]] )
									a :						Any      =							model(lowerCAmelCase__ )[0]

									a :						Union[str, Any]      =							33

									a :						str      =							torch.Size((1, 6, vocab_size) )
									self.assertEqual(output.shape ,   lowerCAmelCase__ )

									a :						Optional[int]      =							torch.tensor(
									    [[[8.9_215, -10.5_898, -6.4_671], [-6.3_967, -13.9_114, -1.1_212], [-7.7_812, -13.9_516, -3.7_406]]] )
									self.assertTrue(torch.allclose(output[:, :3, :3] ,   lowerCAmelCase__ ,   atol=1E-4 ) )






							@slow
							def        __a  (  self )      ->					str:
								with torch.no_grad():
									a :						Optional[int]      =							EsmModel.from_pretrained("facebook/esm2_t6_8M_UR50D" )
									model.eval()

									a :						str      =							torch.tensor([[0, 6, 4, 13, 5, 4, 16, 12, 11, 7, 2]] )
									a :						Tuple      =							model(lowerCAmelCase__ )[0]
									# compare the actual values for a slice.
									a :						Union[str, Any]      =							torch.tensor(
									    [[[0.1_444, 0.5_413, 0.3_248], [0.3_034, 0.0_053, 0.3_108], [0.3_228, -0.2_499, 0.3_415]]] )
									self.assertTrue(torch.allclose(output[:, :3, :3] ,   lowerCAmelCase__ ,   atol=1E-4 ) )


