



"""simple docstring"""







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',
    'tokenizer_config_file': 'tokenizer_config.json',
}

A    =			{
    'vocab_file': {'facebook/blenderbot-3B': 'https://huggingface.co/facebook/blenderbot-3B/resolve/main/vocab.json'},
    'merges_file': {'facebook/blenderbot-3B': 'https://huggingface.co/facebook/blenderbot-3B/resolve/main/merges.txt'},
    'tokenizer_config_file': {
        'facebook/blenderbot-3B': 'https://huggingface.co/facebook/blenderbot-3B/resolve/main/tokenizer_config.json'
    },
}

A    =			{'facebook/blenderbot-3B': 1_2_8}

@lru_cache()
# Copied from transformers.models.roberta.tokenization_roberta.bytes_to_unicode
def   lowerCAmelCase__			(     ) ->							Optional[Any]:
							A         =					(
							    list(range(ord('!'			)				,      ord('~'			) + 1			)			) + list(range(ord('¡'			)				,      ord('¬'			) + 1			)			) + list(range(ord('®'			)				,      ord('ÿ'			) + 1			)			)
							)
							A         =					bs[:]
							A         =					0
							for b in range(2**8			):
														if b not in bs:
																					bs.append(lowerCamelCase__			)
																					cs.append(2**8 + n			)
																					n += 1
							A         =					[chr(lowerCamelCase__			) for n in cs]
							return dict(zip(lowerCamelCase__				,      lowerCamelCase__			)			)

def   lowerCAmelCase__			(     lowerCamelCase__			) ->							Optional[Any]:
							A         =					set()
							A         =					word[0]
							for char in word[1:]:
														pairs.add((prev_char, char)			)
														A         =					char
							return pairs






