




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

import sentencepiece as spm

from ...tokenization_utils import AddedToken, BatchEncoding, PreTrainedTokenizer
from ...utils import logging


__SCREAMING_SNAKE_CASE 						=logging.get_logger(__name__)

__SCREAMING_SNAKE_CASE 						="""▁"""

__SCREAMING_SNAKE_CASE 						={"""vocab_file""": """sentencepiece.bpe.model"""}

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

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

# fmt: off
__SCREAMING_SNAKE_CASE 						=["""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  __magic_name__						(  __UpperCAmelCase):
				'''simple docstring'''







				SCREAMING_SNAKE_CASE__					:							Union[str, Any]	      =					VOCAB_FILES_NAMES
				SCREAMING_SNAKE_CASE__					:							Optional[Any]	      =					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				SCREAMING_SNAKE_CASE__					:							Tuple	      =					PRETRAINED_VOCAB_FILES_MAP
				SCREAMING_SNAKE_CASE__					:							Union[str, Any]	      =					["input_ids", "attention_mask"]

				SCREAMING_SNAKE_CASE__					:							List[int]	      =					[]
				SCREAMING_SNAKE_CASE__					:							List[int]	      =					[]


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

											SCREAMING_SNAKE_CASE_   	=    {} if sp_model_kwargs is None else sp_model_kwargs

											SCREAMING_SNAKE_CASE_   	=    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__(
											    src_lang=_lowerCamelCase		,					tgt_lang=_lowerCamelCase		,					eos_token=_lowerCamelCase		,					unk_token=_lowerCamelCase		,					sep_token=_lowerCamelCase		,					cls_token=_lowerCamelCase		,					pad_token=_lowerCamelCase		,					mask_token=_lowerCamelCase		,					sp_model_kwargs=self.sp_model_kwargs		,					**_lowerCamelCase		,					)

											SCREAMING_SNAKE_CASE_   	=    spm.SentencePieceProcessor(**self.sp_model_kwargs		)
											self.sp_model.Load(str(_lowerCamelCase		)		)
											SCREAMING_SNAKE_CASE_   	=    vocab_file

											# Original fairseq vocab and spm vocab must be "aligned":
											# Vocab    |    0    |    1    |   2    |    3    |  4  |  5  |  6  |   7   |   8   |  9
											# -------- | ------- | ------- | ------ | ------- | --- | --- | --- | ----- | ----- | ----
											# fairseq  | '<s>'   | '<pad>' | '</s>' | '<unk>' | ',' | '.' | '▁' | 's'   | '▁de' | '-'
											# spm      | '<unk>' | '<s>'   | '</s>' | ','     | '.' | '▁' | 's' | '▁de' | '-'   | '▁a'

											# Mimic fairseq token-to-id alignment for the first 4 token
											SCREAMING_SNAKE_CASE_   	=    {'''<s>''': 0, '''<pad>''': 1, '''</s>''': 2, '''<unk>''': 3}

											# The first "real" token "," has position 4 in the original fairseq vocab and position 3 in the spm vocab
											SCREAMING_SNAKE_CASE_   	=    1

											SCREAMING_SNAKE_CASE_   	=    len(self.sp_model		)
											SCREAMING_SNAKE_CASE_   	=    {
											    code: self.sp_model_size + i + self.fairseq_offset for i, code in enumerate(_lowerCamelCase		)
											}
											SCREAMING_SNAKE_CASE_   	=    {v: k for k, v in self.lang_code_to_id.items()}
											SCREAMING_SNAKE_CASE_   	=    len(self.sp_model		) + len(self.lang_code_to_id		) + self.fairseq_offset

											self.fairseq_tokens_to_ids.update(self.lang_code_to_id		)
											SCREAMING_SNAKE_CASE_   	=    {v: k for k, v in self.fairseq_tokens_to_ids.items()}

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


				@property
				def     _A							(							self:   Any		):
											return len(self.sp_model		) + len(self.lang_code_to_id		) + self.fairseq_offset + 1  # Plus 1 for the mask token


				@property
				def     _A							(							self:   int		):
											return self._src_lang


				@src_lang.setter
				def     _A							(							self:   List[str]		,					_lowerCamelCase:   str		):
											SCREAMING_SNAKE_CASE_   	=    new_src_lang
											self.set_src_lang_special_tokens(self._src_lang		)


				def __getstate__(							self:   Optional[Any]		):
											SCREAMING_SNAKE_CASE_   	=    self.__dict__.copy()
											SCREAMING_SNAKE_CASE_   	=    None
											return state


				def __setstate__(							self:   Dict		,					_lowerCamelCase:   Dict		):
											SCREAMING_SNAKE_CASE_   	=    d

											# for backward compatibility
											if not hasattr(self		,					'''sp_model_kwargs'''		):
																		SCREAMING_SNAKE_CASE_   	=    {}

											SCREAMING_SNAKE_CASE_   	=    spm.SentencePieceProcessor(**self.sp_model_kwargs		)
											self.sp_model.Load(self.vocab_file		)


				def     _A							(							self:   Union[str, Any]		):
											SCREAMING_SNAKE_CASE_   	=    {self.convert_ids_to_tokens(_lowerCamelCase		): i for i in range(self.vocab_size		)}
											vocab.update(self.added_tokens_encoder		)
											return vocab


				def     _A							(							self:   Dict		,					_lowerCamelCase:   str		):
											return self.sp_model.encode(_lowerCamelCase		,					out_type=_lowerCamelCase		)


				def     _A							(							self:   str		,					_lowerCamelCase:   str		):
											if token in self.fairseq_tokens_to_ids:
																		return self.fairseq_tokens_to_ids[token]
											SCREAMING_SNAKE_CASE_   	=    self.sp_model.PieceToId(_lowerCamelCase		)

											# Need to return unknown token if the SP model returned 0
											return spm_id + self.fairseq_offset if spm_id else self.unk_token_id


				def     _A							(							self:   Union[str, Any]		,					_lowerCamelCase:   int		):
											if index in self.fairseq_ids_to_tokens:
																		return self.fairseq_ids_to_tokens[index]
											return self.sp_model.IdToPiece(index - self.fairseq_offset		)


				def     _A							(							self:   List[str]		,					_lowerCamelCase:   Any		):
											SCREAMING_SNAKE_CASE_   	=    []
											SCREAMING_SNAKE_CASE_   	=    ''''''
											SCREAMING_SNAKE_CASE_   	=    False
											for token in tokens:
																		# make sure that special tokens are not decoded using sentencepiece model
																		if token in self.all_special_tokens:
																									if not prev_is_special:
																																out_string += " "
																									out_string += self.sp_model.decode(_lowerCamelCase		) + token
																									SCREAMING_SNAKE_CASE_   	=    True
																									SCREAMING_SNAKE_CASE_   	=    []
																		else:
																									current_sub_tokens.append(_lowerCamelCase		)
																									SCREAMING_SNAKE_CASE_   	=    False
											out_string += self.sp_model.decode(_lowerCamelCase		)
											return out_string.strip()


				def     _A							(							self:   List[Any]		,					_lowerCamelCase:   str		,					_lowerCamelCase:   Optional[str] = None		):
											if not os.path.isdir(_lowerCamelCase		):
																		logger.error(f"Vocabulary path ({save_directory}) should be a directory"		)
																		return
											SCREAMING_SNAKE_CASE_   	=    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		) and os.path.isfile(self.vocab_file		):
																		copyfile(self.vocab_file		,					_lowerCamelCase		)
											elif not os.path.isfile(self.vocab_file		):
																		with open(_lowerCamelCase		,					'''wb'''		) as fi:
																									SCREAMING_SNAKE_CASE_   	=    self.sp_model.serialized_model_proto()
																									fi.write(_lowerCamelCase		)

											return (out_vocab_file,)


				def     _A							(							self:   Tuple		,					_lowerCamelCase:   List[int]		,					_lowerCamelCase:   Optional[List[int]] = None		,					_lowerCamelCase:   bool = False		):

											if already_has_special_tokens:
																		return super().get_special_tokens_mask(
																		    token_ids_a=_lowerCamelCase		,					token_ids_a=_lowerCamelCase		,					already_has_special_tokens=_lowerCamelCase		)

											SCREAMING_SNAKE_CASE_   	=    [1] * len(self.prefix_tokens		)
											SCREAMING_SNAKE_CASE_   	=    [1] * len(self.suffix_tokens		)
											if token_ids_a is None:
																		return prefix_ones + ([0] * len(_lowerCamelCase		)) + suffix_ones
											return prefix_ones + ([0] * len(_lowerCamelCase		)) + ([0] * len(_lowerCamelCase		)) + suffix_ones


				def     _A							(							self:   str		,					_lowerCamelCase:   List[int]		,					_lowerCamelCase:   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     _A							(							self:   Optional[int]		,					_lowerCamelCase:   Dict		,					_lowerCamelCase:   str		,					_lowerCamelCase:   Optional[str]		,					_lowerCamelCase:   Optional[str]		,					**_lowerCamelCase:   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'''		)
											SCREAMING_SNAKE_CASE_   	=    src_lang
											SCREAMING_SNAKE_CASE_   	=    self(_lowerCamelCase		,					add_special_tokens=_lowerCamelCase		,					return_tensors=_lowerCamelCase		,					**_lowerCamelCase		)
											SCREAMING_SNAKE_CASE_   	=    self.convert_tokens_to_ids(_lowerCamelCase		)
											SCREAMING_SNAKE_CASE_   	=    tgt_lang_id
											return inputs


				def     _A							(							self:   Union[str, Any]		,					_lowerCamelCase:   List[str]		,					_lowerCamelCase:   str = "en_XX"		,					_lowerCamelCase:   Optional[List[str]] = None		,					_lowerCamelCase:   str = "ro_RO"		,					**_lowerCamelCase:   Dict		,					):
											SCREAMING_SNAKE_CASE_   	=    src_lang
											SCREAMING_SNAKE_CASE_   	=    tgt_lang
											return super().prepare_seqaseq_batch(_lowerCamelCase		,					_lowerCamelCase		,					**_lowerCamelCase		)


				def     _A							(							self:   Dict		):
											return self.set_src_lang_special_tokens(self.src_lang		)


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


				def     _A							(							self:   List[str]		,					_lowerCamelCase:   str		):
											SCREAMING_SNAKE_CASE_   	=    self.lang_code_to_id[src_lang]
											SCREAMING_SNAKE_CASE_   	=    [self.cur_lang_code_id]
											SCREAMING_SNAKE_CASE_   	=    [self.eos_token_id]





				def     _A							(							self:   Optional[Any]		,					_lowerCamelCase:   str		):
											SCREAMING_SNAKE_CASE_   	=    self.lang_code_to_id[tgt_lang]
											SCREAMING_SNAKE_CASE_   	=    [self.cur_lang_code_id]
											SCREAMING_SNAKE_CASE_   	=    [self.eos_token_id]








import unittest

import numpy as np

from transformers.testing_utils import is_flaky, 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 DonutImageProcessor






class  __magic_name__						(  unittest.TestCase):
				'''simple docstring'''







				def __init__(							self:   Union[str, Any]		,					_lowerCamelCase:   List[str]		,					_lowerCamelCase:   Union[str, Any]=7		,					_lowerCamelCase:   int=3		,					_lowerCamelCase:   Optional[int]=18		,					_lowerCamelCase:   Optional[Any]=30		,					_lowerCamelCase:   Any=4_00		,					_lowerCamelCase:   List[str]=True		,					_lowerCamelCase:   str=None		,					_lowerCamelCase:   Union[str, Any]=True		,					_lowerCamelCase:   Union[str, Any]=False		,					_lowerCamelCase:   int=True		,					_lowerCamelCase:   Any=True		,					_lowerCamelCase:   List[str]=[0.5, 0.5, 0.5]		,					_lowerCamelCase:   Dict=[0.5, 0.5, 0.5]		,					):
											SCREAMING_SNAKE_CASE_   	=    parent
											SCREAMING_SNAKE_CASE_   	=    batch_size
											SCREAMING_SNAKE_CASE_   	=    num_channels
											SCREAMING_SNAKE_CASE_   	=    image_size
											SCREAMING_SNAKE_CASE_   	=    min_resolution
											SCREAMING_SNAKE_CASE_   	=    max_resolution
											SCREAMING_SNAKE_CASE_   	=    do_resize
											SCREAMING_SNAKE_CASE_   	=    size if size is not None else {'''height''': 18, '''width''': 20}
											SCREAMING_SNAKE_CASE_   	=    do_thumbnail
											SCREAMING_SNAKE_CASE_   	=    do_align_axis
											SCREAMING_SNAKE_CASE_   	=    do_pad
											SCREAMING_SNAKE_CASE_   	=    do_normalize
											SCREAMING_SNAKE_CASE_   	=    image_mean
											SCREAMING_SNAKE_CASE_   	=    image_std





				def     _A							(							self:   str		):
											return {
											    "do_resize": self.do_resize,
											    "size": self.size,
											    "do_thumbnail": self.do_thumbnail,
											    "do_align_long_axis": self.do_align_axis,
											    "do_pad": self.do_pad,
											    "do_normalize": self.do_normalize,
											    "image_mean": self.image_mean,
											    "image_std": self.image_std,
											}






@require_torch
@require_vision
class  __magic_name__						(  __UpperCAmelCase    ,  unittest.TestCase):
				'''simple docstring'''







				SCREAMING_SNAKE_CASE__					:							Dict	      =					DonutImageProcessor if is_vision_available() else None


				def     _A							(							self:   Optional[int]		):
											SCREAMING_SNAKE_CASE_   	=    DonutImageProcessingTester(self		)


				@property
				def     _A							(							self:   List[str]		):
											return self.image_processor_tester.prepare_image_processor_dict()


				def     _A							(							self:   Union[str, Any]		):
											SCREAMING_SNAKE_CASE_   	=    self.image_processing_class(**self.image_processor_dict		)
											self.assertTrue(hasattr(_lowerCamelCase		,					'''do_resize'''		)		)
											self.assertTrue(hasattr(_lowerCamelCase		,					'''size'''		)		)
											self.assertTrue(hasattr(_lowerCamelCase		,					'''do_thumbnail'''		)		)
											self.assertTrue(hasattr(_lowerCamelCase		,					'''do_align_long_axis'''		)		)
											self.assertTrue(hasattr(_lowerCamelCase		,					'''do_pad'''		)		)
											self.assertTrue(hasattr(_lowerCamelCase		,					'''do_normalize'''		)		)
											self.assertTrue(hasattr(_lowerCamelCase		,					'''image_mean'''		)		)
											self.assertTrue(hasattr(_lowerCamelCase		,					'''image_std'''		)		)


				def     _A							(							self:   List[str]		):
											SCREAMING_SNAKE_CASE_   	=    self.image_processing_class.from_dict(self.image_processor_dict		)
											self.assertEqual(image_processor.size		,					{'''height''': 18, '''width''': 20}		)

											SCREAMING_SNAKE_CASE_   	=    self.image_processing_class.from_dict(self.image_processor_dict		,					size=42		)
											self.assertEqual(image_processor.size		,					{'''height''': 42, '''width''': 42}		)

											# Previous config had dimensions in (width, height) order
											SCREAMING_SNAKE_CASE_   	=    self.image_processing_class.from_dict(self.image_processor_dict		,					size=(42, 84)		)
											self.assertEqual(image_processor.size		,					{'''height''': 84, '''width''': 42}		)


				def     _A							(							self:   List[Any]		):
											pass


				@is_flaky()
				def     _A							(							self:   int		):
											# Initialize image_processing
											SCREAMING_SNAKE_CASE_   	=    self.image_processing_class(**self.image_processor_dict		)
											# create random PIL images
											SCREAMING_SNAKE_CASE_   	=    prepare_image_inputs(self.image_processor_tester		,					equal_resolution=_lowerCamelCase		)
											for image in image_inputs:
																		self.assertIsInstance(_lowerCamelCase		,					Image.Image		)

											# Test not batched input
											SCREAMING_SNAKE_CASE_   	=    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
											SCREAMING_SNAKE_CASE_   	=    image_processing(_lowerCamelCase		,					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'''],
											    )		,					)


				@is_flaky()
				def     _A							(							self:   Optional[Any]		):
											# Initialize image_processing
											SCREAMING_SNAKE_CASE_   	=    self.image_processing_class(**self.image_processor_dict		)
											# create random numpy tensors
											SCREAMING_SNAKE_CASE_   	=    prepare_image_inputs(self.image_processor_tester		,					equal_resolution=_lowerCamelCase		,					numpify=_lowerCamelCase		)
											for image in image_inputs:
																		self.assertIsInstance(_lowerCamelCase		,					np.ndarray		)

											# Test not batched input
											SCREAMING_SNAKE_CASE_   	=    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
											SCREAMING_SNAKE_CASE_   	=    image_processing(_lowerCamelCase		,					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'''],
											    )		,					)





				@is_flaky()
				def     _A							(							self:   List[str]		):
											# Initialize image_processing
											SCREAMING_SNAKE_CASE_   	=    self.image_processing_class(**self.image_processor_dict		)
											# create random PyTorch tensors
											SCREAMING_SNAKE_CASE_   	=    prepare_image_inputs(self.image_processor_tester		,					equal_resolution=_lowerCamelCase		,					torchify=_lowerCamelCase		)
											for image in image_inputs:
																		self.assertIsInstance(_lowerCamelCase		,					torch.Tensor		)

											# Test not batched input
											SCREAMING_SNAKE_CASE_   	=    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
											SCREAMING_SNAKE_CASE_   	=    image_processing(_lowerCamelCase		,					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'''],
											    )		,					)


