
'''simple docstring'''





from queue import Queue
from typing import TYPE_CHECKING, Optional


if TYPE_CHECKING:
						from ..models.auto import AutoTokenizer

class 						a   :




				def 	__snake_case							(							self	:				str,    SCREAMING_SNAKE_CASE_	:				Tuple      ):
										raise NotImplementedError()




				def 	__snake_case							(							self	:				Optional[int]      ):
										raise NotImplementedError()




class 						a   (				__magic_name__    ):




				def __init__(							self	:				Optional[int],    SCREAMING_SNAKE_CASE_	:				"AutoTokenizer",    SCREAMING_SNAKE_CASE_	:				bool = False,    **SCREAMING_SNAKE_CASE_	:				Dict      ):
										snake_case       :     Tuple =    tokenizer
										snake_case       :     Optional[Any] =    skip_prompt
										snake_case       :     Any =    decode_kwargs

										# variables used in the streaming process
										snake_case       :     str =    []
										snake_case       :     List[str] =    0
										snake_case       :     List[Any] =    True




				def 	__snake_case							(							self	:				List[Any],    SCREAMING_SNAKE_CASE_	:				Optional[Any]      ):
										if len(value.shape      ) > 1 and value.shape[0] > 1:
																raise ValueError('''TextStreamer only supports batch size 1'''      )
										elif len(value.shape      ) > 1:
																snake_case       :     Union[str, Any] =    value[0]

										if self.skip_prompt and self.next_tokens_are_prompt:
																snake_case       :     Dict =    False
																return

										# Add the new token to the cache and decodes the entire thing.
										self.token_cache.extend(value.tolist()      )
										snake_case       :     Tuple =    self.tokenizer.decode(self.token_cache,    **self.decode_kwargs      )

										# After the symbol for a new line, we flush the cache.
										if text.endswith('''\n'''      ):
																snake_case       :     List[str] =    text[self.print_len :]
																snake_case       :     Optional[int] =    []
																snake_case       :     Optional[Any] =    0
										# If the last token is a CJK character, we print the characters.
										elif len(SCREAMING_SNAKE_CASE_      ) > 0 and self._is_chinese_char(ord(text[-1]      )      ):
																snake_case       :     Optional[Any] =    text[self.print_len :]
																self.print_len += len(SCREAMING_SNAKE_CASE_      )
										# Otherwise, prints until the last space char (simple heuristic to avoid printing incomplete words,
										# which may change with the subsequent token -- there are probably smarter ways to do this!)
										else:
																snake_case       :     int =    text[self.print_len : text.rfind(''' '''      ) + 1]
																self.print_len += len(SCREAMING_SNAKE_CASE_      )

										self.on_finalized_text(SCREAMING_SNAKE_CASE_      )




				def 	__snake_case							(							self	:				Optional[Any]      ):
										# Flush the cache, if it exists
										if len(self.token_cache      ) > 0:
																snake_case       :     List[Any] =    self.tokenizer.decode(self.token_cache,    **self.decode_kwargs      )
																snake_case       :     Optional[Any] =    text[self.print_len :]
																snake_case       :     Tuple =    []
																snake_case       :     Optional[int] =    0
										else:
																snake_case       :     int =    ''''''

										snake_case       :     str =    True
										self.on_finalized_text(SCREAMING_SNAKE_CASE_,    stream_end=SCREAMING_SNAKE_CASE_      )




				def 	__snake_case							(							self	:				Dict,    SCREAMING_SNAKE_CASE_	:				str,    SCREAMING_SNAKE_CASE_	:				bool = False      ):
										print(SCREAMING_SNAKE_CASE_,    flush=SCREAMING_SNAKE_CASE_,    end='''''' if not stream_end else None      )




				def 	__snake_case							(							self	:				int,    SCREAMING_SNAKE_CASE_	:				int      ):
										# This defines a "chinese character" as anything in the CJK Unicode block:
										#   https://en.wikipedia.org/wiki/CJK_Unified_Ideographs_(Unicode_block)
										#
										# Note that the CJK Unicode block is NOT all Japanese and Korean characters,
										# despite its name. The modern Korean Hangul alphabet is a different block,
										# as is Japanese Hiragana and Katakana. Those alphabets are used to write
										# space-separated words, so they are not treated specially and handled
										# like the all of the other languages.
										if (
										    (cp >= 0x4_e_0_0 and cp <= 0x9_f_f_f)
										    or (cp >= 0x3_4_0_0 and cp <= 0x4_d_b_f)  #
										    or (cp >= 0x2_0_0_0_0 and cp <= 0x2_a_6_d_f)  #
										    or (cp >= 0x2_a_7_0_0 and cp <= 0x2_b_7_3_f)  #
										    or (cp >= 0x2_b_7_4_0 and cp <= 0x2_b_8_1_f)  #
										    or (cp >= 0x2_b_8_2_0 and cp <= 0x2_c_e_a_f)  #
										    or (cp >= 0xf_9_0_0 and cp <= 0xf_a_f_f)
										    or (cp >= 0x2_f_8_0_0 and cp <= 0x2_f_a_1_f)  #
										):  #
																return True

										return False




