

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

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


lowerCAmelCase							:  int            =     """▁"""

lowerCAmelCase							:  Optional[Any]            =     {"""vocab_file""": """spiece.model"""}

lowerCAmelCase							:  List[str]            =     {
    """vocab_file""": {"""google/pegasus-xsum""": """https://huggingface.co/google/pegasus-xsum/resolve/main/spiece.model"""}
}

lowerCAmelCase							:  List[str]            =     {
    """google/pegasus-xsum""": 512,
}


lowerCAmelCase							:  Union[str, Any]            =     logging.get_logger(__name__)



class   __lowercase    (			__A     ):




			"""simple docstring"""







			_UpperCAmelCase       :							List[Any]      							=					VOCAB_FILES_NAMES

			_UpperCAmelCase       :							List[str]      							=					VOCAB_FILES_NAMES
			_UpperCAmelCase       :							Tuple      							=					PRETRAINED_VOCAB_FILES_MAP
			_UpperCAmelCase       :							Dict      							=					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
			_UpperCAmelCase       :							Tuple      							=					['input_ids', 'attention_mask']



			def __init__(			self  :					Tuple       ,						lowerCAmelCase__  :					Dict       ,						lowerCAmelCase__  :					List[str]="<pad>"       ,						lowerCAmelCase__  :					Tuple="</s>"       ,						lowerCAmelCase__  :					List[str]="<unk>"       ,						lowerCAmelCase__  :					Any="<mask_2>"       ,						lowerCAmelCase__  :					List[Any]="<mask_1>"       ,						lowerCAmelCase__  :					str=None       ,						lowerCAmelCase__  :					Any=103       ,						lowerCAmelCase__  :					Any = None       ,						**lowerCAmelCase__  :					Optional[int]       ,						):
									SCREAMING_SNAKE_CASE_:			List[Any]  		=	offset
									if additional_special_tokens is not None:
															if not isinstance(__lowercase       ,						__lowercase):
																					raise TypeError(
																					    F"additional_special_tokens should be of type {type(__lowercase)}, but is"
																					    F" {type(__lowercase)}")

															SCREAMING_SNAKE_CASE_:			Tuple  		=	(
															    ([mask_token_sent] + additional_special_tokens)
															    if mask_token_sent not in additional_special_tokens and mask_token_sent is not None
															    else additional_special_tokens
															)
															# fill additional tokens with ..., <unk_token_102> in case not all additional tokens are already taken
															additional_special_tokens_extended += [
															    F"<unk_{i}>" for i in range(len(__lowercase)       ,						self.offset - 1)
															]

															if len(set(__lowercase)) != len(__lowercase):
																					raise ValueError(
																					    "Please make sure that the provided additional_special_tokens do not contain an incorrectly"
																					    F" shifted list of <unk_x> tokens. Found {additional_special_tokens_extended}.")
															SCREAMING_SNAKE_CASE_:			Union[str, Any]  		=	additional_special_tokens_extended
									else:
															SCREAMING_SNAKE_CASE_:			Dict  		=	[mask_token_sent] if mask_token_sent is not None else []
															additional_special_tokens += [F"<unk_{i}>" for i in range(2       ,						self.offset)]

									SCREAMING_SNAKE_CASE_:			Dict  		=	{} if sp_model_kwargs is None else sp_model_kwargs

									super().__init__(
									    eos_token=__lowercase       ,						unk_token=__lowercase       ,						mask_token=__lowercase       ,						pad_token=__lowercase       ,						mask_token_sent=__lowercase       ,						offset=__lowercase       ,						additional_special_tokens=__lowercase       ,						sp_model_kwargs=self.sp_model_kwargs       ,						**__lowercase       ,						)
									SCREAMING_SNAKE_CASE_:			Dict  		=	mask_token_sent
									SCREAMING_SNAKE_CASE_:			Tuple  		=	vocab_file
									SCREAMING_SNAKE_CASE_:			Optional[Any]  		=	spm.SentencePieceProcessor(**self.sp_model_kwargs)
									self.sp_model.Load(__lowercase)

									# add special tokens to encoder dict
									SCREAMING_SNAKE_CASE_:			List[Any]  		=	{
									    0: self.pad_token,
									    1: self.eos_token,
									}

									if self.mask_token_sent is not None:
															self.encoder.update(
															    {
															        2: self.mask_token_sent,
															        3: self.mask_token,
															    })

									if self.offset > 0:
															# entries 2-104 are only used for pretraining and called <mask_1>, <mask_2>, unk_2, ...unk_102
															# mask_token_sent is already added to list -> so start at 1
															self.encoder.update({i + 3: additional_special_tokens[i] for i in range(1       ,						self.offset - 1)})

									SCREAMING_SNAKE_CASE_:			Dict  		=	{v: k for k, v in self.encoder.items()}



			@property
			def 	_SCREAMING_SNAKE_CASE      (			self  :					Tuple):
									return len(self.sp_model) + self.offset



			def 	_SCREAMING_SNAKE_CASE      (			self  :					Optional[Any]):
									SCREAMING_SNAKE_CASE_:			List[str]  		=	{self.convert_ids_to_tokens(__lowercase): i for i in range(self.vocab_size)}
									vocab.update(self.added_tokens_encoder)
									return vocab



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



			def __setstate__(			self  :					Optional[int]       ,						lowerCAmelCase__  :					str):
									SCREAMING_SNAKE_CASE_:			Tuple  		=	d

									# for backward compatibility
									if not hasattr(self       ,						"sp_model_kwargs"):
															SCREAMING_SNAKE_CASE_:			List[str]  		=	{}

									SCREAMING_SNAKE_CASE_:			List[Any]  		=	spm.SentencePieceProcessor(**self.sp_model_kwargs)
									self.sp_model.Load(self.vocab_file)



			def 	_SCREAMING_SNAKE_CASE      (			self  :					Optional[Any]       ,						lowerCAmelCase__  :					int):
									return self.sp_model.encode(__lowercase       ,						out_type=__lowercase)



			def 	_SCREAMING_SNAKE_CASE      (			self  :					List[Any]       ,						lowerCAmelCase__  :					int):
									if token in self.decoder:
															return self.decoder[token]
									elif token in self.added_tokens_decoder:
															return self.added_tokens_decoder[token]
									SCREAMING_SNAKE_CASE_:			Union[str, Any]  		=	self.sp_model.piece_to_id(__lowercase)
									return sp_id + self.offset



			def 	_SCREAMING_SNAKE_CASE      (			self  :					Optional[int]       ,						lowerCAmelCase__  :					Union[str, Any]):
									if index in self.encoder:
															return self.encoder[index]
									elif index in self.added_tokens_encoder:
															return self.added_tokens_encoder[index]
									else:
															SCREAMING_SNAKE_CASE_:			str  		=	self.sp_model.IdToPiece(index - self.offset)
									return token



			def 	_SCREAMING_SNAKE_CASE      (			self  :					Union[str, Any]       ,						lowerCAmelCase__  :					Tuple):
									SCREAMING_SNAKE_CASE_:			Optional[Any]  		=	[]
									SCREAMING_SNAKE_CASE_:			str  		=	""
									for token in tokens:
															# make sure that special tokens are not decoded using sentencepiece model
															if token in self.all_special_tokens:
																					out_string += self.sp_model.decode(__lowercase) + token
																					SCREAMING_SNAKE_CASE_:			List[str]  		=	[]
															else:
																					current_sub_tokens.append(__lowercase)
									out_string += self.sp_model.decode(__lowercase)
									return out_string.strip()



			def 	_SCREAMING_SNAKE_CASE      (			self  :					List[str]       ,						lowerCAmelCase__  :					Optional[Any]=False):
									return 1



			def 	_SCREAMING_SNAKE_CASE      (			self  :					Optional[int]       ,						lowerCAmelCase__  :					Optional[Any]):
									SCREAMING_SNAKE_CASE_:			List[Any]  		=	set(self.all_special_ids)  # call it once instead of inside list comp
									all_special_ids.remove(self.unk_token_id)  # <unk> is only sometimes special

									return [1 if x in all_special_ids else 0 for x in seq]



			def 	_SCREAMING_SNAKE_CASE      (			self  :					Optional[Any]       ,						lowerCAmelCase__  :					int       ,						lowerCAmelCase__  :					List[Any] = None       ,						lowerCAmelCase__  :					str = False):
									if already_has_special_tokens:
															return self._special_token_mask(__lowercase)
									elif token_ids_a is None:
															return self._special_token_mask(__lowercase) + [1]
									else:
															return self._special_token_mask(token_ids_a + token_ids_a) + [1]



			def 	_SCREAMING_SNAKE_CASE      (			self  :					List[Any]       ,						lowerCAmelCase__  :					str       ,						lowerCAmelCase__  :					Tuple=None):
									if token_ids_a is None:
															return token_ids_a + [self.eos_token_id]
									# We don't expect to process pairs, but leave the pair logic for API consistency
									return token_ids_a + token_ids_a + [self.eos_token_id]




			def 	_SCREAMING_SNAKE_CASE      (			self  :					List[Any]       ,						lowerCAmelCase__  :					Dict       ,						lowerCAmelCase__  :					List[Any] = None):
									if not os.path.isdir(__lowercase):
															logger.error(F"Vocabulary path ({save_directory}) should be a directory")
															return
									SCREAMING_SNAKE_CASE_:			Optional[Any]  		=	os.path.join(
									    __lowercase       ,						(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"])

									if os.path.abspath(self.vocab_file) != os.path.abspath(__lowercase) and os.path.isfile(self.vocab_file):
															copyfile(self.vocab_file       ,						__lowercase)
									elif not os.path.isfile(self.vocab_file):
															with open(__lowercase       ,						"wb") as fi:
																					SCREAMING_SNAKE_CASE_:			Any  		=	self.sp_model.serialized_model_proto()
																					fi.write(__lowercase)

									return (out_vocab_file,)









from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


lowerCAmelCase							:  Optional[int]            =						{
    """configuration_nllb_moe""": [
        """NLLB_MOE_PRETRAINED_CONFIG_ARCHIVE_MAP""",
        """NllbMoeConfig""",
    ]
}

try:
						if not is_torch_available():
												raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
						pass
else:
						lowerCAmelCase							:  Optional[int]            =						[
						    """NLLB_MOE_PRETRAINED_MODEL_ARCHIVE_LIST""",
						    """NllbMoeForConditionalGeneration""",
						    """NllbMoeModel""",
						    """NllbMoePreTrainedModel""",
						    """NllbMoeTop2Router""",
						    """NllbMoeSparseMLP""",
						]


if TYPE_CHECKING:
						from .configuration_nllb_moe import (
						    NLLB_MOE_PRETRAINED_CONFIG_ARCHIVE_MAP,
						    NllbMoeConfig,
						)

						try:
												if not is_torch_available():
																		raise OptionalDependencyNotAvailable()
						except OptionalDependencyNotAvailable:
												pass
						else:
												from .modeling_nllb_moe import (
												    NLLB_MOE_PRETRAINED_MODEL_ARCHIVE_LIST,
												    NllbMoeForConditionalGeneration,
												    NllbMoeModel,
												    NllbMoePreTrainedModel,
												    NllbMoeSparseMLP,
												    NllbMoeTopaRouter,
												)


else:
						import sys

						lowerCAmelCase							:  Any            =						_LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)






