






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""": 128}


@lru_cache()
# Copied from transformers.models.roberta.tokenization_roberta.bytes_to_unicode
def 				_a     (   )       ->      Optional[Any]:
						"""simple docstring"""







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


def 				_a     (   UpperCamelCase_    : Any      )       ->      Optional[int]:
						"""simple docstring"""







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





class    lowercase__   (	lowercase_		):
	a_   			=VOCAB_FILES_NAMES
	a_   			=PRETRAINED_VOCAB_FILES_MAP
	a_   			=PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	a_   			=['''input_ids''', '''attention_mask''']

	def __init__(							self						,			__UpperCAmelCase						,			__UpperCAmelCase						,			__UpperCAmelCase="replace"						,			__UpperCAmelCase="<s>"						,			__UpperCAmelCase="</s>"						,			__UpperCAmelCase="</s>"						,			__UpperCAmelCase="<s>"						,			__UpperCAmelCase="<unk>"						,			__UpperCAmelCase="<pad>"						,			__UpperCAmelCase="<mask>"						,			__UpperCAmelCase=False						,			**__UpperCAmelCase						,			)->    str:







							'''simple docstring'''





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

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

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

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

							# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
							lowerCAmelCase__       					=  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 						UpperCAmelCase     (							self				)->    List[Any]:







							'''simple docstring'''





							return len(self.encoder				)

	def 						UpperCAmelCase     (							self				)->    Any:







							'''simple docstring'''





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

	def 						UpperCAmelCase     (							self						,			__UpperCAmelCase				)->    Tuple:







							'''simple docstring'''





							if token in self.cache:
													return self.cache[token]
							lowerCAmelCase__       					=  tuple(__UpperCAmelCase				)
							lowerCAmelCase__       					=  get_pairs(__UpperCAmelCase				)

							if not pairs:
													return token

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

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

	def 						UpperCAmelCase     (							self						,			__UpperCAmelCase				)->    List[str]:







							'''simple docstring'''





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

	def 						UpperCAmelCase     (							self						,			__UpperCAmelCase				)->    int:







							'''simple docstring'''





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

	def 						UpperCAmelCase     (							self						,			__UpperCAmelCase				)->    Optional[int]:







							'''simple docstring'''





							return self.decoder.get(__UpperCAmelCase				)

	def 						UpperCAmelCase     (							self						,			__UpperCAmelCase				)->    int:







							'''simple docstring'''





							lowerCAmelCase__       					=  "".join(__UpperCAmelCase				)
							lowerCAmelCase__       					=  bytearray([self.byte_decoder[c] for c in text]				).decode("utf-8"						,			errors=self.errors				)
							return text

	def 						UpperCAmelCase     (							self						,			__UpperCAmelCase						,			__UpperCAmelCase = None				)->    Optional[int]:







							'''simple docstring'''





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

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

							lowerCAmelCase__       					=  0
							with open(__UpperCAmelCase						,			"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!"				)
																									lowerCAmelCase__       					=  token_index
																			writer.write(" ".join(__UpperCAmelCase				) + "\n"				)
																			index += 1

							return vocab_file, merge_file

	def 						UpperCAmelCase     (							self						,			__UpperCAmelCase						,			__UpperCAmelCase = None						,			__UpperCAmelCase = False				)->    Union[str, Any]:







							'''simple docstring'''





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

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

	def 						UpperCAmelCase     (							self						,			__UpperCAmelCase						,			__UpperCAmelCase = None				)->    List[str]:







							'''simple docstring'''





							lowerCAmelCase__       					=  [self.sep_token_id]
							lowerCAmelCase__       					=  [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 						UpperCAmelCase     (							self						,			__UpperCAmelCase						,			__UpperCAmelCase=False						,			**__UpperCAmelCase				)->    List[Any]:







							'''simple docstring'''





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

	def 						UpperCAmelCase     (							self						,			__UpperCAmelCase						,			__UpperCAmelCase = None				)->    Optional[int]:







							'''simple docstring'''





							return token_ids_a + [self.eos_token_id]


	def 						UpperCAmelCase     (							self						,			__UpperCAmelCase				)->    Optional[Any]:







							'''simple docstring'''





							lowerCAmelCase__       					=  []
							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(__UpperCAmelCase				)

							lowerCAmelCase__       					=  "  ".join(__UpperCAmelCase				)
							lowerCAmelCase__       					=  self.encode(__UpperCAmelCase				)
							if len(__UpperCAmelCase				) > self.model_max_length:
													lowerCAmelCase__       					=  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









import json
import os
import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers import BertTokenizer, BertTokenizerFast
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES
from transformers.testing_utils import require_vision
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_vision_available


if is_vision_available():
					from PIL import Image

					from transformers import ChineseCLIPImageProcessor, ChineseCLIPProcessor







@require_vision
class  lowercase				(       unittest.TestCase			):




				def     a							(					self       ):
											snake_case_							=    tempfile.mkdtemp()

											snake_case_							=    [
											    '[UNK]',
											    '[CLS]',
											    '[SEP]',
											    '[PAD]',
											    '[MASK]',
											    '的',
											    '价',
											    '格',
											    '是',
											    '15',
											    '便',
											    'alex',
											    '##andra',
											    '，',
											    '。',
											    '-',
											    't',
											    'shirt',
											]
											snake_case_							=    os.path.join(self.tmpdirname      ,   VOCAB_FILES_NAMES['vocab_file']       )
											with open(self.vocab_file      ,   'w'      ,   encoding='utf-8'       ) as vocab_writer:
																		vocab_writer.write(''.join([x + '\n' for x in vocab_tokens]       )       )

											snake_case_							=    {
											    'do_resize': True,
											    'size': {'height': 224, 'width': 224},
											    'do_center_crop': True,
											    'crop_size': {'height': 18, 'width': 18},
											    'do_normalize': True,
											    'image_mean': [0.48_14_54_66, 0.4_57_82_75, 0.40_82_10_73],
											    'image_std': [0.26_86_29_54, 0.26_13_02_58, 0.27_57_77_11],
											    'do_convert_rgb': True,
											}
											snake_case_							=    os.path.join(self.tmpdirname      ,   snake_case       )
											with open(self.image_processor_file      ,   'w'      ,   encoding='utf-8'       ) as fp:
																		json.dump(snake_case      ,   snake_case       )




				def     a							(					self      ,   **snake_case       ):
											return BertTokenizer.from_pretrained(self.tmpdirname      ,   **snake_case       )




				def     a							(					self      ,   **snake_case       ):
											return BertTokenizerFast.from_pretrained(self.tmpdirname      ,   **snake_case       )




				def     a							(					self      ,   **snake_case       ):
											return ChineseCLIPImageProcessor.from_pretrained(self.tmpdirname      ,   **snake_case       )




				def     a							(					self       ):
											shutil.rmtree(self.tmpdirname       )




				def     a							(					self       ):
											snake_case_							=    [np.random.randint(255      ,   size=(3, 30, 400)      ,   dtype=np.uinta       )]

											snake_case_							=    [Image.fromarray(np.moveaxis(snake_case      ,   0      ,   -1       )       ) for x in image_inputs]

											return image_inputs




				def     a							(					self       ):
											snake_case_							=    self.get_tokenizer()
											snake_case_							=    self.get_rust_tokenizer()
											snake_case_							=    self.get_image_processor()

											snake_case_							=    ChineseCLIPProcessor(tokenizer=snake_case      ,   image_processor=snake_case       )
											processor_slow.save_pretrained(self.tmpdirname       )
											snake_case_							=    ChineseCLIPProcessor.from_pretrained(self.tmpdirname      ,   use_fast=snake_case       )

											snake_case_							=    ChineseCLIPProcessor(tokenizer=snake_case      ,   image_processor=snake_case       )
											processor_fast.save_pretrained(self.tmpdirname       )
											snake_case_							=    ChineseCLIPProcessor.from_pretrained(self.tmpdirname       )

											self.assertEqual(processor_slow.tokenizer.get_vocab()      ,   tokenizer_slow.get_vocab()       )
											self.assertEqual(processor_fast.tokenizer.get_vocab()      ,   tokenizer_fast.get_vocab()       )
											self.assertEqual(tokenizer_slow.get_vocab()      ,   tokenizer_fast.get_vocab()       )
											self.assertIsInstance(processor_slow.tokenizer      ,   snake_case       )
											self.assertIsInstance(processor_fast.tokenizer      ,   snake_case       )

											self.assertEqual(processor_slow.image_processor.to_json_string()      ,   image_processor.to_json_string()       )
											self.assertEqual(processor_fast.image_processor.to_json_string()      ,   image_processor.to_json_string()       )
											self.assertIsInstance(processor_slow.image_processor      ,   snake_case       )
											self.assertIsInstance(processor_fast.image_processor      ,   snake_case       )




				def     a							(					self       ):
											snake_case_							=    ChineseCLIPProcessor(tokenizer=self.get_tokenizer()      ,   image_processor=self.get_image_processor()       )
											processor.save_pretrained(self.tmpdirname       )

											snake_case_							=    self.get_tokenizer(cls_token='(CLS)'      ,   sep_token='(SEP)'       )
											snake_case_							=    self.get_image_processor(do_normalize=snake_case       )

											snake_case_							=    ChineseCLIPProcessor.from_pretrained(
											    self.tmpdirname      ,   cls_token='(CLS)'      ,   sep_token='(SEP)'      ,   do_normalize=snake_case       )

											self.assertEqual(processor.tokenizer.get_vocab()      ,   tokenizer_add_kwargs.get_vocab()       )
											self.assertIsInstance(processor.tokenizer      ,   snake_case       )

											self.assertEqual(processor.image_processor.to_json_string()      ,   image_processor_add_kwargs.to_json_string()       )
											self.assertIsInstance(processor.image_processor      ,   snake_case       )




				def     a							(					self       ):
											snake_case_							=    self.get_image_processor()
											snake_case_							=    self.get_tokenizer()

											snake_case_							=    ChineseCLIPProcessor(tokenizer=snake_case      ,   image_processor=snake_case       )

											snake_case_							=    self.prepare_image_inputs()

											snake_case_							=    image_processor(snake_case      ,   return_tensors='np'       )
											snake_case_							=    processor(images=snake_case      ,   return_tensors='np'       )

											for key in input_feat_extract.keys():
																		self.assertAlmostEqual(input_feat_extract[key].sum()      ,   input_processor[key].sum()      ,   delta=1e-2       )




				def     a							(					self       ):
											snake_case_							=    self.get_image_processor()
											snake_case_							=    self.get_tokenizer()

											snake_case_							=    ChineseCLIPProcessor(tokenizer=snake_case      ,   image_processor=snake_case       )

											snake_case_							=    'Alexandra，T-shirt的价格是15便士。'

											snake_case_							=    processor(text=snake_case       )

											snake_case_							=    tokenizer(snake_case       )

											for key in encoded_tok.keys():
																		self.assertListEqual(encoded_tok[key]      ,   encoded_processor[key]       )




				def     a							(					self       ):
											snake_case_							=    self.get_image_processor()
											snake_case_							=    self.get_tokenizer()

											snake_case_							=    ChineseCLIPProcessor(tokenizer=snake_case      ,   image_processor=snake_case       )

											snake_case_							=    'Alexandra，T-shirt的价格是15便士。'
											snake_case_							=    self.prepare_image_inputs()

											snake_case_							=    processor(text=snake_case      ,   images=snake_case       )

											self.assertListEqual(list(inputs.keys()       )      ,   ['input_ids', 'token_type_ids', 'attention_mask', 'pixel_values']       )

											# test if it raises when no input is passed
											with pytest.raises(snake_case       ):
																		processor()




				def     a							(					self       ):
											snake_case_							=    self.get_image_processor()
											snake_case_							=    self.get_tokenizer()

											snake_case_							=    ChineseCLIPProcessor(tokenizer=snake_case      ,   image_processor=snake_case       )

											snake_case_							=    [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

											snake_case_							=    processor.batch_decode(snake_case       )
											snake_case_							=    tokenizer.batch_decode(snake_case       )

											self.assertListEqual(snake_case      ,   snake_case       )




				def     a							(					self       ):
											snake_case_							=    self.get_image_processor()
											snake_case_							=    self.get_tokenizer()

											snake_case_							=    ChineseCLIPProcessor(tokenizer=snake_case      ,   image_processor=snake_case       )

											snake_case_							=    'Alexandra，T-shirt的价格是15便士。'
											snake_case_							=    self.prepare_image_inputs()

											snake_case_							=    processor(text=snake_case      ,   images=snake_case       )

											self.assertListEqual(list(inputs.keys()       )      ,   processor.model_input_names       )




