



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

import sentencepiece as spm

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


__SCREAMING_SNAKE_CASE           =  logging.get_logger(__name__)

__SCREAMING_SNAKE_CASE           =  {"""vocab_file""": """spiece.model"""}

__SCREAMING_SNAKE_CASE           =  {
    """vocab_file""": {
        """xlnet-base-cased""": """https://huggingface.co/xlnet-base-cased/resolve/main/spiece.model""",
        """xlnet-large-cased""": """https://huggingface.co/xlnet-large-cased/resolve/main/spiece.model""",
    }
}

__SCREAMING_SNAKE_CASE           =  {
    """xlnet-base-cased""": None,
    """xlnet-large-cased""": None,
}

# Segments (not really needed)
__SCREAMING_SNAKE_CASE           =  0
__SCREAMING_SNAKE_CASE           =  1
__SCREAMING_SNAKE_CASE           =  2
__SCREAMING_SNAKE_CASE           =  3
__SCREAMING_SNAKE_CASE           =  4

class 	lowerCamelCase_    ( _A						):






						'''simple docstring'''
						a__   =  VOCAB_FILES_NAMES
						a__   =  PRETRAINED_VOCAB_FILES_MAP
						a__   =  PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
						a__   =  "left"







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

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

									super().__init__(
									    do_lower_case=__lowerCamelCase	,	remove_space=__lowerCamelCase	,	keep_accents=__lowerCamelCase	,	bos_token=__lowerCamelCase	,	eos_token=__lowerCamelCase	,	unk_token=__lowerCamelCase	,	sep_token=__lowerCamelCase	,	pad_token=__lowerCamelCase	,	cls_token=__lowerCamelCase	,	mask_token=__lowerCamelCase	,	additional_special_tokens=__lowerCamelCase	,	sp_model_kwargs=self.sp_model_kwargs	,	**__lowerCamelCase	,	)

									A							:					int              =						3

									A							:					Union[str, Any]              =						do_lower_case
									A							:					List[Any]              =						remove_space
									A							:					List[Any]              =						keep_accents
									A							:					str              =						vocab_file

									A							:					Optional[Any]              =						spm.SentencePieceProcessor(**self.sp_model_kwargs )
									self.sp_model.Load(__lowerCamelCase )







						@property
						def 			SCREAMING_SNAKE_CASE__		(	self      :			str )      ->				int:
									return len(self.sp_model )







						def 			SCREAMING_SNAKE_CASE__		(	self      :			Union[str, Any] )      ->				Dict:
									A							:					str              =						{self.convert_ids_to_tokens(__lowerCamelCase ): i for i in range(self.vocab_size )}
									vocab.update(self.added_tokens_encoder )
									return vocab







						def __getstate__(	self      :			Tuple )      ->				Optional[Any]:
									A							:					Optional[int]              =						self.__dict__.copy()
									A							:					Dict              =						None
									return state







						def __setstate__(	self      :			Any	,	__lowerCamelCase      :			Any )      ->				Dict:
									A							:					Union[str, Any]              =						d

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

									A							:					Optional[int]              =						spm.SentencePieceProcessor(**self.sp_model_kwargs )
									self.sp_model.Load(self.vocab_file )







						def 			SCREAMING_SNAKE_CASE__		(	self      :			Union[str, Any]	,	__lowerCamelCase      :			int )      ->				Any:
									if self.remove_space:
												A							:					Union[str, Any]              =						" ".join(inputs.strip().split() )
									else:
												A							:					str              =						inputs
									A							:					Dict              =						outputs.replace("``"	,	"\"" ).replace("''"	,	"\"" )

									if not self.keep_accents:
												A							:					Optional[int]              =						unicodedata.normalize("NFKD"	,	__lowerCamelCase )
												A							:					Tuple              =						"".join([c for c in outputs if not unicodedata.combining(__lowerCamelCase )] )
									if self.do_lower_case:
												A							:					Tuple              =						outputs.lower()

									return outputs







						def 			SCREAMING_SNAKE_CASE__		(	self      :			Optional[int]	,	__lowerCamelCase      :			str )      ->				List[str]:
									A							:					Dict              =						self.preprocess_text(__lowerCamelCase )
									A							:					Tuple              =						self.sp_model.encode(__lowerCamelCase	,	out_type=__lowerCamelCase )
									A							:					Any              =						[]
									for piece in pieces:
												if len(__lowerCamelCase ) > 1 and piece[-1] == str("," ) and piece[-2].isdigit():
															A							:					Optional[Any]              =						self.sp_model.EncodeAsPieces(piece[:-1].replace(__lowerCamelCase	,	"" ) )
															if piece[0] != SPIECE_UNDERLINE and cur_pieces[0][0] == SPIECE_UNDERLINE:
																		if len(cur_pieces[0] ) == 1:
																					A							:					Optional[Any]              =						cur_pieces[1:]
																		else:
																					A							:					List[str]              =						cur_pieces[0][1:]
															cur_pieces.append(piece[-1] )
															new_pieces.extend(__lowerCamelCase )
												else:
															new_pieces.append(__lowerCamelCase )

									return new_pieces







						def 			SCREAMING_SNAKE_CASE__		(	self      :			Any	,	__lowerCamelCase      :			Tuple )      ->				Dict:
									return self.sp_model.PieceToId(__lowerCamelCase )







						def 			SCREAMING_SNAKE_CASE__		(	self      :			int	,	__lowerCamelCase      :			Optional[int] )      ->				List[str]:
									return self.sp_model.IdToPiece(__lowerCamelCase )







						def 			SCREAMING_SNAKE_CASE__		(	self      :			Optional[Any]	,	__lowerCamelCase      :			Tuple )      ->				Tuple:
									A							:					Any              =						"".join(__lowerCamelCase ).replace(__lowerCamelCase	,	" " ).strip()
									return out_string







						def 			SCREAMING_SNAKE_CASE__		(	self      :			Union[str, Any]	,	__lowerCamelCase      :			List[int]	,	__lowerCamelCase      :			bool = False	,	__lowerCamelCase      :			bool = None	,	__lowerCamelCase      :			bool = True	,	**__lowerCamelCase      :			Union[str, Any]	,	)      ->				str:
									A							:					Any              =						kwargs.pop("use_source_tokenizer"	,	__lowerCamelCase )

									A							:					Any              =						self.convert_ids_to_tokens(__lowerCamelCase	,	skip_special_tokens=__lowerCamelCase )

									# To avoid mixing byte-level and unicode for byte-level BPT
									# we need to build string separately for added tokens and byte-level tokens
									# cf. https://github.com/huggingface/transformers/issues/1133
									A							:					Dict              =						[]
									A							:					Dict              =						[]
									for token in filtered_tokens:
												if skip_special_tokens and token in self.all_special_ids:
															continue
												if token in self.added_tokens_encoder:
															if current_sub_text:
																		sub_texts.append(self.convert_tokens_to_string(__lowerCamelCase ) )
																		A							:					Optional[Any]              =						[]
															sub_texts.append(__lowerCamelCase )
												else:
															current_sub_text.append(__lowerCamelCase )
									if current_sub_text:
												sub_texts.append(self.convert_tokens_to_string(__lowerCamelCase ) )

									# Mimic the behavior of the Rust tokenizer:
									# By default, there are no spaces between special tokens
									A							:					Any              =						"".join(__lowerCamelCase )

									A							:					Optional[int]              =						(
									    clean_up_tokenization_spaces
									    if clean_up_tokenization_spaces is not None
									    else self.clean_up_tokenization_spaces
									)
									if clean_up_tokenization_spaces:
												A							:					Dict              =						self.clean_up_tokenization(__lowerCamelCase )
												return clean_text
									else:
												return text







						def 			SCREAMING_SNAKE_CASE__		(	self      :			Any	,	__lowerCamelCase      :			List[int]	,	__lowerCamelCase      :			Optional[List[int]] = None )      ->				List[int]:
									A							:					Dict              =						[self.sep_token_id]
									A							:					Dict              =						[self.cls_token_id]
									if token_ids_a is None:
												return token_ids_a + sep + cls
									return token_ids_a + sep + token_ids_a + sep + cls







						def 			SCREAMING_SNAKE_CASE__		(	self      :			str	,	__lowerCamelCase      :			List[int]	,	__lowerCamelCase      :			Optional[List[int]] = None	,	__lowerCamelCase      :			bool = False )      ->				List[int]:

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

									if token_ids_a is not None:
												return ([0] * len(__lowerCamelCase )) + [1] + ([0] * len(__lowerCamelCase )) + [1, 1]
									return ([0] * len(__lowerCamelCase )) + [1, 1]







						def 			SCREAMING_SNAKE_CASE__		(	self      :			List[Any]	,	__lowerCamelCase      :			List[int]	,	__lowerCamelCase      :			Optional[List[int]] = None )      ->				List[int]:
									A							:					Optional[Any]              =						[self.sep_token_id]
									A							:					Dict              =						[2]

									if token_ids_a is None:
												return len(token_ids_a + sep ) * [0] + cls_segment_id
									return len(token_ids_a + sep ) * [0] + len(token_ids_a + sep ) * [1] + cls_segment_id







						def 			SCREAMING_SNAKE_CASE__		(	self      :			Union[str, Any]	,	__lowerCamelCase      :			str	,	__lowerCamelCase      :			Optional[str] = None )      ->				Tuple[str]:
									if not os.path.isdir(__lowerCamelCase ):
												logger.error(F"""Vocabulary path ({save_directory}) should be a directory""" )
												return
									A							:					str              =						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:
															A							:					Tuple              =						self.sp_model.serialized_model_proto()
															fi.write(__lowerCamelCase )

									return (out_vocab_file,)