class 						a   (				__magic_name__    ):




				def __init__(							self	:				Tuple,    SCREAMING_SNAKE_CASE_	:				"AutoTokenizer",    SCREAMING_SNAKE_CASE_	:				bool = False,    SCREAMING_SNAKE_CASE_	:				Optional[float] = None,    **SCREAMING_SNAKE_CASE_	:				Any      ):
										super().__init__(SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_,    **SCREAMING_SNAKE_CASE_      )
										snake_case       :     Optional[Any] =    Queue()
										snake_case       :     Optional[int] =    None
										snake_case       :     Tuple =    timeout




				def 	__snake_case							(							self	:				int,    SCREAMING_SNAKE_CASE_	:				str,    SCREAMING_SNAKE_CASE_	:				bool = False      ):
										self.text_queue.put(SCREAMING_SNAKE_CASE_,    timeout=self.timeout      )
										if stream_end:
																self.text_queue.put(self.stop_signal,    timeout=self.timeout      )




				def __iter__(							self	:				Optional[Any]      ):
										return self




				def 	__snake_case							(							self	:				str      ):
										snake_case       :     Tuple =    self.text_queue.get(timeout=self.timeout      )
										if value == self.stop_signal:
																raise StopIteration()
										else:
																return value



'''simple docstring'''





import json
from typing import TYPE_CHECKING, List, Optional, Tuple

from tokenizers import pre_tokenizers

from ...tokenization_utils_base import BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_gpta import GPTaTokenizer


if TYPE_CHECKING:
						from transformers.pipelines.conversational import Conversation


UpperCAmelCase    					=   logging.get_logger(__name__)

UpperCAmelCase    					=   {"vocab_file": "vocab.json", "merges_file": "merges.txt", "tokenizer_file": "tokenizer.json"}

UpperCAmelCase    					=   {
    "vocab_file": {
        "gpt2": "https://huggingface.co/gpt2/resolve/main/vocab.json",
        "gpt2-medium": "https://huggingface.co/gpt2-medium/resolve/main/vocab.json",
        "gpt2-large": "https://huggingface.co/gpt2-large/resolve/main/vocab.json",
        "gpt2-xl": "https://huggingface.co/gpt2-xl/resolve/main/vocab.json",
        "distilgpt2": "https://huggingface.co/distilgpt2/resolve/main/vocab.json",
    },
    "merges_file": {
        "gpt2": "https://huggingface.co/gpt2/resolve/main/merges.txt",
        "gpt2-medium": "https://huggingface.co/gpt2-medium/resolve/main/merges.txt",
        "gpt2-large": "https://huggingface.co/gpt2-large/resolve/main/merges.txt",
        "gpt2-xl": "https://huggingface.co/gpt2-xl/resolve/main/merges.txt",
        "distilgpt2": "https://huggingface.co/distilgpt2/resolve/main/merges.txt",
    },
    "tokenizer_file": {
        "gpt2": "https://huggingface.co/gpt2/resolve/main/tokenizer.json",
        "gpt2-medium": "https://huggingface.co/gpt2-medium/resolve/main/tokenizer.json",
        "gpt2-large": "https://huggingface.co/gpt2-large/resolve/main/tokenizer.json",
        "gpt2-xl": "https://huggingface.co/gpt2-xl/resolve/main/tokenizer.json",
        "distilgpt2": "https://huggingface.co/distilgpt2/resolve/main/tokenizer.json",
    },
}

UpperCAmelCase    					=   {
    "gpt2": 1_024,
    "gpt2-medium": 1_024,
    "gpt2-large": 1_024,
    "gpt2-xl": 1_024,
    "distilgpt2": 1_024,
}

