



"""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:
     lowercase__    :  Dict							    =	None


lowercase__    :  Optional[int]							    =	logging.get_logger(__name__)


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

lowercase__    :  Optional[Any]							    =	{
    '''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''',
    },
}

lowercase__    :  Dict							    =	{
    '''facebook/mbart-large-en-ro''': 10_24,
    '''facebook/mbart-large-cc25''': 10_24,
}

# fmt: off
lowercase__    :  Dict							    =	['''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        _UpperCAmelCase       (			lowerCAmelCase__):
 _lowerCAmelCase   :			str							=		VOCAB_FILES_NAMES
 _lowerCAmelCase   :			Optional[int]							=		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 _lowerCAmelCase   :			List[str]							=		PRETRAINED_VOCAB_FILES_MAP
 _lowerCAmelCase   :			Tuple							=		["""input_ids""", """attention_mask"""]
 _lowerCAmelCase   :			Dict							=		MBartTokenizer

 _lowerCAmelCase   :			List[int]							=		[]
 _lowerCAmelCase   :			List[int]							=		[]

 def __init__(     self		:       Optional[int]  ,     lowercase_		:       Dict=None  ,     lowercase_		:       str=None  ,     lowercase_		:       Any="<s>"  ,     lowercase_		:       Dict="</s>"  ,     lowercase_		:       List[Any]="</s>"  ,     lowercase_		:       Dict="<s>"  ,     lowercase_		:       List[Any]="<unk>"  ,     lowercase_		:       Any="<pad>"  ,     lowercase_		:       Union[str, Any]="<mask>"  ,     lowercase_		:       List[str]=None  ,     lowercase_		:       Union[str, Any]=None  ,     lowercase_		:       Tuple=None  ,     **lowercase_		:       str  ,     ):
     # Mask token behave like a normal word, i.e. include the space before it
     snake_case_ :     Union[str, Any]				=	AddedToken(lowercase_  ,     lstrip=lowercase_  ,     rstrip=lowercase_				) if isinstance(lowercase_  ,     lowercase_				) else mask_token

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

     snake_case_ :     Optional[int]				=	vocab_file
     snake_case_ :     Optional[int]				=	False if not self.vocab_file else True

     snake_case_ :     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}				)
     snake_case_ :     int				=	{
         lang_code: self.convert_tokens_to_ids(lowercase_				) for lang_code in FAIRSEQ_LANGUAGE_CODES
     }

     snake_case_ :     List[Any]				=	src_lang if src_lang is not None else '''en_XX'''
     snake_case_ :     str				=	self.convert_tokens_to_ids(self._src_lang				)
     snake_case_ :     Any				=	tgt_lang
     self.set_src_lang_special_tokens(self._src_lang				)

 @property
 def 						_snake_case					(     self		:       Optional[Any]				):
     return self._src_lang

 @src_lang.setter
 def 						_snake_case					(     self		:       Tuple  ,     lowercase_		:       str				):
     snake_case_ :     Optional[int]				=	new_src_lang
     self.set_src_lang_special_tokens(self._src_lang				)

 def 						_snake_case					(     self		:       Dict  ,     lowercase_		:       List[int]  ,     lowercase_		:       Optional[List[int]] = None				):
     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 						_snake_case					(     self		:       Tuple  ,     lowercase_		:       List[int]  ,     lowercase_		:       Optional[List[int]] = None				):
     snake_case_ :     str				=	[self.sep_token_id]
     snake_case_ :     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 						_snake_case					(     self		:       Tuple  ,     lowercase_		:       List[Any]  ,     lowercase_		:       str  ,     lowercase_		:       Optional[str]  ,     lowercase_		:       Optional[str]  ,     **lowercase_		:       Tuple				):
     if src_lang is None or tgt_lang is None:
         raise ValueError('''Translation requires a `src_lang` and a `tgt_lang` for this model'''				)
     snake_case_ :     Tuple				=	src_lang
     snake_case_ :     Any				=	self(lowercase_  ,     add_special_tokens=lowercase_  ,     return_tensors=lowercase_  ,     **lowercase_				)
     snake_case_ :     Dict				=	self.convert_tokens_to_ids(lowercase_				)
     snake_case_ :     Optional[int]				=	tgt_lang_id
     return inputs

 def 						_snake_case					(     self		:       List[str]  ,     lowercase_		:       List[str]  ,     lowercase_		:       str = "en_XX"  ,     lowercase_		:       Optional[List[str]] = None  ,     lowercase_		:       str = "ro_RO"  ,     **lowercase_		:       Optional[int]  ,     ):
     snake_case_ :     List[str]				=	src_lang
     snake_case_ :     Union[str, Any]				=	tgt_lang
     return super().prepare_seqaseq_batch(lowercase_  ,     lowercase_  ,     **lowercase_				)

 def 						_snake_case					(     self		:       Optional[Any]				):
     return self.set_src_lang_special_tokens(self.src_lang				)

 def 						_snake_case					(     self		:       str				):
     return self.set_tgt_lang_special_tokens(self.tgt_lang				)

 def 						_snake_case					(     self		:       List[Any]  ,     lowercase_		:       List[str]				):
     snake_case_ :     Dict				=	self.convert_tokens_to_ids(lowercase_				)
     snake_case_ :     Optional[int]				=	[]
     snake_case_ :     Union[str, Any]				=	[self.eos_token_id, self.cur_lang_code]

     snake_case_ :     List[str]				=	self.convert_ids_to_tokens(self.prefix_tokens				)
     snake_case_ :     int				=	self.convert_ids_to_tokens(self.suffix_tokens				)

     snake_case_ :     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 						_snake_case					(     self		:       List[str]  ,     lowercase_		:       str				):
     snake_case_ :     Optional[int]				=	self.convert_tokens_to_ids(lowercase_				)
     snake_case_ :     List[str]				=	[]
     snake_case_ :     Union[str, Any]				=	[self.eos_token_id, self.cur_lang_code]

     snake_case_ :     Dict				=	self.convert_ids_to_tokens(self.prefix_tokens				)
     snake_case_ :     int				=	self.convert_ids_to_tokens(self.suffix_tokens				)

     snake_case_ :     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 						_snake_case					(     self		:       List[str]  ,     lowercase_		:       str  ,     lowercase_		:       Optional[str] = None				):
     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(lowercase_				):
         logger.error(f"Vocabulary path ({save_directory}) should be a directory."				)
         return
     snake_case_ :     Any				=	os.path.join(
         lowercase_  ,     (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']				)

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

     return (out_vocab_file,)










"""simple docstring"""

import unittest

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

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
     import torch

     from transformers import (
         OPENAI_GPT_PRETRAINED_MODEL_ARCHIVE_LIST,
         OpenAIGPTConfig,
         OpenAIGPTDoubleHeadsModel,
         OpenAIGPTForSequenceClassification,
         OpenAIGPTLMHeadModel,
         OpenAIGPTModel,
     )




class        _UpperCAmelCase       :

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

 def 						_snake_case					(     self		:       int				):
     snake_case_ :     Optional[Any]				=	ids_tensor([self.batch_size, self.seq_length]  ,     self.vocab_size				)

     snake_case_ :     Optional[Any]				=	None
     if self.use_token_type_ids:
         snake_case_ :     Tuple				=	ids_tensor([self.batch_size, self.seq_length]  ,     self.type_vocab_size				)

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

     snake_case_ :     int				=	OpenAIGPTConfig(
         vocab_size=self.vocab_size  ,     n_embd=self.hidden_size  ,     n_layer=self.num_hidden_layers  ,     n_head=self.num_attention_heads  ,     n_positions=self.max_position_embeddings  ,     pad_token_id=self.pad_token_id  ,     )

     snake_case_ :     Any				=	ids_tensor([self.num_hidden_layers, self.num_attention_heads]  ,     2				)

     return (
         config,
         input_ids,
         head_mask,
         token_type_ids,
         sequence_labels,
         token_labels,
         choice_labels,
     )

 def 						_snake_case					(     self		:       Tuple  ,     lowercase_		:       Any  ,     lowercase_		:       Union[str, Any]  ,     lowercase_		:       str  ,     lowercase_		:       Dict  ,     *lowercase_		:       Dict				):
     snake_case_ :     List[Any]				=	OpenAIGPTModel(config=lowercase_				)
     model.to(lowercase_				)
     model.eval()

     snake_case_ :     Any				=	model(lowercase_  ,     token_type_ids=lowercase_  ,     head_mask=lowercase_				)
     snake_case_ :     Optional[Any]				=	model(lowercase_  ,     token_type_ids=lowercase_				)
     snake_case_ :     Optional[Any]				=	model(lowercase_				)

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

 def 						_snake_case					(     self		:       Tuple  ,     lowercase_		:       Dict  ,     lowercase_		:       str  ,     lowercase_		:       Optional[Any]  ,     lowercase_		:       List[Any]  ,     *lowercase_		:       Optional[Any]				):
     snake_case_ :     Union[str, Any]				=	OpenAIGPTLMHeadModel(lowercase_				)
     model.to(lowercase_				)
     model.eval()

     snake_case_ :     Union[str, Any]				=	model(lowercase_  ,     token_type_ids=lowercase_  ,     labels=lowercase_				)
     self.parent.assertEqual(result.loss.shape  ,     ()				)
     self.parent.assertEqual(result.logits.shape  ,     (self.batch_size, self.seq_length, self.vocab_size)				)

 def 						_snake_case					(     self		:       List[str]  ,     lowercase_		:       Dict  ,     lowercase_		:       List[str]  ,     lowercase_		:       Any  ,     lowercase_		:       Dict  ,     *lowercase_		:       Union[str, Any]				):
     snake_case_ :     Tuple				=	OpenAIGPTDoubleHeadsModel(lowercase_				)
     model.to(lowercase_				)
     model.eval()

     snake_case_ :     Dict				=	model(lowercase_  ,     token_type_ids=lowercase_  ,     labels=lowercase_				)
     self.parent.assertEqual(result.loss.shape  ,     ()				)
     self.parent.assertEqual(result.logits.shape  ,     (self.batch_size, self.seq_length, self.vocab_size)				)

 def 						_snake_case					(     self		:       Any  ,     lowercase_		:       str  ,     lowercase_		:       List[str]  ,     lowercase_		:       Optional[Any]  ,     lowercase_		:       Optional[Any]  ,     *lowercase_		:       Any				):
     snake_case_ :     int				=	self.num_labels
     snake_case_ :     Any				=	OpenAIGPTForSequenceClassification(lowercase_				)
     model.to(lowercase_				)
     model.eval()

     snake_case_ :     Optional[int]				=	ids_tensor([self.batch_size]  ,     self.type_sequence_label_size				)
     snake_case_ :     Optional[Any]				=	model(lowercase_  ,     token_type_ids=lowercase_  ,     labels=lowercase_				)
     self.parent.assertEqual(result.logits.shape  ,     (self.batch_size, self.num_labels)				)






 def 						_snake_case					(     self		:       int				):
     snake_case_ :     Dict				=	self.prepare_config_and_inputs()
     (
         (
         snake_case_
     ),	(
         snake_case_
     ),	(
         snake_case_
     ),	(
         snake_case_
     ),	(
         snake_case_
     ),	(
         snake_case_
     ),	(
         snake_case_
     ),	
     ) :     str				=	config_and_inputs
     snake_case_ :     str				=	{
         '''input_ids''': input_ids,
         '''token_type_ids''': token_type_ids,
         '''head_mask''': head_mask,
     }

     return config, inputs_dict





@require_torch
class        _UpperCAmelCase       (			lowerCAmelCase__    ,     lowerCAmelCase__    ,     lowerCAmelCase__    ,     unittest.TestCase):
 _lowerCAmelCase   :			Dict							=		(
     (OpenAIGPTModel, OpenAIGPTLMHeadModel, OpenAIGPTDoubleHeadsModel, OpenAIGPTForSequenceClassification)
     if is_torch_available()
     else ()
 )
 _lowerCAmelCase   :			int							=		(
     (OpenAIGPTLMHeadModel,) if is_torch_available() else ()
 )  # TODO (PVP): Add Double HeadsModel when generate() function is changed accordingly
 _lowerCAmelCase   :			Union[str, Any]							=		(
     {
         """feature-extraction""": OpenAIGPTModel,
         """text-classification""": OpenAIGPTForSequenceClassification,
         """text-generation""": OpenAIGPTLMHeadModel,
         """zero-shot""": OpenAIGPTForSequenceClassification,
     }
     if is_torch_available()
     else {}
 )

 def 						_snake_case					(     self		:       Tuple  ,     lowercase_		:       Optional[int]  ,     lowercase_		:       int  ,     lowercase_		:       List[Any]  ,     lowercase_		:       List[Any]  ,     lowercase_		:       Union[str, Any]				):
     if pipeline_test_casse_name == "ZeroShotClassificationPipelineTests":
         # Get `tokenizer does not have a padding token` error for both fast/slow tokenizers.
         # `OpenAIGPTConfig` was never used in pipeline tests, either because of a missing checkpoint or because a
         # tiny config could not be created.
         return True

     return False

 def 						_snake_case					(     self		:       Optional[int]  ,     lowercase_		:       List[Any]  ,     lowercase_		:       Optional[int]  ,     lowercase_		:       List[str]=False				):
     snake_case_ :     Dict				=	super()._prepare_for_class(lowercase_  ,     lowercase_  ,     return_labels=lowercase_				)

     if return_labels:
         if model_class.__name__ == "OpenAIGPTDoubleHeadsModel":
             snake_case_ :     List[str]				=	torch.zeros(
                 (self.model_tester.batch_size, self.model_tester.num_choices, self.model_tester.seq_length)  ,     dtype=torch.long  ,     device=lowercase_  ,     )
             snake_case_ :     int				=	inputs_dict['''labels''']
             snake_case_ :     Optional[Any]				=	inputs_dict['''labels''']
             snake_case_ :     int				=	torch.zeros(
                 (self.model_tester.batch_size, self.model_tester.num_choices)  ,     dtype=torch.long  ,     device=lowercase_  ,     )
             snake_case_ :     Tuple				=	torch.zeros(
                 self.model_tester.batch_size  ,     dtype=torch.long  ,     device=lowercase_				)
     return inputs_dict

 def 						_snake_case					(     self		:       Any				):
     snake_case_ :     List[str]				=	OpenAIGPTModelTester(self				)
     snake_case_ :     Dict				=	ConfigTester(self  ,     config_class=lowercase_  ,     n_embd=37				)

 def 						_snake_case					(     self		:       List[str]				):
     self.config_tester.run_common_tests()

 def 						_snake_case					(     self		:       Optional[Any]				):
     snake_case_ :     Tuple				=	self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_openai_gpt_model(*lowercase_				)

 def 						_snake_case					(     self		:       List[str]				):
     snake_case_ :     List[str]				=	self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_lm_head_model(*lowercase_				)

 def 						_snake_case					(     self		:       int				):
     snake_case_ :     Any				=	self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_double_lm_head_model(*lowercase_				)

 def 						_snake_case					(     self		:       List[str]				):
     snake_case_ :     Optional[int]				=	self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_openai_gpt_for_sequence_classification(*lowercase_				)






 @slow
 def 						_snake_case					(     self		:       Dict				):
     for model_name in OPENAI_GPT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
         snake_case_ :     Optional[Any]				=	OpenAIGPTModel.from_pretrained(lowercase_				)
         self.assertIsNotNone(lowercase_				)





@require_torch
class        _UpperCAmelCase       (			unittest.TestCase):

 @slow
 def 						_snake_case					(     self		:       Optional[int]				):
     snake_case_ :     Optional[Any]				=	OpenAIGPTLMHeadModel.from_pretrained('''openai-gpt'''				)
     model.to(lowercase_				)
     snake_case_ :     List[str]				=	torch.tensor([[481, 4735, 544]]  ,     dtype=torch.long  ,     device=lowercase_				)  # the president is
     snake_case_ :     List[Any]				=	[
         481,
         4735,
         544,
         246,
         963,
         870,
         762,
         239,
         244,
         40477,
         244,
         249,
         719,
         881,
         487,
         544,
         240,
         244,
         603,
         481,
     ]  # the president is a very good man. " \n " i\'m sure he is, " said the

     snake_case_ :     Optional[Any]				=	model.generate(lowercase_  ,     do_sample=lowercase_				)
     self.assertListEqual(output_ids[0].tolist()  ,     lowercase_				)