class      UpperCAmelCase__		(      UpperCamelCase     ):
		lowerCAmelCase_       :	str 							=     VOCAB_FILES_NAMES
		lowerCAmelCase_       :	Tuple 							=     PRETRAINED_VOCAB_FILES_MAP
		lowerCAmelCase_       :	Dict 							=     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
		lowerCAmelCase_       :	Dict 							=     ["""input_ids""", """attention_mask"""]



		def __init__( self  :    Tuple      ,							snake_case  :    Optional[Any]      ,							snake_case  :    Union[str, Any]      ,							snake_case  :    List[Any]="replace"      ,							snake_case  :    List[str]="<s>"      ,							snake_case  :    str="</s>"      ,							snake_case  :    str="</s>"      ,							snake_case  :    str="<s>"      ,							snake_case  :    List[str]="<unk>"      ,							snake_case  :    List[str]="<pad>"      ,							snake_case  :    Optional[Any]="<mask>"      ,							snake_case  :    str=False      ,							**snake_case  :    Any      ,							)			->						str:






									'''simple docstring'''

									A         =					AddedToken(snake_case      ,							lstrip=snake_case      ,							rstrip=snake_case				) if isinstance(snake_case      ,							snake_case				) else bos_token
									A         =					AddedToken(snake_case      ,							lstrip=snake_case      ,							rstrip=snake_case				) if isinstance(snake_case      ,							snake_case				) else eos_token
									A         =					AddedToken(snake_case      ,							lstrip=snake_case      ,							rstrip=snake_case				) if isinstance(snake_case      ,							snake_case				) else sep_token
									A         =					AddedToken(snake_case      ,							lstrip=snake_case      ,							rstrip=snake_case				) if isinstance(snake_case      ,							snake_case				) else cls_token
									A         =					AddedToken(snake_case      ,							lstrip=snake_case      ,							rstrip=snake_case				) if isinstance(snake_case      ,							snake_case				) else unk_token
									A         =					AddedToken(snake_case      ,							lstrip=snake_case      ,							rstrip=snake_case				) if isinstance(snake_case      ,							snake_case				) else pad_token

									# Mask token behave like a normal word, i.e. include the space before it
									A         =					AddedToken(snake_case      ,							lstrip=snake_case      ,							rstrip=snake_case				) if isinstance(snake_case      ,							snake_case				) else mask_token

									super().__init__(
									    errors=snake_case      ,							bos_token=snake_case      ,							eos_token=snake_case      ,							unk_token=snake_case      ,							sep_token=snake_case      ,							cls_token=snake_case      ,							pad_token=snake_case      ,							mask_token=snake_case      ,							add_prefix_space=snake_case      ,							**snake_case      ,							)

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

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



		@property
		# Copied from transformers.models.roberta.tokenization_roberta.RobertaTokenizer.vocab_size with Roberta->Blenderbot, RoBERTa->Blenderbot
		def 					A_		( self  :    Optional[Any]				)			->						List[Any]:






									'''simple docstring'''

									return len(self.encoder				)



		def 					A_		( self  :    str				)			->						int:






									'''simple docstring'''

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



		def 					A_		( self  :    Any      ,							snake_case  :    List[str]				)			->						List[Any]:






									'''simple docstring'''

									if token in self.cache:
																return self.cache[token]
									A         =					tuple(snake_case				)
									A         =					get_pairs(snake_case				)

									if not pairs:
																return token

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

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



		def 					A_		( self  :    List[str]      ,							snake_case  :    Dict				)			->						List[Any]:






									'''simple docstring'''

									A         =					[]
									for token in re.findall(self.pat      ,							snake_case				):
																A         =					''.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(snake_case				).split(' '				)				)
									return bpe_tokens



		def 					A_		( self  :    int      ,							snake_case  :    Any				)			->						Optional[Any]:






									'''simple docstring'''

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



		def 					A_		( self  :    Union[str, Any]      ,							snake_case  :    Optional[int]				)			->						List[str]:






									'''simple docstring'''

									return self.decoder.get(snake_case				)



		def 					A_		( self  :    Optional[Any]      ,							snake_case  :    Any				)			->						Optional[int]:






									'''simple docstring'''

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



		def 					A_		( self  :    List[str]      ,							snake_case  :    str      ,							snake_case  :    Optional[str] = None				)			->						Tuple[str]:






									'''simple docstring'''

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

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

									A         =					0
									with open(snake_case      ,							'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 snake_case				: 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!'				)
																														A         =					token_index
																							writer.write(' '.join(snake_case				) + '\n'				)
																							index += 1

									return vocab_file, merge_file



		def 					A_		( self  :    Optional[int]      ,							snake_case  :    List[int]      ,							snake_case  :    Optional[List[int]] = None      ,							snake_case  :    bool = False				)			->						List[int]:






									'''simple docstring'''

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

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



		def 					A_		( self  :    Any      ,							snake_case  :    List[int]      ,							snake_case  :    Optional[List[int]] = None				)			->						List[int]:






									'''simple docstring'''

									A         =					[self.sep_token_id]
									A         =					[self.cls_token_id]

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



		def 					A_		( self  :    Union[str, Any]      ,							snake_case  :    Optional[int]      ,							snake_case  :    List[str]=False      ,							**snake_case  :    Union[str, Any]				)			->						Union[str, Any]:






									'''simple docstring'''

									A         =					kwargs.pop('add_prefix_space'      ,							self.add_prefix_space				)
									if (is_split_into_words or add_prefix_space) and (len(snake_case				) > 0 and not text[0].isspace()):
																A         =					' ' + text
									return (text, kwargs)



		def 					A_		( self  :    Any      ,							snake_case  :    List[int]      ,							snake_case  :    Optional[List[int]] = None				)			->						Dict:






									'''simple docstring'''

									return token_ids_a + [self.eos_token_id]



		def 					A_		( self  :    Optional[Any]      ,							snake_case  :    "Conversation"				)			->						List[int]:






									'''simple docstring'''

									A         =					[]
									for is_user, text in conversation.iter_texts():
																if is_user:
																							# We need to space prefix as it's being done within blenderbot
																							inputs.append(' ' + text				)
																else:
																							# Generated responses should contain them already.
																							inputs.append(snake_case				)

									A         =					'  '.join(snake_case				)
									A         =					self.encode(snake_case				)
									if len(snake_case				) > self.model_max_length:
																A         =					input_ids[-self.model_max_length :]
																logger.warning(f"""Trimmed input from conversation as it was longer than {self.model_max_length} tokens."""				)
									return input_ids








"""simple docstring"""







import unittest
from pathlib import Path
from tempfile import TemporaryDirectory

from transformers import AutoConfig, TFAutoModel, is_tensorflow_text_available, is_tf_available
from transformers.models.bert.tokenization_bert import BertTokenizer
from transformers.testing_utils import require_tensorflow_text, require_tf, slow


if is_tf_available():
							import tensorflow as tf

if is_tensorflow_text_available():
							from transformers.models.bert import TFBertTokenizer


A    =			['bert-base-uncased', 'bert-base-cased']
A    =			'hf-internal-testing/tiny-bert-tf-only'

