

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


lowerCAmelCase							:  str   	=						logging.get_logger(__name__)

lowerCAmelCase							:  Tuple   	=						{"""vocab_file""": """spiece.model"""}

lowerCAmelCase							:  Optional[Any]   	=						{
    """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""",
    }
}

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

# Segments (not really needed)
lowerCAmelCase							:  Any   	=						0
lowerCAmelCase							:  List[Any]   	=						1
lowerCAmelCase							:  Union[str, Any]   	=						2
lowerCAmelCase							:  Dict   	=						3
lowerCAmelCase							:  List[Any]   	=						4



class   __lowercase				( UpperCAmelCase_	):




			"""simple docstring"""







			_UpperCAmelCase   :							List[Any]										=     VOCAB_FILES_NAMES
			_UpperCAmelCase   :							List[Any]										=     PRETRAINED_VOCAB_FILES_MAP
			_UpperCAmelCase   :							List[str]										=     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
			_UpperCAmelCase   :							List[Any]										=     '''left'''



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

									SCREAMING_SNAKE_CASE_: Optional[Any]  		=       {} 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__   ,						)

									SCREAMING_SNAKE_CASE_: Dict  		=       3

									SCREAMING_SNAKE_CASE_: List[str]  		=       do_lower_case
									SCREAMING_SNAKE_CASE_: List[Any]  		=       remove_space
									SCREAMING_SNAKE_CASE_: int  		=       keep_accents
									SCREAMING_SNAKE_CASE_: Tuple  		=       vocab_file

									SCREAMING_SNAKE_CASE_: Any  		=       spm.SentencePieceProcessor(**self.sp_model_kwargs)
									self.sp_model.Load(lowerCAmelCase__)



			@property
			def        _SCREAMING_SNAKE_CASE			( self  :     List[str]):
									return len(self.sp_model)



			def        _SCREAMING_SNAKE_CASE			( self  :     List[str]):
									SCREAMING_SNAKE_CASE_: Optional[Any]  		=       {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  :     Optional[int]):
									SCREAMING_SNAKE_CASE_: Optional[Any]  		=       self.__dict__.copy()
									SCREAMING_SNAKE_CASE_: Optional[int]  		=       None
									return state



			def __setstate__( self  :     Tuple   ,						lowerCAmelCase__  :     Any):
									SCREAMING_SNAKE_CASE_: Any  		=       d

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

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



			def        _SCREAMING_SNAKE_CASE			( self  :     Tuple   ,						lowerCAmelCase__  :     Any):
									if self.remove_space:
															SCREAMING_SNAKE_CASE_: int  		=       " ".join(inputs.strip().split())
									else:
															SCREAMING_SNAKE_CASE_: int  		=       inputs
									SCREAMING_SNAKE_CASE_: Tuple  		=       outputs.replace("``"   ,						"\"").replace("''"   ,						"\"")

									if not self.keep_accents:
															SCREAMING_SNAKE_CASE_: List[str]  		=       unicodedata.normalize("NFKD"   ,						lowerCAmelCase__)
															SCREAMING_SNAKE_CASE_: int  		=       "".join([c for c in outputs if not unicodedata.combining(lowerCAmelCase__)])
									if self.do_lower_case:
															SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       outputs.lower()

									return outputs



			def        _SCREAMING_SNAKE_CASE			( self  :     str   ,						lowerCAmelCase__  :     str):
									SCREAMING_SNAKE_CASE_: str  		=       self.preprocess_text(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: Optional[Any]  		=       self.sp_model.encode(lowerCAmelCase__   ,						out_type=lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: Tuple  		=       []
									for piece in pieces:
															if len(lowerCAmelCase__) > 1 and piece[-1] == str(",") and piece[-2].isdigit():
																					SCREAMING_SNAKE_CASE_: List[str]  		=       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:
																																	SCREAMING_SNAKE_CASE_: List[str]  		=       cur_pieces[1:]
																											else:
																																	SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       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  :     List[Any]   ,						lowerCAmelCase__  :     Dict):
									return self.sp_model.PieceToId(lowerCAmelCase__)



			def        _SCREAMING_SNAKE_CASE			( self  :     int   ,						lowerCAmelCase__  :     Dict):
									return self.sp_model.IdToPiece(lowerCAmelCase__)



			def        _SCREAMING_SNAKE_CASE			( self  :     Dict   ,						lowerCAmelCase__  :     Union[str, Any]):
									SCREAMING_SNAKE_CASE_: List[str]  		=       "".join(lowerCAmelCase__).replace(lowerCAmelCase__   ,						" ").strip()
									return out_string



			def        _SCREAMING_SNAKE_CASE			( self  :     Any   ,						lowerCAmelCase__  :     List[int]   ,						lowerCAmelCase__  :     bool = False   ,						lowerCAmelCase__  :     bool = None   ,						lowerCAmelCase__  :     bool = True   ,						**lowerCAmelCase__  :     Optional[int]   ,						):
									SCREAMING_SNAKE_CASE_: List[Any]  		=       kwargs.pop("use_source_tokenizer"   ,						lowerCAmelCase__)

									SCREAMING_SNAKE_CASE_: int  		=       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
									SCREAMING_SNAKE_CASE_: Optional[int]  		=       []
									SCREAMING_SNAKE_CASE_: Optional[Any]  		=       []
									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__))
																											SCREAMING_SNAKE_CASE_: Dict  		=       []
																					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
									SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       "".join(lowerCAmelCase__)

									SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       (
									    clean_up_tokenization_spaces
									    if clean_up_tokenization_spaces is not None
									    else self.clean_up_tokenization_spaces
									)
									if clean_up_tokenization_spaces:
															SCREAMING_SNAKE_CASE_: Dict  		=       self.clean_up_tokenization(lowerCAmelCase__)
															return clean_text
									else:
															return text



			def        _SCREAMING_SNAKE_CASE			( self  :     int   ,						lowerCAmelCase__  :     List[int]   ,						lowerCAmelCase__  :     Optional[List[int]] = None):
									SCREAMING_SNAKE_CASE_: Optional[int]  		=       [self.sep_token_id]
									SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       [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  :     Optional[int]   ,						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__)

									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  :     Dict   ,						lowerCAmelCase__  :     List[int]   ,						lowerCAmelCase__  :     Optional[List[int]] = None):
									SCREAMING_SNAKE_CASE_: Tuple  		=       [self.sep_token_id]
									SCREAMING_SNAKE_CASE_: List[str]  		=       [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  :     Optional[int]   ,						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_: int  		=       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_: Dict  		=       self.sp_model.serialized_model_proto()
																					fi.write(lowerCAmelCase__)

									return (out_vocab_file,)









import unittest

from transformers import (
    MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING,
    TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING,
    TextaTextGenerationPipeline,
    pipeline,
)
from transformers.testing_utils import is_pipeline_test, require_tf, require_torch
from transformers.utils import is_torch_available

from .test_pipelines_common import ANY


if is_torch_available():
						import torch



@is_pipeline_test
class   __lowercase				( unittest.TestCase	):




			"""simple docstring"""







			_UpperCAmelCase   :							List[Any]										=     MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING
			_UpperCAmelCase   :							str										=     TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING



			def        _SCREAMING_SNAKE_CASE			( self  :     Any   ,						lowerCAmelCase__  :     List[str]   ,						lowerCAmelCase__  :     List[str]   ,						lowerCAmelCase__  :     Dict):
									SCREAMING_SNAKE_CASE_: Any  		=       TextaTextGenerationPipeline(model=lowerCAmelCase__   ,						tokenizer=lowerCAmelCase__)
									return generator, ["Something to write", "Something else"]



			def        _SCREAMING_SNAKE_CASE			( self  :     str   ,						lowerCAmelCase__  :     Dict   ,						lowerCAmelCase__  :     Any):
									SCREAMING_SNAKE_CASE_: List[Any]  		=       generator("Something there")
									self.assertEqual(lowerCAmelCase__   ,						[{"generated_text": ANY(lowerCAmelCase__)}])
									# These are encoder decoder, they don't just append to incoming string
									self.assertFalse(outputs[0]["generated_text"].startswith("Something there"))

									SCREAMING_SNAKE_CASE_: List[Any]  		=       generator(["This is great !", "Something else"]   ,						num_return_sequences=2   ,						do_sample=lowerCAmelCase__)
									self.assertEqual(
									    lowerCAmelCase__   ,						[
									        [{"generated_text": ANY(lowerCAmelCase__)}, {"generated_text": ANY(lowerCAmelCase__)}],
									        [{"generated_text": ANY(lowerCAmelCase__)}, {"generated_text": ANY(lowerCAmelCase__)}],
									    ]   ,						)

									SCREAMING_SNAKE_CASE_: Dict  		=       generator(
									    ["This is great !", "Something else"]   ,						num_return_sequences=2   ,						batch_size=2   ,						do_sample=lowerCAmelCase__)
									self.assertEqual(
									    lowerCAmelCase__   ,						[
									        [{"generated_text": ANY(lowerCAmelCase__)}, {"generated_text": ANY(lowerCAmelCase__)}],
									        [{"generated_text": ANY(lowerCAmelCase__)}, {"generated_text": ANY(lowerCAmelCase__)}],
									    ]   ,						)

									with self.assertRaises(lowerCAmelCase__):
															generator(4)



			@require_torch
			def        _SCREAMING_SNAKE_CASE			( self  :     Tuple):
									SCREAMING_SNAKE_CASE_: Optional[int]  		=       pipeline("text2text-generation"   ,						model="patrickvonplaten/t5-tiny-random"   ,						framework="pt")
									# do_sample=False necessary for reproducibility
									SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       generator("Something there"   ,						do_sample=lowerCAmelCase__)
									self.assertEqual(lowerCAmelCase__   ,						[{"generated_text": ""}])

									SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       3
									SCREAMING_SNAKE_CASE_: Any  		=       generator(
									    "Something there"   ,						num_return_sequences=lowerCAmelCase__   ,						num_beams=lowerCAmelCase__   ,						)
									SCREAMING_SNAKE_CASE_: Any  		=       [
									    {"generated_text": "Beide Beide Beide Beide Beide Beide Beide Beide Beide"},
									    {"generated_text": "Beide Beide Beide Beide Beide Beide Beide Beide"},
									    {"generated_text": ""},
									]
									self.assertEqual(lowerCAmelCase__   ,						lowerCAmelCase__)

									SCREAMING_SNAKE_CASE_: int  		=       generator("This is a test"   ,						do_sample=lowerCAmelCase__   ,						num_return_sequences=2   ,						return_tensors=lowerCAmelCase__)
									self.assertEqual(
									    lowerCAmelCase__   ,						[
									        {"generated_token_ids": ANY(torch.Tensor)},
									        {"generated_token_ids": ANY(torch.Tensor)},
									    ]   ,						)
									SCREAMING_SNAKE_CASE_: str  		=       generator.model.config.eos_token_id
									SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       "<pad>"
									SCREAMING_SNAKE_CASE_: Tuple  		=       generator(
									    ["This is a test", "This is a second test"]   ,						do_sample=lowerCAmelCase__   ,						num_return_sequences=2   ,						batch_size=2   ,						return_tensors=lowerCAmelCase__   ,						)
									self.assertEqual(
									    lowerCAmelCase__   ,						[
									        [
									            {"generated_token_ids": ANY(torch.Tensor)},
									            {"generated_token_ids": ANY(torch.Tensor)},
									        ],
									        [
									            {"generated_token_ids": ANY(torch.Tensor)},
									            {"generated_token_ids": ANY(torch.Tensor)},
									        ],
									    ]   ,						)




			@require_tf
			def        _SCREAMING_SNAKE_CASE			( self  :     Union[str, Any]):
									SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       pipeline("text2text-generation"   ,						model="patrickvonplaten/t5-tiny-random"   ,						framework="tf")
									# do_sample=False necessary for reproducibility
									SCREAMING_SNAKE_CASE_: List[Any]  		=       generator("Something there"   ,						do_sample=lowerCAmelCase__)
									self.assertEqual(lowerCAmelCase__   ,						[{"generated_text": ""}])






