





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:
      __UpperCAmelCase	:       Optional[int]    	=							None


__UpperCAmelCase	:       Dict    	=							logging.get_logger(__name__)


__UpperCAmelCase	:       List[Any]    	=							{"vocab_file": "sentencepiece.bpe.model", "tokenizer_file": "tokenizer.json"}

__UpperCAmelCase	:       Tuple    	=							{
    "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",
    },
}

__UpperCAmelCase	:       List[str]    	=							{
    "facebook/mbart-large-en-ro": 1_024,
    "facebook/mbart-large-cc25": 1_024,
}

# fmt: off
__UpperCAmelCase	:       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    __snake_case	(							__lowerCamelCase      ):





       '''simple docstring'''







       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			:   Dict							,    A			:   Optional[int]=None							,    A			:   Optional[Any]=None							,    A			:   Tuple="<s>"							,    A			:   Tuple="</s>"							,    A			:   Optional[int]="</s>"							,    A			:   List[Any]="<s>"							,    A			:   Tuple="<unk>"							,    A			:   Tuple="<pad>"							,    A			:   List[str]="<mask>"							,    A			:   int=None							,    A			:   Tuple=None							,    A			:   Any=None							,    **A			:   Union[str, Any]							,    ):
         # Mask token behave like a normal word, i.e. include the space before it
         __snake_case:						Optional[int]			  =						AddedToken(A							,    lstrip=A							,    rstrip=A		) if isinstance(A							,    A		) else mask_token

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

         __snake_case:						Tuple			  =						vocab_file
         __snake_case:						Optional[Any]			  =						False if not self.vocab_file else True

         __snake_case:						Any			  =						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:						Tuple			  =						{
             lang_code: self.convert_tokens_to_ids(A		) for lang_code in FAIRSEQ_LANGUAGE_CODES
         }

         __snake_case:						Dict			  =						src_lang if src_lang is not None else """en_XX"""
         __snake_case:						List[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 							UpperCAmelCase__			(  self			:   Tuple		):
         return self._src_lang




       @src_lang.setter
       def 							UpperCAmelCase__			(  self			:   int							,    A			:   str		):
         __snake_case:						Dict			  =						new_src_lang
         self.set_src_lang_special_tokens(self._src_lang		)




       def 							UpperCAmelCase__			(  self			:   str							,    A			:   List[int]							,    A			:   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 							UpperCAmelCase__			(  self			:   Optional[int]							,    A			:   List[int]							,    A			:   Optional[List[int]] = None		):
         __snake_case:						Any			  =						[self.sep_token_id]
         __snake_case:						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 							UpperCAmelCase__			(  self			:   List[Any]							,    A			:   Optional[Any]							,    A			:   str							,    A			:   Optional[str]							,    A			:   Optional[str]							,    **A			:   Any		):
         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:						str			  =						src_lang
         __snake_case:						Optional[Any]			  =						self(A							,    add_special_tokens=A							,    return_tensors=A							,    **A		)
         __snake_case:						List[Any]			  =						self.convert_tokens_to_ids(A		)
         __snake_case:						str			  =						tgt_lang_id
         return inputs




       def 							UpperCAmelCase__			(  self			:   Optional[int]							,    A			:   List[str]							,    A			:   str = "en_XX"							,    A			:   Optional[List[str]] = None							,    A			:   str = "ro_RO"							,    **A			:   Tuple							,    ):
         __snake_case:						str			  =						src_lang
         __snake_case:						List[str]			  =						tgt_lang
         return super().prepare_seqaseq_batch(A							,    A							,    **A		)




       def 							UpperCAmelCase__			(  self			:   Union[str, Any]		):
         return self.set_src_lang_special_tokens(self.src_lang		)




       def 							UpperCAmelCase__			(  self			:   List[Any]		):
         return self.set_tgt_lang_special_tokens(self.tgt_lang		)




       def 							UpperCAmelCase__			(  self			:   str							,    A			:   List[str]		):
         __snake_case:						int			  =						self.convert_tokens_to_ids(A		)
         __snake_case:						List[Any]			  =						[]
         __snake_case:						str			  =						[self.eos_token_id, self.cur_lang_code]

         __snake_case:						Any			  =						self.convert_ids_to_tokens(self.prefix_tokens		)
         __snake_case:						Dict			  =						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 							UpperCAmelCase__			(  self			:   Dict							,    A			:   str		):
         __snake_case:						Any			  =						self.convert_tokens_to_ids(A		)
         __snake_case:						List[str]			  =						[]
         __snake_case:						Optional[int]			  =						[self.eos_token_id, self.cur_lang_code]

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

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

         return (out_vocab_file,)













import copy
import tempfile
import unittest

from huggingface_hub import HfFolder, delete_repo
from parameterized import parameterized
from requests.exceptions import HTTPError

from transformers import AutoConfig, GenerationConfig
from transformers.testing_utils import TOKEN, USER, is_staging_test







class    __snake_case	(							unittest.TestCase      ):





       '''simple docstring'''







       @parameterized.expand([(None,), ("""foo.json""",)]		)
       def 							UpperCAmelCase__			(  self			:   List[str]							,    A			:   Optional[Any]		):
         __snake_case:						Any			  =						GenerationConfig(
             do_sample=A							,    temperature=0.7							,    length_penalty=1.0							,    bad_words_ids=[[1, 2, 3], [4, 5]]							,    )
         with tempfile.TemporaryDirectory() as tmp_dir:
           config.save_pretrained(A							,    config_name=A		)
           __snake_case:						Optional[int]			  =						GenerationConfig.from_pretrained(A							,    config_name=A		)

         # Checks parameters that were specified
         self.assertEqual(loaded_config.do_sample							,    A		)
         self.assertEqual(loaded_config.temperature							,    0.7		)
         self.assertEqual(loaded_config.length_penalty							,    1.0		)
         self.assertEqual(loaded_config.bad_words_ids							,    [[1, 2, 3], [4, 5]]		)

         # Checks parameters that were not specified (defaults)
         self.assertEqual(loaded_config.top_k							,    50		)
         self.assertEqual(loaded_config.max_length							,    20		)
         self.assertEqual(loaded_config.max_time							,    A		)




       def 							UpperCAmelCase__			(  self			:   Dict		):
         __snake_case:						str			  =						AutoConfig.from_pretrained("""gpt2"""		)
         __snake_case:						Any			  =						GenerationConfig.from_model_config(A		)
         __snake_case:						str			  =						GenerationConfig()

         # The generation config has loaded a few non-default parameters from the model config
         self.assertNotEqual(A							,    A		)

         # One of those parameters is eos_token_id -- check if it matches
         self.assertNotEqual(generation_config_from_model.eos_token_id							,    default_generation_config.eos_token_id		)
         self.assertEqual(generation_config_from_model.eos_token_id							,    model_config.eos_token_id		)




       def 							UpperCAmelCase__			(  self			:   str		):
         __snake_case:						List[str]			  =						GenerationConfig()
         __snake_case:						Tuple			  =						{
             """max_new_tokens""": 1_024,
             """foo""": """bar""",
         }
         __snake_case:						List[str]			  =						copy.deepcopy(A		)
         __snake_case:						Optional[int]			  =						generation_config.update(**A		)

         # update_kwargs was not modified (no side effects)
         self.assertEqual(A							,    A		)

         # update_kwargs was used to update the config on valid attributes
         self.assertEqual(generation_config.max_new_tokens							,    1_024		)

         # `.update()` returns a dictionary of unused kwargs
         self.assertEqual(A							,    {"""foo""": """bar"""}		)




       def 							UpperCAmelCase__			(  self			:   Tuple		):
         __snake_case:						List[str]			  =						GenerationConfig()
         __snake_case:						Optional[int]			  =						"""bar"""

         with tempfile.TemporaryDirectory("""test-generation-config"""		) as tmp_dir:
           generation_config.save_pretrained(A		)

           __snake_case:						Any			  =						GenerationConfig.from_pretrained(A		)
         # update_kwargs was used to update the config on valid attributes
         self.assertEqual(new_config.foo							,    """bar"""		)

         __snake_case:						int			  =						GenerationConfig.from_model_config(A		)
         assert not hasattr(A							,    """foo"""		)  # no new kwargs should be initialized if from config






       def 							UpperCAmelCase__			(  self			:   Dict		):
         __snake_case:						Dict			  =						GenerationConfig()
         self.assertEqual(default_config.temperature							,    1.0		)
         self.assertEqual(default_config.do_sample							,    A		)
         self.assertEqual(default_config.num_beams							,    1		)

         __snake_case:						Union[str, Any]			  =						GenerationConfig(
             do_sample=A							,    temperature=0.7							,    length_penalty=1.0							,    bad_words_ids=[[1, 2, 3], [4, 5]]							,    )
         self.assertEqual(config.temperature							,    0.7		)
         self.assertEqual(config.do_sample							,    A		)
         self.assertEqual(config.num_beams							,    1		)

         with tempfile.TemporaryDirectory() as tmp_dir:
           config.save_pretrained(A		)
           __snake_case:						Tuple			  =						GenerationConfig.from_pretrained(A							,    temperature=1.0		)

         self.assertEqual(loaded_config.temperature							,    1.0		)
         self.assertEqual(loaded_config.do_sample							,    A		)
         self.assertEqual(loaded_config.num_beams							,    1		)  # default value







@is_staging_test
class    __snake_case	(							unittest.TestCase      ):





       '''simple docstring'''







       @classmethod
       def 							UpperCAmelCase__			(  cls			:   List[str]		):
         __snake_case:						Optional[int]			  =						TOKEN
         HfFolder.save_token(A		)




       @classmethod
       def 							UpperCAmelCase__			(  cls			:   List[Any]		):
         try:
           delete_repo(token=cls._token							,    repo_id="""test-generation-config"""		)
         except HTTPError:
           pass

         try:
           delete_repo(token=cls._token							,    repo_id="""valid_org/test-generation-config-org"""		)
         except HTTPError:
           pass




       def 							UpperCAmelCase__			(  self			:   Tuple		):
         __snake_case:						Optional[int]			  =						GenerationConfig(
             do_sample=A							,    temperature=0.7							,    length_penalty=1.0							,    )
         config.push_to_hub("""test-generation-config"""							,    use_auth_token=self._token		)

         __snake_case:						str			  =						GenerationConfig.from_pretrained(f'''{USER}/test-generation-config'''		)
         for k, v in config.to_dict().items():
           if k != "transformers_version":
             self.assertEqual(A							,    getattr(A							,    A		)		)

        # Reset repo
         delete_repo(token=self._token							,    repo_id="""test-generation-config"""		)

         # Push to hub via save_pretrained
         with tempfile.TemporaryDirectory() as tmp_dir:
           config.save_pretrained(
               A							,    repo_id="""test-generation-config"""							,    push_to_hub=A							,    use_auth_token=self._token		)

         __snake_case:						Optional[Any]			  =						GenerationConfig.from_pretrained(f'''{USER}/test-generation-config'''		)
         for k, v in config.to_dict().items():
           if k != "transformers_version":
             self.assertEqual(A							,    getattr(A							,    A		)		)






       def 							UpperCAmelCase__			(  self			:   List[Any]		):
         __snake_case:						Union[str, Any]			  =						GenerationConfig(
             do_sample=A							,    temperature=0.7							,    length_penalty=1.0							,    )
         config.push_to_hub("""valid_org/test-generation-config-org"""							,    use_auth_token=self._token		)

         __snake_case:						int			  =						GenerationConfig.from_pretrained("""valid_org/test-generation-config-org"""		)
         for k, v in config.to_dict().items():
           if k != "transformers_version":
             self.assertEqual(A							,    getattr(A							,    A		)		)

        # Reset repo
         delete_repo(token=self._token							,    repo_id="""valid_org/test-generation-config-org"""		)

         # Push to hub via save_pretrained
         with tempfile.TemporaryDirectory() as tmp_dir:
           config.save_pretrained(
               A							,    repo_id="""valid_org/test-generation-config-org"""							,    push_to_hub=A							,    use_auth_token=self._token		)

         __snake_case:						Optional[int]			  =						GenerationConfig.from_pretrained("""valid_org/test-generation-config-org"""		)
         for k, v in config.to_dict().items():
           if k != "transformers_version":
             self.assertEqual(A							,    getattr(A							,    A		)		)






