






import json
import os
from functools import lru_cache
from typing import TYPE_CHECKING, List, Optional, Tuple

import regex as re

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


if TYPE_CHECKING:
		from transformers.pipelines.conversational import Conversation

_a       =    logging.get_logger(__name__)

_a       =    {
    'vocab_file': 'vocab.json',
    'merges_file': 'merges.txt',
}

_a       =    {
    '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',
    },
}

_a       =    {
    'gpt2': 1024,
    'gpt2-medium': 1024,
    'gpt2-large': 1024,
    'gpt2-xl': 1024,
    'distilgpt2': 1024,
}


@lru_cache()
def __A       ( )->    Optional[int]:





						"""simple docstring"""



						_UpperCAmelCase							   =     (
						    list(range(ord('!'      )					,						ord('~'      ) + 1      )      ) + list(range(ord('¡'      )					,						ord('¬'      ) + 1      )      ) + list(range(ord('®'      )					,						ord('ÿ'      ) + 1      )      )
						)
						_UpperCAmelCase							   =     bs[:]
						_UpperCAmelCase							   =     0
						for b in range(2**8      ):
												if b not in bs:
																		bs.append(__lowerCAmelCase      )
																		cs.append(2**8 + n      )
																		n += 1
						_UpperCAmelCase							   =     [chr(__lowerCAmelCase      ) for n in cs]
						return dict(zip(__lowerCAmelCase					,						__lowerCAmelCase      )      )


def __A       ( __lowerCAmelCase      )->    List[str]:





						"""simple docstring"""



						_UpperCAmelCase							   =     set()
						_UpperCAmelCase							   =     word[0]
						for char in word[1:]:
												pairs.add((prev_char, char)      )
												_UpperCAmelCase							   =     char
						return pairs