import time
import warnings
from abc import ABC
from copy import deepcopy
from typing import Optional

import torch

from ..utils import add_start_docstrings, logging


__SCREAMING_SNAKE_CASE           =  logging.get_logger(__name__)


__SCREAMING_SNAKE_CASE           =  r"""
    Args:
        input_ids (`torch.LongTensor` of shape `(batch_size, sequence_length)`):
            Indices of input sequence tokens in the vocabulary.

            Indices can be obtained using [`AutoTokenizer`]. See [`PreTrainedTokenizer.encode`] and
            [`PreTrainedTokenizer.__call__`] for details.

            [What are input IDs?](../glossary#input-ids)
        scores (`torch.FloatTensor` of shape `(batch_size, config.vocab_size)`):
            Prediction scores of a language modeling head. These can be scores for each vocabulary token before SoftMax
            or scores for each vocabulary token after SoftMax.
        kwargs (`Dict[str, Any]`, *optional*):
            Additional stopping criteria specific kwargs.

    Return:
        `bool`. `False` indicates we should continue, `True` indicates we should stop.

"""

class 	lowerCamelCase_    ( _A						):






						'''simple docstring'''







						@add_start_docstrings(__lowerCamelCase )
						def __call__(	self      :			Optional[int]	,	__lowerCamelCase      :			torch.LongTensor	,	__lowerCamelCase      :			torch.FloatTensor	,	**__lowerCamelCase      :			Optional[Any] )      ->				bool:
									raise NotImplementedError("StoppingCriteria needs to be subclassed" )







