



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


__lowerCAmelCase							=logging.get_logger(__name__)

__lowerCAmelCase							={"vocab_file": "sentencepiece.bpe.model", "tokenizer_file": "tokenizer.json"}

__lowerCAmelCase							={
    "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"
        ),
    },
}

__lowerCAmelCase							={
    "facebook/mbart-large-50-one-to-many-mmt": 1024,
}

# fmt: off
__lowerCAmelCase							=["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    _snake_case   (				snake_case					):

					"""simple docstring"""





					_UpperCamelCase		 =      VOCAB_FILES_NAMES
					_UpperCamelCase		 =      PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
					_UpperCamelCase		 =      PRETRAINED_VOCAB_FILES_MAP
					_UpperCamelCase		 =      ["input_ids", "attention_mask"]
					_UpperCamelCase		 =      MBartaaTokenizer

					_UpperCamelCase		 =      []
					_UpperCamelCase		 =      []




					def __init__(     self   , UpperCAmelCase__=None   , UpperCAmelCase__=None   , UpperCAmelCase__=None   , UpperCAmelCase__=None   , UpperCAmelCase__="</s>"   , UpperCAmelCase__="</s>"   , UpperCAmelCase__="<s>"   , UpperCAmelCase__="<unk>"   , UpperCAmelCase__="<pad>"   , UpperCAmelCase__="<mask>"   , **UpperCAmelCase__   , )      ->	Optional[Any]:
										# Mask token behave like a normal word, i.e. include the space before it
										a_									=   AddedToken(UpperCAmelCase__   , lstrip=UpperCAmelCase__   , rstrip=UpperCAmelCase__     ) if isinstance(UpperCAmelCase__   , UpperCAmelCase__     ) 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__(
										    UpperCAmelCase__   , src_lang=UpperCAmelCase__   , tgt_lang=UpperCAmelCase__   , tokenizer_file=UpperCAmelCase__   , eos_token=UpperCAmelCase__   , sep_token=UpperCAmelCase__   , cls_token=UpperCAmelCase__   , unk_token=UpperCAmelCase__   , pad_token=UpperCAmelCase__   , mask_token=UpperCAmelCase__   , **UpperCAmelCase__   , )

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

										a_									=   {
										    lang_code: self.convert_tokens_to_ids(UpperCAmelCase__     ) 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  __SCREAMING_SNAKE_CASE (     self     )      ->	str:
										return self._src_lang




					@src_lang.setter
					def  __SCREAMING_SNAKE_CASE (     self   , UpperCAmelCase__     )      ->	None:
										a_									=   new_src_lang
										self.set_src_lang_special_tokens(self._src_lang     )




					def  __SCREAMING_SNAKE_CASE (     self   , UpperCAmelCase__   , UpperCAmelCase__ = None     )      ->	List[int]:
										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  __SCREAMING_SNAKE_CASE (     self   , UpperCAmelCase__   , UpperCAmelCase__ = "en_XX"   , UpperCAmelCase__ = None   , UpperCAmelCase__ = "ro_RO"   , **UpperCAmelCase__   , )      ->	BatchEncoding:
										a_									=   src_lang
										a_									=   tgt_lang
										return super().prepare_seqaseq_batch(UpperCAmelCase__   , UpperCAmelCase__   , **UpperCAmelCase__     )




					def  __SCREAMING_SNAKE_CASE (     self     )      ->	List[str]:
										return self.set_src_lang_special_tokens(self.src_lang     )




					def  __SCREAMING_SNAKE_CASE (     self     )      ->	Optional[Any]:
										return self.set_tgt_lang_special_tokens(self.tgt_lang     )




					def  __SCREAMING_SNAKE_CASE (     self   , UpperCAmelCase__     )      ->	None:
										a_									=   self.convert_tokens_to_ids(UpperCAmelCase__     )
										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  __SCREAMING_SNAKE_CASE (     self   , UpperCAmelCase__     )      ->	None:
										a_									=   self.convert_tokens_to_ids(UpperCAmelCase__     )
										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  __SCREAMING_SNAKE_CASE (     self   , UpperCAmelCase__   , UpperCAmelCase__   , UpperCAmelCase__   , UpperCAmelCase__   , **UpperCAmelCase__     )      ->	List[str]:
										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(UpperCAmelCase__   , add_special_tokens=UpperCAmelCase__   , return_tensors=UpperCAmelCase__   , **UpperCAmelCase__     )
										a_									=   self.convert_tokens_to_ids(UpperCAmelCase__     )
										a_									=   tgt_lang_id
										return inputs




					def  __SCREAMING_SNAKE_CASE (     self   , UpperCAmelCase__   , UpperCAmelCase__ = None     )      ->	Tuple[str]:
										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(UpperCAmelCase__     ):
															logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''     )
															return
										a_									=   os.path.join(
										    UpperCAmelCase__   , (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']     )

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

										return (out_vocab_file,)









'''simple docstring'''
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 ConvNextImageProcessor


class    _snake_case   (				unittest.TestCase					):

					"""simple docstring"""





					def __init__(     self   , UpperCAmelCase__   , UpperCAmelCase__=7   , UpperCAmelCase__=3   , UpperCAmelCase__=18   , UpperCAmelCase__=30   , UpperCAmelCase__=400   , UpperCAmelCase__=True   , UpperCAmelCase__=None   , UpperCAmelCase__=0.8_7_5   , UpperCAmelCase__=True   , UpperCAmelCase__=[0.5, 0.5, 0.5]   , UpperCAmelCase__=[0.5, 0.5, 0.5]   , )      ->	Optional[int]:
										a_									=   size if size is not None else {'shortest_edge': 20}
										a_									=   parent
										a_									=   batch_size
										a_									=   num_channels
										a_									=   image_size
										a_									=   min_resolution
										a_									=   max_resolution
										a_									=   do_resize
										a_									=   size
										a_									=   crop_pct
										a_									=   do_normalize
										a_									=   image_mean
										a_									=   image_std




					def  __SCREAMING_SNAKE_CASE (     self     )      ->	List[str]:
										return {
										    "image_mean": self.image_mean,
										    "image_std": self.image_std,
										    "do_normalize": self.do_normalize,
										    "do_resize": self.do_resize,
										    "size": self.size,
										    "crop_pct": self.crop_pct,
										}



@require_torch
@require_vision
class    _snake_case   (				snake_case     ,					unittest.TestCase					):

					"""simple docstring"""





					_UpperCamelCase		 =      ConvNextImageProcessor if is_vision_available() else None




					def  __SCREAMING_SNAKE_CASE (     self     )      ->	Tuple:
										a_									=   ConvNextImageProcessingTester(self     )




					@property
					def  __SCREAMING_SNAKE_CASE (     self     )      ->	Dict:
										return self.image_processor_tester.prepare_image_processor_dict()




					def  __SCREAMING_SNAKE_CASE (     self     )      ->	Any:
										a_									=   self.image_processing_class(**self.image_processor_dict     )
										self.assertTrue(hasattr(UpperCAmelCase__   , 'do_resize'     )     )
										self.assertTrue(hasattr(UpperCAmelCase__   , 'size'     )     )
										self.assertTrue(hasattr(UpperCAmelCase__   , 'crop_pct'     )     )
										self.assertTrue(hasattr(UpperCAmelCase__   , 'do_normalize'     )     )
										self.assertTrue(hasattr(UpperCAmelCase__   , 'image_mean'     )     )
										self.assertTrue(hasattr(UpperCAmelCase__   , 'image_std'     )     )




					def  __SCREAMING_SNAKE_CASE (     self     )      ->	Tuple:
										a_									=   self.image_processing_class.from_dict(self.image_processor_dict     )
										self.assertEqual(image_processor.size   , {'shortest_edge': 20}     )

										a_									=   self.image_processing_class.from_dict(self.image_processor_dict   , size=42     )
										self.assertEqual(image_processor.size   , {'shortest_edge': 42}     )




					def  __SCREAMING_SNAKE_CASE (     self     )      ->	Optional[Any]:
										pass




					def  __SCREAMING_SNAKE_CASE (     self     )      ->	Tuple:
										# Initialize image_processing
										a_									=   self.image_processing_class(**self.image_processor_dict     )
										# create random PIL images
										a_									=   prepare_image_inputs(self.image_processor_tester   , equal_resolution=UpperCAmelCase__     )
										for image in image_inputs:
															self.assertIsInstance(UpperCAmelCase__   , Image.Image     )

										# Test not batched input
										a_									=   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['shortest_edge'],
										        self.image_processor_tester.size['shortest_edge'],
										    )   , )

										# Test batched
										a_									=   image_processing(UpperCAmelCase__   , 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['shortest_edge'],
										        self.image_processor_tester.size['shortest_edge'],
										    )   , )




					def  __SCREAMING_SNAKE_CASE (     self     )      ->	int:
										# Initialize image_processing
										a_									=   self.image_processing_class(**self.image_processor_dict     )
										# create random numpy tensors
										a_									=   prepare_image_inputs(self.image_processor_tester   , equal_resolution=UpperCAmelCase__   , numpify=UpperCAmelCase__     )
										for image in image_inputs:
															self.assertIsInstance(UpperCAmelCase__   , np.ndarray     )

										# Test not batched input
										a_									=   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['shortest_edge'],
										        self.image_processor_tester.size['shortest_edge'],
										    )   , )

										# Test batched
										a_									=   image_processing(UpperCAmelCase__   , 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['shortest_edge'],
										        self.image_processor_tester.size['shortest_edge'],
										    )   , )




					def  __SCREAMING_SNAKE_CASE (     self     )      ->	int:
										# Initialize image_processing
										a_									=   self.image_processing_class(**self.image_processor_dict     )
										# create random PyTorch tensors
										a_									=   prepare_image_inputs(self.image_processor_tester   , equal_resolution=UpperCAmelCase__   , torchify=UpperCAmelCase__     )
										for image in image_inputs:
															self.assertIsInstance(UpperCAmelCase__   , torch.Tensor     )

										# Test not batched input
										a_									=   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['shortest_edge'],
										        self.image_processor_tester.size['shortest_edge'],
										    )   , )

										# Test batched
										a_									=   image_processing(UpperCAmelCase__   , 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['shortest_edge'],
										        self.image_processor_tester.size['shortest_edge'],
										    )   , )