class __lowerCamelCase					( __UpperCAmelCase):





	"""simple docstring"""






	UpperCamelCase__								= VOCAB_FILES_NAMES
	UpperCamelCase__								= PRETRAINED_VOCAB_FILES_MAP
	UpperCamelCase__								= PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	UpperCamelCase__								= ["input_ids", "attention_mask"]

	def __init__(				self			,       UpperCAmelCase			,       UpperCAmelCase			,       UpperCAmelCase="replace"			,       UpperCAmelCase="<|endoftext|>"			,       UpperCAmelCase="<|endoftext|>"			,       UpperCAmelCase="<|endoftext|>"			,       UpperCAmelCase=None			,       UpperCAmelCase=False			,       UpperCAmelCase=False			,       **UpperCAmelCase			,       ):
							"""simple docstring"""







							_UpperCAmelCase							   =     AddedToken(lowerCamelCase__			,       lstrip=lowerCamelCase__			,       rstrip=lowerCamelCase__       ) if isinstance(lowerCamelCase__			,       lowerCamelCase__       ) else bos_token
							_UpperCAmelCase							   =     AddedToken(lowerCamelCase__			,       lstrip=lowerCamelCase__			,       rstrip=lowerCamelCase__       ) if isinstance(lowerCamelCase__			,       lowerCamelCase__       ) else eos_token
							_UpperCAmelCase							   =     AddedToken(lowerCamelCase__			,       lstrip=lowerCamelCase__			,       rstrip=lowerCamelCase__       ) if isinstance(lowerCamelCase__			,       lowerCamelCase__       ) else unk_token
							_UpperCAmelCase							   =     AddedToken(lowerCamelCase__			,       lstrip=lowerCamelCase__			,       rstrip=lowerCamelCase__       ) if isinstance(lowerCamelCase__			,       lowerCamelCase__       ) else pad_token
							super().__init__(
							    errors=lowerCamelCase__			,       unk_token=lowerCamelCase__			,       bos_token=lowerCamelCase__			,       eos_token=lowerCamelCase__			,       pad_token=lowerCamelCase__			,       add_prefix_space=lowerCamelCase__			,       add_bos_token=lowerCamelCase__			,       **lowerCamelCase__			,       )
							_UpperCAmelCase							   =     add_bos_token

							with open(lowerCamelCase__			,       encoding='utf-8'       ) as vocab_handle:
													_UpperCAmelCase							   =     json.load(lowerCamelCase__       )
							_UpperCAmelCase							   =     {v: k for k, v in self.encoder.items()}
							_UpperCAmelCase							   =     errors  # how to handle errors in decoding
							_UpperCAmelCase							   =     bytes_to_unicode()
							_UpperCAmelCase							   =     {v: k for k, v in self.byte_encoder.items()}
							with open(lowerCamelCase__			,       encoding='utf-8'       ) as merges_handle:
													_UpperCAmelCase							   =     merges_handle.read().split('\n'       )[1:-1]
							_UpperCAmelCase							   =     [tuple(merge.split()       ) for merge in bpe_merges]
							_UpperCAmelCase							   =     dict(zip(lowerCamelCase__			,       range(len(lowerCamelCase__       )       )       )       )
							_UpperCAmelCase							   =     {}
							_UpperCAmelCase							   =     add_prefix_space

							# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
							_UpperCAmelCase							   =     re.compile(R'\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+'       )

	@property
	def 			UpperCamelCase       (				self       ):
							"""simple docstring"""







							return len(self.encoder       )

	def 			UpperCamelCase       (				self       ):
							"""simple docstring"""







							return dict(self.encoder			,       **self.added_tokens_encoder       )

	def 			UpperCamelCase       (				self			,       UpperCAmelCase       ):
							"""simple docstring"""







							if token in self.cache:
													return self.cache[token]
							_UpperCAmelCase							   =     tuple(lowerCamelCase__       )
							_UpperCAmelCase							   =     get_pairs(lowerCamelCase__       )

							if not pairs:
													return token

							while True:
													_UpperCAmelCase							   =     min(lowerCamelCase__			,       key=lambda UpperCAmelCase       : self.bpe_ranks.get(lowerCamelCase__			,       float('inf'       )       )       )
													if bigram not in self.bpe_ranks:
																			break
													_UpperCAmelCase   ,					_UpperCAmelCase							   =     bigram
													_UpperCAmelCase							   =     []
													_UpperCAmelCase							   =     0
													while i < len(lowerCamelCase__       ):
																			try:
																									_UpperCAmelCase							   =     word.index(lowerCamelCase__			,       lowerCamelCase__       )
																			except ValueError:
																									new_word.extend(word[i:]       )
																									break
																			else:
																									new_word.extend(word[i:j]       )
																									_UpperCAmelCase							   =     j

																			if word[i] == first and i < len(lowerCamelCase__       ) - 1 and word[i + 1] == second:
																									new_word.append(first + second       )
																									i += 2
																			else:
																									new_word.append(word[i]       )
																									i += 1
													_UpperCAmelCase							   =     tuple(lowerCamelCase__       )
													_UpperCAmelCase							   =     new_word
													if len(lowerCamelCase__       ) == 1:
																			break
													else:
																			_UpperCAmelCase							   =     get_pairs(lowerCamelCase__       )
							_UpperCAmelCase							   =     ' '.join(lowerCamelCase__       )
							_UpperCAmelCase							   =     word
							return word

	def 			UpperCamelCase       (				self			,       UpperCAmelCase			,       UpperCAmelCase=None       ):
							"""simple docstring"""







							if self.add_bos_token:
													_UpperCAmelCase							   =     [self.bos_token_id]
							else:
													_UpperCAmelCase							   =     []

							_UpperCAmelCase							   =     bos_token_ids + token_ids_a

							if token_ids_a is None:
													return output

							return output + bos_token_ids + token_ids_a

	def 			UpperCamelCase       (				self			,       UpperCAmelCase			,       UpperCAmelCase = None			,       UpperCAmelCase = False       ):
							"""simple docstring"""







							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 not self.add_bos_token:
													return super().get_special_tokens_mask(
													    token_ids_a=lowerCamelCase__			,       token_ids_a=lowerCamelCase__			,       already_has_special_tokens=lowerCamelCase__       )

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

	def 			UpperCamelCase       (				self			,       UpperCAmelCase       ):
							"""simple docstring"""







							_UpperCAmelCase							   =     []
							for token in re.findall(self.pat			,       lowerCamelCase__       ):
													_UpperCAmelCase							   =     ''.join(
													    self.byte_encoder[b] for b in token.encode('utf-8'       )       )  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
													bpe_tokens.extend(bpe_token for bpe_token in self.bpe(lowerCamelCase__       ).split(' '       )       )
							return bpe_tokens

	def 			UpperCamelCase       (				self			,       UpperCAmelCase       ):
							"""simple docstring"""







							return self.encoder.get(lowerCamelCase__			,       self.encoder.get(self.unk_token       )       )

	def 			UpperCamelCase       (				self			,       UpperCAmelCase       ):
							"""simple docstring"""







							return self.decoder.get(lowerCamelCase__       )

	def 			UpperCamelCase       (				self			,       UpperCAmelCase       ):
							"""simple docstring"""







							_UpperCAmelCase							   =     ''.join(lowerCamelCase__       )
							_UpperCAmelCase							   =     bytearray([self.byte_decoder[c] for c in text]       ).decode('utf-8'			,       errors=self.errors       )
							return text

	def 			UpperCamelCase       (				self			,       UpperCAmelCase			,       UpperCAmelCase = None       ):
							"""simple docstring"""







							if not os.path.isdir(lowerCamelCase__       ):
													logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""       )
													return
							_UpperCAmelCase							   =     os.path.join(
							    lowerCamelCase__			,       (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']       )
							_UpperCAmelCase							   =     os.path.join(
							    lowerCamelCase__			,       (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file']       )

							with open(lowerCamelCase__			,       'w'			,       encoding='utf-8'       ) as f:
													f.write(json.dumps(self.encoder			,       indent=2			,       sort_keys=lowerCamelCase__			,       ensure_ascii=lowerCamelCase__       ) + '\n'       )

							_UpperCAmelCase							   =     0
							with open(lowerCamelCase__			,       'w'			,       encoding='utf-8'       ) as writer:
													writer.write('#version: 0.2\n'       )
													for bpe_tokens, token_index in sorted(self.bpe_ranks.items()			,       key=lambda UpperCAmelCase       : kv[1]       ):
																			if index != token_index:
																									logger.warning(
																									    F"""Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."""
																									    ' Please check that the tokenizer is not corrupted!'       )
																									_UpperCAmelCase							   =     token_index
																			writer.write(' '.join(lowerCamelCase__       ) + '\n'       )
																			index += 1

							return vocab_file, merge_file

	def 			UpperCamelCase       (				self			,       UpperCAmelCase			,       UpperCAmelCase=False			,       **UpperCAmelCase       ):
							"""simple docstring"""







							_UpperCAmelCase							   =     kwargs.pop('add_prefix_space'			,       self.add_prefix_space       )
							if is_split_into_words or add_prefix_space:
													_UpperCAmelCase							   =     ' ' + text
							return (text, kwargs)

	def 			UpperCamelCase       (				self			,       UpperCAmelCase       ):
							"""simple docstring"""







							_UpperCAmelCase							   =     []
							for is_user, text in conversation.iter_texts():
													input_ids.extend(self.encode(lowerCamelCase__			,       add_special_tokens=lowerCamelCase__       ) + [self.eos_token_id]       )
							if len(lowerCamelCase__       ) > self.model_max_length:
													_UpperCAmelCase							   =     input_ids[-self.model_max_length :]
							return input_ids

"""simple docstring"""






import os
import unittest

import pytest
from parameterized import parameterized

from transformers.testing_utils import require_torch, require_torch_gpu, slow, torch_device


# skipping in unittest tests

lowerCAmelCase__    :    Union[str, Any]			=      [(1,)]



def     a_		(   ):
				UpperCAmelCase__			 =				bool(os.getenv('RUN_SLOW'   ,     0			)			)
				if run_slow:
								assert True
				else:
								assert False, "should have been skipped"



def     a_		(   ):
				UpperCAmelCase__			 =				bool(os.getenv('RUN_SLOW'   ,     0			)			)
				if run_slow and torch_device == "cuda":
								assert True
				else:
								assert False, "should have been skipped"
@require_torch
class 							snake_case			(					unittest.TestCase							):




					"""simple docstring"""

					@slow
					@require_torch_gpu
					def 						__lowerCAmelCase		(							self  :      Optional[Any]     ):
									check_slow_torch_cuda()







					@require_torch_gpu
					@slow
					def 						__lowerCAmelCase		(							self  :      Any     ):
									check_slow_torch_cuda()







					@parameterized.expand(lowerCamelCase__     )
					@slow
					def 						__lowerCAmelCase		(							self  :      Union[str, Any]       ,lowerCamelCase__  :      str=None     ):
									check_slow()






@slow
@require_torch_gpu
def     a_		(   ):
				check_slow_torch_cuda()



@require_torch_gpu
@slow
def     a_		(   ):
				check_slow_torch_cuda()



@slow
@pytest.mark.parametrize('param'   ,     [1]			)
def     a_		(   lowerCamelCase			):
				check_slow()







@pytest.mark.parametrize('param'   ,     [1]			)
@slow
def     a_		(   lowerCamelCase			):
				check_slow()



