




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_mbartaa import MBartaaTokenizer
else:
  UpperCamelCase__       				=	None


UpperCamelCase__       				=	logging.get_logger(__name__)

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

UpperCamelCase__       				=	{
    '''vocab_file''': {
        '''facebook/mbart-large-50-one-to-many-mmt''': (
            '''https://huggingface.co/facebook/mbart-large-50-one-to-many-mmt/resolve/main/sentencepiece.bpe.model'''
        ),
    },
    '''tokenizer_file''': {
        '''facebook/mbart-large-50-one-to-many-mmt''': (
            '''https://huggingface.co/facebook/mbart-large-50-one-to-many-mmt/resolve/main/tokenizer.json'''
        ),
    },
}

UpperCamelCase__       				=	{
    '''facebook/mbart-large-50-one-to-many-mmt''': 1_024,
}

# fmt: off
UpperCamelCase__       				=	['''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''', '''af_ZA''', '''az_AZ''', '''bn_IN''', '''fa_IR''', '''he_IL''', '''hr_HR''', '''id_ID''', '''ka_GE''', '''km_KH''', '''mk_MK''', '''ml_IN''', '''mn_MN''', '''mr_IN''', '''pl_PL''', '''ps_AF''', '''pt_XX''', '''sv_SE''', '''sw_KE''', '''ta_IN''', '''te_IN''', '''th_TH''', '''tl_XX''', '''uk_UA''', '''ur_PK''', '''xh_ZA''', '''gl_ES''', '''sl_SI''']
class 							__lowercase   (			lowerCamelCase__    ):
     _lowerCAmelCase		       =						VOCAB_FILES_NAMES
     _lowerCAmelCase		       =						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     _lowerCAmelCase		       =						PRETRAINED_VOCAB_FILES_MAP
     _lowerCAmelCase		       =						["""input_ids""", """attention_mask"""]
     _lowerCAmelCase		       =						MBartaaTokenizer

     _lowerCAmelCase		       =						[]
     _lowerCAmelCase		       =						[]



     def __init__(  self   :							List[str]      ,		lowercase__   :							int=None      ,		lowercase__   :							Optional[int]=None      ,		lowercase__   :							List[Any]=None      ,		lowercase__   :							Optional[Any]=None      ,		lowercase__   :							Tuple="</s>"      ,		lowercase__   :							Any="</s>"      ,		lowercase__   :							List[Any]="<s>"      ,		lowercase__   :							List[Any]="<unk>"      ,		lowercase__   :							str="<pad>"      ,		lowercase__   :							Union[str, Any]="<mask>"      ,		**lowercase__   :							int      ,		):
       a_		    =  AddedToken(snake_case__      ,		lstrip=snake_case__      ,		rstrip=snake_case__       ) if isinstance(snake_case__      ,		snake_case__       ) else mask_token

       a_		    =  kwargs.get('''additional_special_tokens'''      ,		[]       )
       kwargs["additional_special_tokens"] += [
           code for code in FAIRSEQ_LANGUAGE_CODES if code not in kwargs["additional_special_tokens"]
       ]

       super().__init__(
           snake_case__      ,		src_lang=snake_case__      ,		tgt_lang=snake_case__      ,		tokenizer_file=snake_case__      ,		eos_token=snake_case__      ,		sep_token=snake_case__      ,		cls_token=snake_case__      ,		unk_token=snake_case__      ,		pad_token=snake_case__      ,		mask_token=snake_case__      ,		**snake_case__      ,		)

       a_		    =  vocab_file
       a_		    =  False if not self.vocab_file else True

       a_		    =  {
           lang_code: self.convert_tokens_to_ids(snake_case__       ) for lang_code in FAIRSEQ_LANGUAGE_CODES
       }

       a_		    =  src_lang if src_lang is not None else '''en_XX'''
       a_		    =  tgt_lang
       a_		    =  self.lang_code_to_id[self._src_lang]
       self.set_src_lang_special_tokens(self._src_lang       )



     @property
     def 				__magic_name__		(  self   :							Any       ):
       return self._src_lang



     @src_lang.setter
     def 				__magic_name__		(  self   :							List[str]      ,		lowercase__   :							List[str]       ):
       a_		    =  new_src_lang
       self.set_src_lang_special_tokens(self._src_lang       )



     def 				__magic_name__		(  self   :							Tuple      ,		lowercase__   :							Any      ,		lowercase__   :							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 				__magic_name__		(  self   :							Optional[Any]      ,		lowercase__   :							List[str]      ,		lowercase__   :							Any = "en_XX"      ,		lowercase__   :							List[str] = None      ,		lowercase__   :							Dict = "ro_RO"      ,		**lowercase__   :							str      ,		):
       a_		    =  src_lang
       a_		    =  tgt_lang
       return super().prepare_seqaseq_batch(snake_case__      ,		snake_case__      ,		**snake_case__       )



     def 				__magic_name__		(  self   :							Tuple       ):
       return self.set_src_lang_special_tokens(self.src_lang       )



     def 				__magic_name__		(  self   :							int       ):
       return self.set_tgt_lang_special_tokens(self.tgt_lang       )



     def 				__magic_name__		(  self   :							Optional[int]      ,		lowercase__   :							Union[str, Any]       ):
       a_		    =  self.convert_tokens_to_ids(snake_case__       )
       a_		    =  [self.cur_lang_code_id]
       a_		    =  [self.eos_token_id]

       a_		    =  self.convert_ids_to_tokens(self.prefix_tokens       )
       a_		    =  self.convert_ids_to_tokens(self.suffix_tokens       )

       a_		    =  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 				__magic_name__		(  self   :							str      ,		lowercase__   :							List[str]       ):
       a_		    =  self.convert_tokens_to_ids(snake_case__       )
       a_		    =  [self.cur_lang_code_id]
       a_		    =  [self.eos_token_id]

       a_		    =  self.convert_ids_to_tokens(self.prefix_tokens       )
       a_		    =  self.convert_ids_to_tokens(self.suffix_tokens       )

       a_		    =  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 				__magic_name__		(  self   :							Tuple      ,		lowercase__   :							Optional[Any]      ,		lowercase__   :							Optional[int]      ,		lowercase__   :							Any      ,		lowercase__   :							Optional[int]      ,		**lowercase__   :							Union[str, 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'''       )
       a_		    =  src_lang
       a_		    =  self(snake_case__      ,		add_special_tokens=snake_case__      ,		return_tensors=snake_case__      ,		**snake_case__       )
       a_		    =  self.convert_tokens_to_ids(snake_case__       )
       a_		    =  tgt_lang_id
       return inputs



     def 				__magic_name__		(  self   :							Dict      ,		lowercase__   :							str      ,		lowercase__   :							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(snake_case__       ):
         logger.error(f"Vocabulary path ({save_directory}) should be a directory"       )
         return
       a_		    =  os.path.join(
           snake_case__      ,		(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']       )

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

       return (out_vocab_file,)

import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
     import torch

if is_vision_available():
     from PIL import Image

     from transformers import EfficientNetImageProcessor







class       __snake_case (  unittest.TestCase ):
 def __init__(	self		, snake_case__		, snake_case__=13		, snake_case__=3		, snake_case__=18		, snake_case__=30		, snake_case__=400		, snake_case__=True		, snake_case__=None		, snake_case__=True		, snake_case__=[0.5, 0.5, 0.5]		, snake_case__=[0.5, 0.5, 0.5]		, )						->				Any:
       '''simple docstring'''





       UpperCAmelCase     :			int     	=size if size is not None else {'''height''': 18, '''width''': 18}
       UpperCAmelCase     :			Tuple     	=parent
       UpperCAmelCase     :			List[str]     	=batch_size
       UpperCAmelCase     :			int     	=num_channels
       UpperCAmelCase     :			List[str]     	=image_size
       UpperCAmelCase     :			str     	=min_resolution
       UpperCAmelCase     :			Optional[int]     	=max_resolution
       UpperCAmelCase     :			Optional[int]     	=do_resize
       UpperCAmelCase     :			int     	=size
       UpperCAmelCase     :			Union[str, Any]     	=do_normalize
       UpperCAmelCase     :			Dict     	=image_mean
       UpperCAmelCase     :			List[Any]     	=image_std






 def 						UpperCAmelCase__							(	self    )						->				int:
       '''simple docstring'''





       return {
           "image_mean": self.image_mean,
           "image_std": self.image_std,
           "do_normalize": self.do_normalize,
           "do_resize": self.do_resize,
           "size": self.size,
       }







@require_torch
@require_vision
class       __snake_case (  lowerCamelCase__       , unittest.TestCase ):
 __lowerCamelCase     : Any 	=   EfficientNetImageProcessor if is_vision_available() else None
 def 						UpperCAmelCase__							(	self    )						->				Tuple:
       '''simple docstring'''





       UpperCAmelCase     :			Dict     	=EfficientNetImageProcessorTester(self    )
 @property
 def 						UpperCAmelCase__							(	self    )						->				int:
       '''simple docstring'''





       return self.image_processor_tester.prepare_image_processor_dict()
 def 						UpperCAmelCase__							(	self    )						->				List[str]:
       '''simple docstring'''





       UpperCAmelCase     :			Tuple     	=self.image_processing_class(**self.image_processor_dict    )
       self.assertTrue(hasattr(snake_case__		, '''image_mean'''    )    )
       self.assertTrue(hasattr(snake_case__		, '''image_std'''    )    )
       self.assertTrue(hasattr(snake_case__		, '''do_normalize'''    )    )
       self.assertTrue(hasattr(snake_case__		, '''do_resize'''    )    )
       self.assertTrue(hasattr(snake_case__		, '''size'''    )    )
 def 						UpperCAmelCase__							(	self    )						->				Tuple:
       '''simple docstring'''





       UpperCAmelCase     :			str     	=self.image_processing_class.from_dict(self.image_processor_dict    )
       self.assertEqual(image_processor.size		, {'''height''': 18, '''width''': 18}    )

       UpperCAmelCase     :			List[str]     	=self.image_processing_class.from_dict(self.image_processor_dict		, size=42    )
       self.assertEqual(image_processor.size		, {'''height''': 42, '''width''': 42}    )
 def 						UpperCAmelCase__							(	self    )						->				Dict:
       '''simple docstring'''





       UpperCAmelCase     :			List[Any]     	=self.image_processing_class(**self.image_processor_dict    )
       # create random PIL images
       UpperCAmelCase     :			Optional[Any]     	=prepare_image_inputs(self.image_processor_tester		, equal_resolution=snake_case__    )
       for image in image_inputs:
             self.assertIsInstance(snake_case__		, Image.Image    )

       # Test not batched input
       UpperCAmelCase     :			int     	=image_processing(image_inputs[0]		, return_tensors='''pt'''    ).pixel_values
       self.assertEqual(
           encoded_images.shape		, (
               1,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.size['''height'''],
               self.image_processor_tester.size['''width'''],
           )		, )

       # Test batched
       UpperCAmelCase     :			int     	=image_processing(snake_case__		, return_tensors='''pt'''    ).pixel_values
       self.assertEqual(
           encoded_images.shape		, (
               self.image_processor_tester.batch_size,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.size['''height'''],
               self.image_processor_tester.size['''width'''],
           )		, )
 def 						UpperCAmelCase__							(	self    )						->				Dict:
       '''simple docstring'''





       UpperCAmelCase     :			Tuple     	=self.image_processing_class(**self.image_processor_dict    )
       # create random numpy tensors
       UpperCAmelCase     :			List[str]     	=prepare_image_inputs(self.image_processor_tester		, equal_resolution=snake_case__		, numpify=snake_case__    )
       for image in image_inputs:
             self.assertIsInstance(snake_case__		, np.ndarray    )

       # Test not batched input
       UpperCAmelCase     :			str     	=image_processing(image_inputs[0]		, return_tensors='''pt'''    ).pixel_values
       self.assertEqual(
           encoded_images.shape		, (
               1,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.size['''height'''],
               self.image_processor_tester.size['''width'''],
           )		, )

       # Test batched
       UpperCAmelCase     :			Union[str, Any]     	=image_processing(snake_case__		, return_tensors='''pt'''    ).pixel_values
       self.assertEqual(
           encoded_images.shape		, (
               self.image_processor_tester.batch_size,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.size['''height'''],
               self.image_processor_tester.size['''width'''],
           )		, )






 def 						UpperCAmelCase__							(	self    )						->				str:
       '''simple docstring'''





       UpperCAmelCase     :			int     	=self.image_processing_class(**self.image_processor_dict    )
       # create random PyTorch tensors
       UpperCAmelCase     :			int     	=prepare_image_inputs(self.image_processor_tester		, equal_resolution=snake_case__		, torchify=snake_case__    )
       for image in image_inputs:
             self.assertIsInstance(snake_case__		, torch.Tensor    )

       # Test not batched input
       UpperCAmelCase     :			List[Any]     	=image_processing(image_inputs[0]		, return_tensors='''pt'''    ).pixel_values
       self.assertEqual(
           encoded_images.shape		, (
               1,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.size['''height'''],
               self.image_processor_tester.size['''width'''],
           )		, )

       # Test batched
       UpperCAmelCase     :			List[str]     	=image_processing(snake_case__		, return_tensors='''pt'''    ).pixel_values
       self.assertEqual(
           encoded_images.shape		, (
               self.image_processor_tester.batch_size,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.size['''height'''],
               self.image_processor_tester.size['''width'''],
           )		, )