class 	lowerCamelCase_    ( _A						):






						'''simple docstring'''







						def __init__(	self      :			List[str]	,	__lowerCamelCase      :			int	,	__lowerCamelCase      :			Optional[int] = None )      ->				List[Any]:
									A							:					str              =						max_length
									A							:					Optional[int]              =						max_position_embeddings







						@add_start_docstrings(__lowerCamelCase )
						def __call__(	self      :			List[str]	,	__lowerCamelCase      :			torch.LongTensor	,	__lowerCamelCase      :			torch.FloatTensor	,	**__lowerCamelCase      :			Any )      ->				bool:
									A							:					List[Any]              =						input_ids.shape[-1]
									A							:					Any              =						cur_len >= self.max_length
									if self.max_position_embeddings is not None and not is_done and cur_len >= self.max_position_embeddings:
												logger.warning_once(
												    "This is a friendly reminder - the current text generation call will exceed the model's predefined "
												    F"""maximum length ({self.max_position_embeddings}). Depending on the model, you may observe """
												    "exceptions, performance degradation, or nothing at all." )
									return is_done







class 	lowerCamelCase_    ( _A						):






						'''simple docstring'''







						def __init__(	self      :			Optional[Any]	,	__lowerCamelCase      :			int	,	__lowerCamelCase      :			int )      ->				List[Any]:
									warnings.warn(
									    "The class `MaxNewTokensCriteria` is deprecated. "
									    F"""Please use `MaxLengthCriteria(max_length={start_length + max_new_tokens})` """
									    "with `max_length = start_length + max_new_tokens` instead."	,	__lowerCamelCase	,	)
									A							:					str              =						start_length
									A							:					Optional[Any]              =						max_new_tokens
									A							:					Dict              =						start_length + max_new_tokens







						@add_start_docstrings(__lowerCamelCase )
						def __call__(	self      :			int	,	__lowerCamelCase      :			torch.LongTensor	,	__lowerCamelCase      :			torch.FloatTensor	,	**__lowerCamelCase      :			Tuple )      ->				bool:
									return input_ids.shape[-1] >= self.max_length







