





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


__lowercase       :Tuple       =  logging.get_logger(__name__)

__lowercase       :Tuple       =  """▁"""

__lowercase       :Optional[Any]       =  {"""vocab_file""": """sentencepiece.bpe.model"""}

__lowercase       :List[str]       =  {
    """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"""
        ),
    }
}

__lowercase       :Dict       =  {
    """facebook/mbart-large-en-ro""": 1_024,
    """facebook/mbart-large-cc25""": 1_024,
}

# fmt: off
__lowercase       :int       =  ["""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    _a     (				__snake_case							):






							"""simple docstring"""

							snake_case_							 =						VOCAB_FILES_NAMES
							snake_case_							 =						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							snake_case_							 =						PRETRAINED_VOCAB_FILES_MAP
							snake_case_							 =						['input_ids', 'attention_mask']

							snake_case_							 =						[]
							snake_case_							 =						[]

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

													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					{} if sp_model_kwargs is None else sp_model_kwargs

													super().__init__(
													    bos_token=A_							,	eos_token=A_							,	unk_token=A_							,	sep_token=A_							,	cls_token=A_							,	pad_token=A_							,	mask_token=A_							,	tokenizer_file=A_							,	src_lang=A_							,	tgt_lang=A_							,	additional_special_tokens=A_							,	sp_model_kwargs=self.sp_model_kwargs							,	**A_							,	)

													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					spm.SentencePieceProcessor(**self.sp_model_kwargs    )
													self.sp_model.Load(str(A_    )    )
													SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					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__     :				Optional[int]   =					{"<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__     :				Any   =					1

													SCREAMING_SNAKE_CASE__     :				int   =					len(self.sp_model    )
													SCREAMING_SNAKE_CASE__     :				str   =					{
													    code: self.sp_model_size + i + self.fairseq_offset for i, code in enumerate(A_    )
													}
													SCREAMING_SNAKE_CASE__     :				List[str]   =					{v: k for k, v in self.lang_code_to_id.items()}
													SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					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__     :				Union[str, Any]   =					{v: k for k, v in self.fairseq_tokens_to_ids.items()}
													SCREAMING_SNAKE_CASE__     :				Any   =					list(self.lang_code_to_id.keys()    )

													if additional_special_tokens is not None:
																			# Only add those special tokens if they are not already there.
																			self._additional_special_tokens.extend(
																			    [t for t in additional_special_tokens if t not in self._additional_special_tokens]    )

													SCREAMING_SNAKE_CASE__     :				int   =					src_lang if src_lang is not None else "en_XX"
													SCREAMING_SNAKE_CASE__     :				str   =					self.lang_code_to_id[self._src_lang]
													SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					tgt_lang
													self.set_src_lang_special_tokens(self._src_lang    )

							def __getstate__(      self				:     List[str]    )	->Union[str, Any]:
													SCREAMING_SNAKE_CASE__     :				Dict   =					self.__dict__.copy()
													SCREAMING_SNAKE_CASE__     :				Tuple   =					None
													SCREAMING_SNAKE_CASE__     :				str   =					self.sp_model.serialized_model_proto()
													return state

							def __setstate__(      self				:     List[Any]							,	a				:     List[str]    )	->List[str]:
													SCREAMING_SNAKE_CASE__     :				int   =					d

													# for backward compatibility
													if not hasattr(self							,	"sp_model_kwargs"    ):
																			SCREAMING_SNAKE_CASE__     :				Optional[int]   =					{}

													SCREAMING_SNAKE_CASE__     :				Tuple   =					spm.SentencePieceProcessor(**self.sp_model_kwargs    )
													self.sp_model.LoadFromSerializedProto(self.sp_model_proto    )

							@property
							def      A_  (      self				:     Optional[int]    )	->int:
													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				:     str    )	->Any:
													return self._src_lang

							@src_lang.setter
							def      A_  (      self				:     Optional[int]							,	a				:     int    )	->Dict:
													SCREAMING_SNAKE_CASE__     :				Dict   =					new_src_lang
													self.set_src_lang_special_tokens(self._src_lang    )

							def      A_  (      self				:     Optional[int]							,	a				:     Dict							,	a				:     Any = None							,	a				:     Tuple = False    )	->int:




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

													SCREAMING_SNAKE_CASE__     :				int   =					[1] * len(self.prefix_tokens    )
													SCREAMING_SNAKE_CASE__     :				List[Any]   =					[1] * len(self.suffix_tokens    )
													if token_ids_a is None:
																			return prefix_ones + ([0] * len(A_    )) + suffix_ones
													return prefix_ones + ([0] * len(A_    )) + ([0] * len(A_    )) + suffix_ones

							def      A_  (      self				:     str							,	a				:     Union[str, Any]							,	a				:     int = None    )	->Optional[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      A_  (      self				:     Any							,	a				:     Tuple							,	a				:     Union[str, Any] = None    )	->Any:
													SCREAMING_SNAKE_CASE__     :				Dict   =					[self.sep_token_id]
													SCREAMING_SNAKE_CASE__     :				Union[str, 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      A_  (      self				:     int							,	a				:     Optional[int]							,	a				:     List[Any]							,	a				:     Optional[int]							,	a				:     int							,	**a				:     Dict    )	->int:




													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__     :				Tuple   =					src_lang
													SCREAMING_SNAKE_CASE__     :				Tuple   =					self(A_							,	add_special_tokens=A_							,	return_tensors=A_							,	**A_    )
													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					self.convert_tokens_to_ids(A_    )
													SCREAMING_SNAKE_CASE__     :				Dict   =					tgt_lang_id
													return inputs

							def      A_  (      self				:     int    )	->Any:
													SCREAMING_SNAKE_CASE__     :				Tuple   =					{self.convert_ids_to_tokens(A_    ): i for i in range(self.vocab_size    )}
													vocab.update(self.added_tokens_encoder    )
													return vocab

							def      A_  (      self				:     List[Any]							,	a				:     Optional[Any]    )	->int:
													return self.sp_model.encode(A_							,	out_type=A_    )

							def      A_  (      self				:     Optional[int]							,	a				:     Dict    )	->Any:




													if token in self.fairseq_tokens_to_ids:
																			return self.fairseq_tokens_to_ids[token]
													SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					self.sp_model.PieceToId(A_    )

													# 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]							,	a				:     int    )	->List[Any]:




													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				:     Tuple							,	a				:     Tuple    )	->Union[str, Any]:
													SCREAMING_SNAKE_CASE__     :				Dict   =					"".join(A_    ).replace(A_							,	" "    ).strip()
													return out_string

							def      A_  (      self				:     Union[str, Any]							,	a				:     Tuple							,	a				:     Dict = None    )	->List[Any]:




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

													return (out_vocab_file,)

							def      A_  (      self				:     List[str]							,	a				:     int							,	a				:     Tuple = "en_XX"							,	a				:     List[Any] = None							,	a				:     Optional[int] = "ro_RO"							,	**a				:     str							,	)	->Dict:
													SCREAMING_SNAKE_CASE__     :				str   =					src_lang
													SCREAMING_SNAKE_CASE__     :				Dict   =					tgt_lang
													return super().prepare_seqaseq_batch(A_							,	A_							,	**A_    )

							def      A_  (      self				:     Tuple    )	->Optional[int]:
													return self.set_src_lang_special_tokens(self.src_lang    )

							def      A_  (      self				:     List[str]    )	->Dict:
													return self.set_tgt_lang_special_tokens(self.tgt_lang    )

							def      A_  (      self				:     Any							,	a				:     str    )	->str:
													SCREAMING_SNAKE_CASE__     :				List[str]   =					self.lang_code_to_id[src_lang]
													SCREAMING_SNAKE_CASE__     :				List[str]   =					[]
													SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					[self.eos_token_id, self.cur_lang_code]






							def      A_  (      self				:     Optional[Any]							,	a				:     Optional[int]    )	->int:
													SCREAMING_SNAKE_CASE__     :				List[Any]   =					self.lang_code_to_id[lang]
													SCREAMING_SNAKE_CASE__     :				Any   =					[]
													SCREAMING_SNAKE_CASE__     :				Dict   =					[self.eos_token_id, self.cur_lang_code]






from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import (
    center_crop,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    IMAGENET_STANDARD_MEAN,
    IMAGENET_STANDARD_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, logging


__lowercase       :int       =  logging.get_logger(__name__)



class    _a     (				lowercase__							):






							"""simple docstring"""

							snake_case_							 =						["pixel_values"]

							def __init__(      self				:     int							,	a				:     bool = True							,	a				:     Optional[Dict[str, int]] = None							,	a				:     PILImageResampling = PILImageResampling.BILINEAR							,	a				:     bool = True							,	a				:     Dict[str, int] = None							,	a				:     bool = True							,	a				:     Union[int, float] = 1 / 2_55							,	a				:     bool = True							,	a				:     Optional[Union[float, List[float]]] = None							,	a				:     Optional[Union[float, List[float]]] = None							,	**a				:     List[str]							,	)	->None:
													super().__init__(**a    )
													SCREAMING_SNAKE_CASE__     :				List[str]   =					size if size is not None else {"shortest_edge": 2_56}
													SCREAMING_SNAKE_CASE__     :				Any   =					get_size_dict(a							,	default_to_square=a    )
													SCREAMING_SNAKE_CASE__     :				List[Any]   =					crop_size if crop_size is not None else {"height": 2_24, "width": 2_24}
													SCREAMING_SNAKE_CASE__     :				Dict   =					get_size_dict(a    )
													SCREAMING_SNAKE_CASE__     :				List[str]   =					do_resize
													SCREAMING_SNAKE_CASE__     :				List[str]   =					size
													SCREAMING_SNAKE_CASE__     :				List[Any]   =					resample
													SCREAMING_SNAKE_CASE__     :				int   =					do_center_crop
													SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					crop_size
													SCREAMING_SNAKE_CASE__     :				Any   =					do_rescale
													SCREAMING_SNAKE_CASE__     :				Any   =					rescale_factor
													SCREAMING_SNAKE_CASE__     :				int   =					do_normalize
													SCREAMING_SNAKE_CASE__     :				str   =					image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
													SCREAMING_SNAKE_CASE__     :				int   =					image_std if image_std is not None else IMAGENET_STANDARD_STD

							def      A_  (      self				:     Tuple							,	a				:     np.ndarray							,	a				:     Dict[str, int]							,	a				:     PILImageResampling = PILImageResampling.BICUBIC							,	a				:     Optional[Union[str, ChannelDimension]] = None							,	**a				:     Optional[int]							,	)	->np.ndarray:
													SCREAMING_SNAKE_CASE__     :				List[Any]   =					get_size_dict(a							,	default_to_square=a    )
													if "shortest_edge" not in size:
																			raise ValueError(f"""The `size` parameter must contain the key `shortest_edge`. Got {size.keys()}"""    )
													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					get_resize_output_image_size(a							,	size=size["shortest_edge"]							,	default_to_square=a    )
													return resize(a							,	size=a							,	resample=a							,	data_format=a							,	**a    )

							def      A_  (      self				:     List[Any]							,	a				:     np.ndarray							,	a				:     Dict[str, int]							,	a				:     Optional[Union[str, ChannelDimension]] = None							,	**a				:     List[Any]							,	)	->np.ndarray:
													SCREAMING_SNAKE_CASE__     :				Tuple   =					get_size_dict(a    )
													return center_crop(a							,	size=(size["height"], size["width"])							,	data_format=a							,	**a    )

							def      A_  (      self				:     Optional[int]							,	a				:     np.ndarray							,	a				:     float							,	a				:     Optional[Union[str, ChannelDimension]] = None							,	**a				:     Dict    )	->np.ndarray:
													return rescale(a							,	scale=a							,	data_format=a							,	**a    )

							def      A_  (      self				:     Union[str, Any]							,	a				:     np.ndarray							,	a				:     Union[float, List[float]]							,	a				:     Union[float, List[float]]							,	a				:     Optional[Union[str, ChannelDimension]] = None							,	**a				:     Union[str, Any]							,	)	->np.ndarray:
													return normalize(a							,	mean=a							,	std=a							,	data_format=a							,	**a    )






							def      A_  (      self				:     Tuple							,	a				:     ImageInput							,	a				:     Optional[bool] = None							,	a				:     Dict[str, int] = None							,	a				:     PILImageResampling = None							,	a				:     bool = None							,	a				:     Dict[str, int] = None							,	a				:     Optional[bool] = None							,	a				:     Optional[float] = None							,	a				:     Optional[bool] = None							,	a				:     Optional[Union[float, List[float]]] = None							,	a				:     Optional[Union[float, List[float]]] = None							,	a				:     Optional[Union[str, TensorType]] = None							,	a				:     Union[str, ChannelDimension] = ChannelDimension.FIRST							,	**a				:     Any							,	)	->Optional[int]:
													SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					do_resize if do_resize is not None else self.do_resize
													SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					size if size is not None else self.size
													SCREAMING_SNAKE_CASE__     :				Dict   =					get_size_dict(a							,	default_to_square=a    )
													SCREAMING_SNAKE_CASE__     :				str   =					resample if resample is not None else self.resample
													SCREAMING_SNAKE_CASE__     :				List[str]   =					do_center_crop if do_center_crop is not None else self.do_center_crop
													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					crop_size if crop_size is not None else self.crop_size
													SCREAMING_SNAKE_CASE__     :				Dict   =					get_size_dict(a    )
													SCREAMING_SNAKE_CASE__     :				List[str]   =					do_rescale if do_rescale is not None else self.do_rescale
													SCREAMING_SNAKE_CASE__     :				int   =					rescale_factor if rescale_factor is not None else self.rescale_factor
													SCREAMING_SNAKE_CASE__     :				Dict   =					do_normalize if do_normalize is not None else self.do_normalize
													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					image_mean if image_mean is not None else self.image_mean
													SCREAMING_SNAKE_CASE__     :				Tuple   =					image_std if image_std is not None else self.image_std

													SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					make_list_of_images(a    )

													if not valid_images(a    ):
																			raise ValueError(
																			    "Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, "
																			    "torch.Tensor, tf.Tensor or jax.ndarray."    )

													if do_resize and size is None:
																			raise ValueError("Size must be specified if do_resize is True."    )

													if do_center_crop and crop_size is None:
																			raise ValueError("Crop size must be specified if do_center_crop is True."    )

													if do_rescale and rescale_factor is None:
																			raise ValueError("Rescale factor must be specified if do_rescale is True."    )

													if do_normalize and (image_mean is None or image_std is None):
																			raise ValueError("Image mean and std must be specified if do_normalize is True."    )

													# All transformations expect numpy arrays.
													SCREAMING_SNAKE_CASE__     :				List[str]   =					[to_numpy_array(a    ) for image in images]

													if do_resize:
																			SCREAMING_SNAKE_CASE__     :				Tuple   =					[self.resize(image=a							,	size=a							,	resample=a    ) for image in images]

													if do_center_crop:
																			SCREAMING_SNAKE_CASE__     :				List[Any]   =					[self.center_crop(image=a							,	size=a    ) for image in images]

													if do_rescale:
																			SCREAMING_SNAKE_CASE__     :				List[str]   =					[self.rescale(image=a							,	scale=a    ) for image in images]

													if do_normalize:
																			SCREAMING_SNAKE_CASE__     :				Dict   =					[self.normalize(image=a							,	mean=a							,	std=a    ) for image in images]

													SCREAMING_SNAKE_CASE__     :				Dict   =					[to_channel_dimension_format(a							,	a    ) for image in images]

													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					{"pixel_values": images}
													return BatchFeature(data=a							,	tensor_type=a    )