
"""simple docstring"""





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_     =	logging.get_logger(__name__)

lowerCAmelCase_     =	{'''vocab_file''': '''spiece.model'''}

lowerCAmelCase_     =	{
    '''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_     =	{
    '''xlnet-base-cased''': None,
    '''xlnet-large-cased''': None,
}

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




class  _snake_case		(    __lowerCamelCase ):







				"""simple docstring"""






				a			=						VOCAB_FILES_NAMES
				a			=						PRETRAINED_VOCAB_FILES_MAP
				a			=						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				a			=						'''left'''





				def __init__(      self							:							Tuple      ,    _A							:							Dict      ,    _A							:							List[str]=False      ,    _A							:							Optional[Any]=True      ,    _A							:							Optional[Any]=False      ,    _A							:							Optional[Any]="<s>"      ,    _A							:							Optional[int]="</s>"      ,    _A							:							Optional[Any]="<unk>"      ,    _A							:							Optional[Any]="<sep>"      ,    _A							:							List[str]="<pad>"      ,    _A							:							Any="<cls>"      ,    _A							:							Any="<mask>"      ,    _A							:							Dict=["<eop>", "<eod>"]      ,    _A							:							Tuple = None      ,    **_A							:							Optional[int]      ,    ):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						str			  =    AddedToken(a_      ,    lstrip=a_      ,    rstrip=a_) if isinstance(a_      ,    a_) else mask_token

											_SCREAMING_SNAKE_CASE		:						int			  =    {} if sp_model_kwargs is None else sp_model_kwargs

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

											_SCREAMING_SNAKE_CASE		:						str			  =    3

											_SCREAMING_SNAKE_CASE		:						Dict			  =    do_lower_case
											_SCREAMING_SNAKE_CASE		:						str			  =    remove_space
											_SCREAMING_SNAKE_CASE		:						Tuple			  =    keep_accents
											_SCREAMING_SNAKE_CASE		:						Dict			  =    vocab_file

											_SCREAMING_SNAKE_CASE		:						List[str]			  =    spm.SentencePieceProcessor(**self.sp_model_kwargs)
											self.sp_model.Load(a_)





				@property
				def 		_lowerCAmelCase		(      self							:							Union[str, Any]):



											"""simple docstring"""







											return len(self.sp_model)





				def 		_lowerCAmelCase		(      self							:							List[Any]):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						List[str]			  =    {self.convert_ids_to_tokens(a_): i for i in range(self.vocab_size)}
											vocab.update(self.added_tokens_encoder)
											return vocab





				def __getstate__(      self							:							Dict):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Any			  =    self.__dict__.copy()
											_SCREAMING_SNAKE_CASE		:						Optional[int]			  =    None
											return state





				def __setstate__(      self							:							Dict      ,    _A							:							Optional[Any]):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Union[str, Any]			  =    d

											# for backward compatibility
											if not hasattr(self      ,    """sp_model_kwargs"""):
																		_SCREAMING_SNAKE_CASE		:						int			  =    {}

											_SCREAMING_SNAKE_CASE		:						Union[str, Any]			  =    spm.SentencePieceProcessor(**self.sp_model_kwargs)
											self.sp_model.Load(self.vocab_file)





				def 		_lowerCAmelCase		(      self							:							str      ,    _A							:							Tuple):



											"""simple docstring"""







											if self.remove_space:
																		_SCREAMING_SNAKE_CASE		:						Optional[int]			  =    " ".join(inputs.strip().split())
											else:
																		_SCREAMING_SNAKE_CASE		:						str			  =    inputs
											_SCREAMING_SNAKE_CASE		:						Any			  =    outputs.replace("""``"""      ,    """\"""").replace("""''"""      ,    """\"""")

											if not self.keep_accents:
																		_SCREAMING_SNAKE_CASE		:						Dict			  =    unicodedata.normalize("""NFKD"""      ,    a_)
																		_SCREAMING_SNAKE_CASE		:						int			  =    "".join([c for c in outputs if not unicodedata.combining(a_)])
											if self.do_lower_case:
																		_SCREAMING_SNAKE_CASE		:						Any			  =    outputs.lower()

											return outputs





				def 		_lowerCAmelCase		(      self							:							Dict      ,    _A							:							int):



											"""simple docstring"""







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

											return new_pieces





				def 		_lowerCAmelCase		(      self							:							List[Any]      ,    _A							:							Optional[Any]):



											"""simple docstring"""







											return self.sp_model.PieceToId(a_)





				def 		_lowerCAmelCase		(      self							:							Optional[Any]      ,    _A							:							str):



											"""simple docstring"""







											return self.sp_model.IdToPiece(a_)





				def 		_lowerCAmelCase		(      self							:							List[str]      ,    _A							:							Dict):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Dict			  =    "".join(a_).replace(a_      ,    """ """).strip()
											return out_string





				def 		_lowerCAmelCase		(      self							:							Optional[Any]      ,    _A							:							Union[str, Any]      ,    _A							:							Any = False      ,    _A							:							Optional[Any] = None      ,    _A							:							List[str] = True      ,    **_A							:							Any      ,    ):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						int			  =    kwargs.pop("""use_source_tokenizer"""      ,    a_)

											_SCREAMING_SNAKE_CASE		:						List[str]			  =    self.convert_ids_to_tokens(a_      ,    skip_special_tokens=a_)

											# 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		:						List[str]			  =    []
											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(a_))
																																_SCREAMING_SNAKE_CASE		:						Union[str, Any]			  =    []
																									sub_texts.append(a_)
																		else:
																									current_sub_text.append(a_)
											if current_sub_text:
																		sub_texts.append(self.convert_tokens_to_string(a_))

											# Mimic the behavior of the Rust tokenizer:
											# By default, there are no spaces between special tokens
											_SCREAMING_SNAKE_CASE		:						Union[str, Any]			  =    "".join(a_)

											_SCREAMING_SNAKE_CASE		:						Optional[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		:						List[Any]			  =    self.clean_up_tokenization(a_)
																		return clean_text
											else:
																		return text





				def 		_lowerCAmelCase		(      self							:							Tuple      ,    _A							:							Optional[Any]      ,    _A							:							Tuple = None):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    [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 		_lowerCAmelCase		(      self							:							int      ,    _A							:							Union[str, Any]      ,    _A							:							Optional[Any] = None      ,    _A							:							Optional[int] = False):



											"""simple docstring"""







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

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





				def 		_lowerCAmelCase		(      self							:							Union[str, Any]      ,    _A							:							Optional[int]      ,    _A							:							List[Any] = None):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    [self.sep_token_id]
											_SCREAMING_SNAKE_CASE		:						Union[str, Any]			  =    [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 		_lowerCAmelCase		(      self							:							int      ,    _A							:							Optional[Any]      ,    _A							:							int = None):



											"""simple docstring"""







											if not os.path.isdir(a_):
																		logger.error(f"""Vocabulary path ({save_directory}) should be a directory""")
																		return
											_SCREAMING_SNAKE_CASE		:						Any			  =    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		:						Dict			  =    self.sp_model.serialized_model_proto()
																									fi.write(a_)

											return (out_vocab_file,)



"""simple docstring"""





import unittest
from queue import Empty
from threading import Thread

from transformers import AutoTokenizer, TextIteratorStreamer, TextStreamer, is_torch_available
from transformers.testing_utils import CaptureStdout, require_torch, torch_device

from ..test_modeling_common import ids_tensor


if is_torch_available():
			import torch

			from transformers import AutoModelForCausalLM




@require_torch
class  _snake_case		(    unittest.TestCase ):







				"""simple docstring"""






				def 		_lowerCAmelCase		(      self							:							List[Any]):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-gpt2""")
											_SCREAMING_SNAKE_CASE		:						Optional[int]			  =    AutoModelForCausalLM.from_pretrained("""hf-internal-testing/tiny-random-gpt2""").to(_A)
											_SCREAMING_SNAKE_CASE		:						Any			  =    -1

											_SCREAMING_SNAKE_CASE		:						List[str]			  =    ids_tensor((1, 5)      ,    vocab_size=model.config.vocab_size).to(_A)
											_SCREAMING_SNAKE_CASE		:						Dict			  =    model.generate(_A      ,    max_new_tokens=1_0      ,    do_sample=_A)
											_SCREAMING_SNAKE_CASE		:						Dict			  =    tokenizer.decode(greedy_ids[0])

											with CaptureStdout() as cs:
																		_SCREAMING_SNAKE_CASE		:						Any			  =    TextStreamer(_A)
																		model.generate(_A      ,    max_new_tokens=1_0      ,    do_sample=_A      ,    streamer=_A)
											# The greedy text should be printed to stdout, except for the final "\n" in the streamer
											_SCREAMING_SNAKE_CASE		:						str			  =    cs.out[:-1]

											self.assertEqual(_A      ,    _A)





				def 		_lowerCAmelCase		(      self							:							Any):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-gpt2""")
											_SCREAMING_SNAKE_CASE		:						Any			  =    AutoModelForCausalLM.from_pretrained("""hf-internal-testing/tiny-random-gpt2""").to(_A)
											_SCREAMING_SNAKE_CASE		:						List[Any]			  =    -1

											_SCREAMING_SNAKE_CASE		:						str			  =    ids_tensor((1, 5)      ,    vocab_size=model.config.vocab_size).to(_A)
											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    model.generate(_A      ,    max_new_tokens=1_0      ,    do_sample=_A)
											_SCREAMING_SNAKE_CASE		:						Any			  =    tokenizer.decode(greedy_ids[0])

											_SCREAMING_SNAKE_CASE		:						List[Any]			  =    TextIteratorStreamer(_A)
											_SCREAMING_SNAKE_CASE		:						Any			  =    {"""input_ids""": input_ids, """max_new_tokens""": 1_0, """do_sample""": False, """streamer""": streamer}
											_SCREAMING_SNAKE_CASE		:						List[Any]			  =    Thread(target=model.generate      ,    kwargs=_A)
											thread.start()
											_SCREAMING_SNAKE_CASE		:						Any			  =    """"""
											for new_text in streamer:
																		streamer_text += new_text

											self.assertEqual(_A      ,    _A)





				def 		_lowerCAmelCase		(      self							:							List[Any]):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Optional[int]			  =    AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-gpt2""")
											_SCREAMING_SNAKE_CASE		:						Dict			  =    AutoModelForCausalLM.from_pretrained("""hf-internal-testing/tiny-random-gpt2""").to(_A)
											_SCREAMING_SNAKE_CASE		:						Any			  =    -1

											_SCREAMING_SNAKE_CASE		:						Any			  =    ids_tensor((1, 5)      ,    vocab_size=model.config.vocab_size).to(_A)
											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    model.generate(_A      ,    max_new_tokens=1_0      ,    do_sample=_A)
											_SCREAMING_SNAKE_CASE		:						str			  =    greedy_ids[:, input_ids.shape[1] :]
											_SCREAMING_SNAKE_CASE		:						Dict			  =    tokenizer.decode(new_greedy_ids[0])

											with CaptureStdout() as cs:
																		_SCREAMING_SNAKE_CASE		:						Any			  =    TextStreamer(_A      ,    skip_prompt=_A)
																		model.generate(_A      ,    max_new_tokens=1_0      ,    do_sample=_A      ,    streamer=_A)
											# The greedy text should be printed to stdout, except for the final "\n" in the streamer
											_SCREAMING_SNAKE_CASE		:						Optional[int]			  =    cs.out[:-1]

											self.assertEqual(_A      ,    _A)





				def 		_lowerCAmelCase		(      self							:							Dict):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						int			  =    AutoTokenizer.from_pretrained("""distilgpt2""")
											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    AutoModelForCausalLM.from_pretrained("""distilgpt2""").to(_A)
											_SCREAMING_SNAKE_CASE		:						int			  =    -1

											_SCREAMING_SNAKE_CASE		:						List[str]			  =    torch.ones((1, 5)      ,    device=_A).long() * model.config.bos_token_id
											with CaptureStdout() as cs:
																		_SCREAMING_SNAKE_CASE		:						Optional[int]			  =    TextStreamer(_A      ,    skip_special_tokens=_A)
																		model.generate(_A      ,    max_new_tokens=1      ,    do_sample=_A      ,    streamer=_A)

											# The prompt contains a special token, so the streamer should not print it. As such, the output text, when
											# re-tokenized, must only contain one token
											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    cs.out[:-1]  # Remove the final "\n"
											_SCREAMING_SNAKE_CASE		:						Tuple			  =    tokenizer(_A      ,    return_tensors="""pt""")
											self.assertEqual(streamer_text_tokenized.input_ids.shape      ,    (1, 1))





				def 		_lowerCAmelCase		(      self							:							str):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-gpt2""")
											_SCREAMING_SNAKE_CASE		:						List[Any]			  =    AutoModelForCausalLM.from_pretrained("""hf-internal-testing/tiny-random-gpt2""").to(_A)
											_SCREAMING_SNAKE_CASE		:						Tuple			  =    -1

											_SCREAMING_SNAKE_CASE		:						Dict			  =    ids_tensor((1, 5)      ,    vocab_size=model.config.vocab_size).to(_A)
											_SCREAMING_SNAKE_CASE		:						int			  =    TextIteratorStreamer(_A      ,    timeout=0.001)
											_SCREAMING_SNAKE_CASE		:						List[Any]			  =    {"""input_ids""": input_ids, """max_new_tokens""": 1_0, """do_sample""": False, """streamer""": streamer}
											_SCREAMING_SNAKE_CASE		:						List[str]			  =    Thread(target=model.generate      ,    kwargs=_A)
											thread.start()

											# The streamer will timeout after 0.001 seconds, so an exception will be raised
											with self.assertRaises(_A):
																		_SCREAMING_SNAKE_CASE		:						str			  =    """"""
																		for new_text in streamer:
																									streamer_text += new_text