if is_tf_available():






							class      UpperCAmelCase__		(      tf.keras.Model     ):



									def __init__( self  :    Tuple      ,							snake_case  :    Optional[int]				)			->						List[str]:






																'''simple docstring'''

																super().__init__()
																A         =					tokenizer
																A         =					AutoConfig.from_pretrained(snake_case				)
																A         =					TFAutoModel.from_config(snake_case				)



									def 					A_		( self  :    Any      ,							snake_case  :    Optional[int]				)			->						Dict:






																'''simple docstring'''

																A         =					self.tokenizer(snake_case				)
																A         =					self.bert(**snake_case				)
																return out["pooler_output"]






@require_tf
@require_tensorflow_text
class      UpperCAmelCase__		(      unittest.TestCase     ):



		def 					A_		( self  :    int				)			->						Tuple:






									'''simple docstring'''

									super().setUp()

									A         =					[
									    BertTokenizer.from_pretrained(snake_case				) for checkpoint in (TOKENIZER_CHECKPOINTS * 2)
									]  # repeat for when fast_bert_tokenizer=false
									A         =					[TFBertTokenizer.from_pretrained(snake_case				) for checkpoint in TOKENIZER_CHECKPOINTS] + [
									    TFBertTokenizer.from_pretrained(snake_case      ,							use_fast_bert_tokenizer=snake_case				)
									    for checkpoint in TOKENIZER_CHECKPOINTS
									]
									assert len(self.tokenizers				) == len(self.tf_tokenizers				)

									A         =					[
									    'This is a straightforward English test sentence.',
									    'This one has some weird characters\rto\nsee\r\nif  those\u00E9break things.',
									    'Now we\'re going to add some Chinese: 一 二 三 一二三',
									    'And some much more rare Chinese: 齉 堃 齉堃',
									    'Je vais aussi écrire en français pour tester les accents',
									    'Classical Irish also has some unusual characters, so in they go: Gaelaċ, ꝼ',
									]
									A         =					list(zip(self.test_sentences      ,							self.test_sentences[::-1]				)				)



		def 					A_		( self  :    List[Any]				)			->						Dict:






									'''simple docstring'''

									for tokenizer, tf_tokenizer in zip(self.tokenizers      ,							self.tf_tokenizers				):
																for test_inputs in (self.test_sentences, self.paired_sentences):
																							A         =					tokenizer(snake_case      ,							return_tensors='tf'      ,							padding='longest'				)
																							A         =					tf_tokenizer(snake_case				)

																							for key in python_outputs.keys():
																														self.assertTrue(tf.reduce_all(python_outputs[key].shape == tf_outputs[key].shape				)				)
																														self.assertTrue(tf.reduce_all(tf.cast(python_outputs[key]      ,							tf.intaa				) == tf_outputs[key]				)				)



		@slow
		def 					A_		( self  :    Tuple				)			->						Tuple:






									'''simple docstring'''

									for tf_tokenizer in self.tf_tokenizers:
																A         =					tf_tokenizer(self.paired_sentences				)
																A         =					tf_tokenizer(
																    text=[sentence[0] for sentence in self.paired_sentences]      ,							text_pair=[sentence[1] for sentence in self.paired_sentences]      ,							)
																for key in merged_outputs.keys():
																							self.assertTrue(tf.reduce_all(tf.cast(merged_outputs[key]      ,							tf.intaa				) == separated_outputs[key]				)				)



		@slow
		def 					A_		( self  :    Any				)			->						Tuple:






									'''simple docstring'''

									for tf_tokenizer in self.tf_tokenizers:
																A         =					tf.function(snake_case				)
																for test_inputs in (self.test_sentences, self.paired_sentences):
																							A         =					tf.constant(snake_case				)
																							A         =					compiled_tokenizer(snake_case				)
																							A         =					tf_tokenizer(snake_case				)

																							for key in eager_outputs.keys():
																														self.assertTrue(tf.reduce_all(eager_outputs[key] == compiled_outputs[key]				)				)



		@slow
		def 					A_		( self  :    Optional[Any]				)			->						Optional[Any]:






									'''simple docstring'''

									for tf_tokenizer in self.tf_tokenizers:
																A         =					ModelToSave(tokenizer=snake_case				)
																A         =					tf.convert_to_tensor(self.test_sentences				)
																A         =					model(snake_case				)  # Build model with some sample inputs
																with TemporaryDirectory() as tempdir:
																							A         =					Path(snake_case				) / 'saved.model'
																							model.save(snake_case				)
																							A         =					tf.keras.models.load_model(snake_case				)
																A         =					loaded_model(snake_case				)
																# We may see small differences because the loaded model is compiled, so we need an epsilon for the test
																self.assertLessEqual(tf.reduce_max(tf.abs(out - loaded_output				)				)      ,							1E-5				)