class 	lowerCamelCase_    ( _A						):






						'''simple docstring'''







						def __init__(	self      :			Optional[int]	,	__lowerCamelCase      :			float	,	__lowerCamelCase      :			Optional[float] = None )      ->				List[Any]:
									A							:					str              =						max_time
									A							:					Dict              =						time.time() if initial_timestamp is None else initial_timestamp







						@add_start_docstrings(__lowerCamelCase )
						def __call__(	self      :			Any	,	__lowerCamelCase      :			torch.LongTensor	,	__lowerCamelCase      :			torch.FloatTensor	,	**__lowerCamelCase      :			Tuple )      ->				bool:
									return time.time() - self.initial_timestamp > self.max_time







class 	lowerCamelCase_    ( _A						):






						'''simple docstring'''







						@add_start_docstrings(__lowerCamelCase )
						def __call__(	self      :			Union[str, Any]	,	__lowerCamelCase      :			torch.LongTensor	,	__lowerCamelCase      :			torch.FloatTensor	,	**__lowerCamelCase      :			int )      ->				bool:
									return any(criteria(__lowerCamelCase	,	__lowerCamelCase ) for criteria in self )







						@property
						def 			SCREAMING_SNAKE_CASE__		(	self      :			List[str] )      ->				Optional[int]:
									for stopping_criterium in self:
												if isinstance(__lowerCamelCase	,	__lowerCamelCase ):
															return stopping_criterium.max_length
												elif isinstance(__lowerCamelCase	,	__lowerCamelCase ):
															return stopping_criterium.max_length
									return None







def 	UpperCAmelCase   (						_lowerCamelCase ,    _lowerCamelCase    ):
			A							:					Optional[int]              =						stopping_criteria.max_length
			A							:					Any              =						deepcopy(_lowerCamelCase    )
			if stopping_max_length is not None and stopping_max_length != max_length:
						warnings.warn("You set different `max_length` for stopping criteria and `max_length` parameter" ,    _lowerCamelCase    )
			elif stopping_max_length is None:
						new_stopping_criteria.append(MaxLengthCriteria(max_length=_lowerCamelCase    )    )
			return new_stopping_criteria