class 						a   (				__magic_name__    ):
				_snake_case		    = VOCAB_FILES_NAMES
				_snake_case		    = PRETRAINED_VOCAB_FILES_MAP
				_snake_case		    = PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				_snake_case		    = ['''input_ids''', '''attention_mask''']
				_snake_case		    = GPTaTokenizer




				def __init__(							self	:				Union[str, Any],    SCREAMING_SNAKE_CASE_	:				Union[str, Any]=None,    SCREAMING_SNAKE_CASE_	:				Any=None,    SCREAMING_SNAKE_CASE_	:				List[str]=None,    SCREAMING_SNAKE_CASE_	:				Union[str, Any]="<|endoftext|>",    SCREAMING_SNAKE_CASE_	:				Optional[int]="<|endoftext|>",    SCREAMING_SNAKE_CASE_	:				Dict="<|endoftext|>",    SCREAMING_SNAKE_CASE_	:				Union[str, Any]=False,    **SCREAMING_SNAKE_CASE_	:				Tuple,    ):
										super().__init__(
										    SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_,    tokenizer_file=SCREAMING_SNAKE_CASE_,    unk_token=SCREAMING_SNAKE_CASE_,    bos_token=SCREAMING_SNAKE_CASE_,    eos_token=SCREAMING_SNAKE_CASE_,    add_prefix_space=SCREAMING_SNAKE_CASE_,    **SCREAMING_SNAKE_CASE_,    )

										snake_case       :     Optional[Any] =    kwargs.pop('''add_bos_token''',    SCREAMING_SNAKE_CASE_      )

										snake_case       :     Dict =    json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__()      )
										if pre_tok_state.get('''add_prefix_space''',    SCREAMING_SNAKE_CASE_      ) != add_prefix_space:
																snake_case       :     List[Any] =    getattr(SCREAMING_SNAKE_CASE_,    pre_tok_state.pop('''type'''      )      )
																snake_case       :     Dict =    add_prefix_space
																snake_case       :     List[Any] =    pre_tok_class(**SCREAMING_SNAKE_CASE_      )

										snake_case       :     Optional[Any] =    add_prefix_space




				def 	__snake_case							(							self	:				Optional[Any],    *SCREAMING_SNAKE_CASE_	:				Dict,    **SCREAMING_SNAKE_CASE_	:				List[str]      ):
										snake_case       :     str =    kwargs.get('''is_split_into_words''',    SCREAMING_SNAKE_CASE_      )
										assert self.add_prefix_space or not is_split_into_words, (
										    F"""You need to instantiate {self.__class__.__name__} with add_prefix_space=True """
										    "to use it with pretokenized inputs."
										)

										return super()._batch_encode_plus(*SCREAMING_SNAKE_CASE_,    **SCREAMING_SNAKE_CASE_      )




				def 	__snake_case							(							self	:				Union[str, Any],    *SCREAMING_SNAKE_CASE_	:				Tuple,    **SCREAMING_SNAKE_CASE_	:				int      ):
										snake_case       :     int =    kwargs.get('''is_split_into_words''',    SCREAMING_SNAKE_CASE_      )

										assert self.add_prefix_space or not is_split_into_words, (
										    F"""You need to instantiate {self.__class__.__name__} with add_prefix_space=True """
										    "to use it with pretokenized inputs."
										)

										return super()._encode_plus(*SCREAMING_SNAKE_CASE_,    **SCREAMING_SNAKE_CASE_      )




				def 	__snake_case							(							self	:				List[Any],    SCREAMING_SNAKE_CASE_	:				str,    SCREAMING_SNAKE_CASE_	:				Optional[str] = None      ):
										snake_case       :     List[Any] =    self._tokenizer.model.save(SCREAMING_SNAKE_CASE_,    name=SCREAMING_SNAKE_CASE_      )
										return tuple(SCREAMING_SNAKE_CASE_      )




				def 	__snake_case							(							self	:				Optional[Any],    SCREAMING_SNAKE_CASE_	:				"Conversation"      ):
										snake_case       :     int =    []
										for is_user, text in conversation.iter_texts():
																input_ids.extend(self.encode(SCREAMING_SNAKE_CASE_,    add_special_tokens=SCREAMING_SNAKE_CASE_      ) + [self.eos_token_id]      )

										if len(SCREAMING_SNAKE_CASE_      ) > self.model_max_length:
																snake_case       :     str =    input_ids[-self.model_max_length :]
										return input_ids

