




"""simple docstring"""




import json
import os
import re
import shutil
import tempfile
import unittest
from typing import Tuple

from transformers import AddedToken, BatchEncoding, PerceiverTokenizer
from transformers.utils import cached_property, is_tf_available, is_torch_available

from ...test_tokenization_common import TokenizerTesterMixin


if is_torch_available():
					lowerCAmelCase				:     str					=				"""pt"""
elif is_tf_available():
					lowerCAmelCase				:     Optional[int]					=				"""tf"""
else:
					lowerCAmelCase				:     Optional[Any]					=				"""jax"""






class        __magic_name__				(			UpperCAmelCase__		,   unittest.TestCase							):






				'''simple docstring'''
				__UpperCamelCase         =		PerceiverTokenizer
				__UpperCamelCase         =		False





				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							super().setUp()
							lowerCamelCase      =   PerceiverTokenizer()
							tokenizer.save_pretrained(self.tmpdirname					)





				@cached_property
				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							return PerceiverTokenizer.from_pretrained("""deepmind/language-perceiver"""					)





				def   _lowerCAmelCase							(			self						,					**_a					):





							"""simple docstring"""
							return self.tokenizer_class.from_pretrained(self.tmpdirname						,					**_a					)





				def   _lowerCAmelCase							(			self						,					_a						,					_a=False						,					_a=20						,					_a=5					):





							"""simple docstring"""
							# XXX The default common tokenizer tests assume that every ID is decodable on its own.
							# This assumption is invalid for Perceiver because single bytes might not be
							# valid utf-8 (byte 128 for instance).
							# Here we're overriding the smallest possible method to provide
							# a clean sequence without making the same assumption.

							lowerCamelCase      =   []
							for i in range(len(_a					)					):
										try:
													lowerCamelCase      =   tokenizer.decode([i]						,					clean_up_tokenization_spaces=_a					)
										except UnicodeDecodeError:
													pass
										toks.append((i, tok)					)

							lowerCamelCase      =   list(filter(lambda _a					: re.match(r"""^[ a-zA-Z]+$"""						,					t[1]					)						,					_a					)					)
							lowerCamelCase      =   list(filter(lambda _a					: [t[0]] == tokenizer.encode(t[1]						,					add_special_tokens=_a					)						,					_a					)					)
							if max_length is not None and len(_a					) > max_length:
										lowerCamelCase      =   toks[:max_length]
							if min_length is not None and len(_a					) < min_length and len(_a					) > 0:
										while len(_a					) < min_length:
													lowerCamelCase      =   toks + toks
        # toks_str = [t[1] for t in toks]
							lowerCamelCase      =   [t[0] for t in toks]

							# Ensure consistency
							lowerCamelCase      =   tokenizer.decode(_a						,					clean_up_tokenization_spaces=_a					)
							if " " not in output_txt and len(_a					) > 1:
										lowerCamelCase      =   (
										    tokenizer.decode([toks_ids[0]]						,					clean_up_tokenization_spaces=_a					)
										    + """ """
										    + tokenizer.decode(toks_ids[1:]						,					clean_up_tokenization_spaces=_a					)
										)
							if with_prefix_space:
										lowerCamelCase      =   """ """ + output_txt
							lowerCamelCase      =   tokenizer.encode(_a						,					add_special_tokens=_a					)
							return output_txt, output_ids





				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							lowerCamelCase      =   self.perceiver_tokenizer
							lowerCamelCase      =   """Unicode €."""
							lowerCamelCase      =   tokenizer(_a					)
							lowerCamelCase      =   [4, 91, 116, 111, 105, 117, 106, 107, 38, 232, 136, 178, 52, 5]
							self.assertEqual(encoded["""input_ids"""]						,					_a					)

							# decoding
							lowerCamelCase      =   tokenizer.decode(_a					)
							self.assertEqual(_a						,					"""[CLS]Unicode €.[SEP]"""					)

							lowerCamelCase      =   tokenizer("""e è é ê ë"""					)
							lowerCamelCase      =   [4, 107, 38, 201, 174, 38, 201, 175, 38, 201, 176, 38, 201, 177, 5]
							self.assertEqual(encoded["""input_ids"""]						,					_a					)
							# decoding
							lowerCamelCase      =   tokenizer.decode(_a					)
							self.assertEqual(_a						,					"""[CLS]e è é ê ë[SEP]"""					)

							# encode/decode, but with `encode` instead of `__call__`
							self.assertEqual(tokenizer.decode(tokenizer.encode("""e è é ê ë"""					)					)						,					"""[CLS]e è é ê ë[SEP]"""					)





				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							lowerCamelCase      =   self.perceiver_tokenizer
							lowerCamelCase      =   ["""A long paragraph for summarization.""", """Another paragraph for summarization."""]
							# fmt: off
							lowerCamelCase      =   [4, 71, 38, 114, 117, 116, 109, 38, 118, 103, 120, 103, 109, 120, 103, 118, 110, 38, 108, 117, 120, 38, 121, 123, 115, 115, 103, 120, 111, 128, 103, 122, 111, 117, 116, 52, 5, 0]
							# fmt: on
							lowerCamelCase      =   tokenizer(_a						,					padding=_a						,					return_tensors=_a					)
							self.assertIsInstance(_a						,					_a					)

							if FRAMEWORK != "jax":
										lowerCamelCase      =   list(batch.input_ids.numpy()[0]					)
							else:
										lowerCamelCase      =   list(batch.input_ids.tolist()[0]					)

							self.assertListEqual(_a						,					_a					)

							self.assertEqual((2, 38)						,					batch.input_ids.shape					)
							self.assertEqual((2, 38)						,					batch.attention_mask.shape					)





				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							lowerCamelCase      =   self.perceiver_tokenizer
							lowerCamelCase      =   ["""A long paragraph for summarization.""", """Another paragraph for summarization."""]
							lowerCamelCase      =   tokenizer(_a						,					padding=_a						,					return_tensors=_a					)
							# check if input_ids are returned and no decoder_input_ids
							self.assertIn("""input_ids"""						,					_a					)
							self.assertIn("""attention_mask"""						,					_a					)
							self.assertNotIn("""decoder_input_ids"""						,					_a					)
							self.assertNotIn("""decoder_attention_mask"""						,					_a					)





				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							lowerCamelCase      =   self.perceiver_tokenizer
							lowerCamelCase      =   [
							    """Summary of the text.""",
							    """Another summary.""",
							]
							lowerCamelCase      =   tokenizer(
							    text_target=_a						,					max_length=32						,					padding="""max_length"""						,					truncation=_a						,					return_tensors=_a					)
							self.assertEqual(32						,					targets["""input_ids"""].shape[1]					)





				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							# safety check on max_len default value so we are sure the test works
							lowerCamelCase      =   self.get_tokenizers()
							for tokenizer in tokenizers:
										with self.subTest(f'{tokenizer.__class__.__name__}'					):
													self.assertNotEqual(tokenizer.model_max_length						,					42					)

        # Now let's start the test
							lowerCamelCase      =   self.get_tokenizers()
							for tokenizer in tokenizers:
										with self.subTest(f'{tokenizer.__class__.__name__}'					):
													# Isolate this from the other tests because we save additional tokens/etc
													lowerCamelCase      =   tempfile.mkdtemp()

													lowerCamelCase      =   """ He is very happy, UNwant\u00E9d,running"""
													lowerCamelCase      =   tokenizer.encode(_a						,					add_special_tokens=_a					)
													tokenizer.save_pretrained(_a					)

													lowerCamelCase      =   tokenizer.__class__.from_pretrained(_a					)
													lowerCamelCase      =   after_tokenizer.encode(_a						,					add_special_tokens=_a					)
													self.assertListEqual(_a						,					_a					)

													shutil.rmtree(_a					)

							lowerCamelCase      =   self.get_tokenizers(model_max_length=42					)
							for tokenizer in tokenizers:
										with self.subTest(f'{tokenizer.__class__.__name__}'					):
													# Isolate this from the other tests because we save additional tokens/etc
													lowerCamelCase      =   tempfile.mkdtemp()

													lowerCamelCase      =   """ He is very happy, UNwant\u00E9d,running"""
													tokenizer.add_tokens(["""bim""", """bambam"""]					)
													lowerCamelCase      =   tokenizer.additional_special_tokens
													additional_special_tokens.append("""new_additional_special_token"""					)
													tokenizer.add_special_tokens({"""additional_special_tokens""": additional_special_tokens}					)
													lowerCamelCase      =   tokenizer.encode(_a						,					add_special_tokens=_a					)
													tokenizer.save_pretrained(_a					)

													lowerCamelCase      =   tokenizer.__class__.from_pretrained(_a					)
													lowerCamelCase      =   after_tokenizer.encode(_a						,					add_special_tokens=_a					)
													self.assertListEqual(_a						,					_a					)
													self.assertIn("""new_additional_special_token"""						,					after_tokenizer.additional_special_tokens					)
													self.assertEqual(after_tokenizer.model_max_length						,					42					)

													lowerCamelCase      =   tokenizer.__class__.from_pretrained(_a						,					model_max_length=43					)
													self.assertEqual(tokenizer.model_max_length						,					43					)

													shutil.rmtree(_a					)





				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							lowerCamelCase      =   []
							if self.test_slow_tokenizer:
										tokenizer_list.append((self.tokenizer_class, self.get_tokenizer())					)

							if self.test_rust_tokenizer:
										tokenizer_list.append((self.rust_tokenizer_class, self.get_rust_tokenizer())					)

							for tokenizer_class, tokenizer_utils in tokenizer_list:
										with tempfile.TemporaryDirectory() as tmp_dir:
													tokenizer_utils.save_pretrained(_a					)

													with open(os.path.join(_a						,					"""special_tokens_map.json"""					)						,					encoding="""utf-8"""					) as json_file:
																lowerCamelCase      =   json.load(_a					)

													with open(os.path.join(_a						,					"""tokenizer_config.json"""					)						,					encoding="""utf-8"""					) as json_file:
																lowerCamelCase      =   json.load(_a					)

													lowerCamelCase      =   [f'<extra_id_{i}>' for i in range(125					)]

													lowerCamelCase      =   added_tokens_extra_ids + [
													    """an_additional_special_token"""
													]
													lowerCamelCase      =   added_tokens_extra_ids + [
													    """an_additional_special_token"""
													]

													with open(os.path.join(_a						,					"""special_tokens_map.json"""					)						,					"""w"""						,					encoding="""utf-8"""					) as outfile:
																json.dump(_a						,					_a					)
													with open(os.path.join(_a						,					"""tokenizer_config.json"""					)						,					"""w"""						,					encoding="""utf-8"""					) as outfile:
																json.dump(_a						,					_a					)

													# the following checks allow us to verify that our test works as expected, i.e. that the tokenizer takes
													# into account the new value of additional_special_tokens given in the "tokenizer_config.json" and
													# "special_tokens_map.json" files
													lowerCamelCase      =   tokenizer_class.from_pretrained(
													    _a						,					)
													self.assertIn(
													    """an_additional_special_token"""						,					tokenizer_without_change_in_init.additional_special_tokens					)
													self.assertEqual(
													    ["""an_additional_special_token"""]						,					tokenizer_without_change_in_init.convert_ids_to_tokens(
													        tokenizer_without_change_in_init.convert_tokens_to_ids(["""an_additional_special_token"""]					)					)						,					)

													# Now we test that we can change the value of additional_special_tokens in the from_pretrained
													lowerCamelCase      =   added_tokens_extra_ids + [AddedToken("""a_new_additional_special_token"""						,					lstrip=_a					)]
													lowerCamelCase      =   tokenizer_class.from_pretrained(
													    _a						,					additional_special_tokens=_a						,					)

													self.assertIn("""a_new_additional_special_token"""						,					tokenizer.additional_special_tokens					)
													self.assertEqual(
													    ["""a_new_additional_special_token"""]						,					tokenizer.convert_ids_to_tokens(
													        tokenizer.convert_tokens_to_ids(["""a_new_additional_special_token"""]					)					)						,					)





				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							lowerCamelCase      =   self.perceiver_tokenizer
							self.assertEqual(tokenizer.decode([178]					)						,					"""�"""					)





				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							pass





				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							pass





				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							pass





				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							pass





				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							# The default common tokenizer tests uses invalid tokens for Perceiver that can only accept one-character
							# strings and special added tokens as tokens
							lowerCamelCase      =   self.get_tokenizers(fast=_a						,					do_lower_case=_a					)
							for tokenizer in tokenizers:
										with self.subTest(f'{tokenizer.__class__.__name__}'					):
													lowerCamelCase      =   ["""[CLS]""", """t""", """h""", """i""", """s""", """ """, """i""", """s""", """ """, """a""", """ """, """t""", """e""", """s""", """t""", """[SEP]"""]
													lowerCamelCase      =   tokenizer.convert_tokens_to_string(_a					)

													self.assertIsInstance(_a						,					_a					)









"""simple docstring"""




import json
import os
from typing import Optional, Tuple

import regex as re

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


lowerCAmelCase				:     Union[str, Any]					=				logging.get_logger(__name__)

lowerCAmelCase				:     Optional[int]					=				{
    """vocab_file""": """vocab.json""",
    """merges_file""": """merges.txt""",
}

lowerCAmelCase				:     Optional[Any]					=				{
    """vocab_file""": {"""ctrl""": """https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-vocab.json"""},
    """merges_file""": {"""ctrl""": """https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-merges.txt"""},
}

lowerCAmelCase				:     Optional[Any]					=				{
    """ctrl""": 256,
}

lowerCAmelCase				:     int					=				{
    """Pregnancy""": 16_8629,
    """Christianity""": 7675,
    """Explain""": 10_6423,
    """Fitness""": 6_3440,
    """Saving""": 6_3163,
    """Ask""": 2_7171,
    """Ass""": 9_5985,
    """Joke""": 16_3509,
    """Questions""": 4_5622,
    """Thoughts""": 4_9605,
    """Retail""": 5_2342,
    """Feminism""": 16_4338,
    """Writing""": 1_1992,
    """Atheism""": 19_2263,
    """Netflix""": 4_8616,
    """Computing""": 3_9639,
    """Opinion""": 4_3213,
    """Alone""": 4_4967,
    """Funny""": 5_8917,
    """Gaming""": 4_0358,
    """Human""": 4088,
    """India""": 1331,
    """Joker""": 7_7138,
    """Diet""": 3_6206,
    """Legal""": 1_1859,
    """Norman""": 4939,
    """Tip""": 7_2689,
    """Weight""": 5_2343,
    """Movies""": 4_6273,
    """Running""": 2_3425,
    """Science""": 2090,
    """Horror""": 3_7793,
    """Confession""": 6_0572,
    """Finance""": 1_2250,
    """Politics""": 1_6360,
    """Scary""": 19_1985,
    """Support""": 1_2654,
    """Technologies""": 3_2516,
    """Teenage""": 6_6160,
    """Event""": 3_2769,
    """Learned""": 6_7460,
    """Notion""": 18_2770,
    """Wikipedia""": 3_7583,
    """Books""": 6665,
    """Extract""": 7_6050,
    """Confessions""": 10_2701,
    """Conspiracy""": 7_5932,
    """Links""": 6_3674,
    """Narcissus""": 15_0425,
    """Relationship""": 5_4766,
    """Relationships""": 13_4796,
    """Reviews""": 4_1671,
    """News""": 4256,
    """Translation""": 2_6820,
    """multilingual""": 12_8406,
}



def        a__					(							snake_case__							)		->   Optional[Any]:
			lowerCamelCase      =   set()
			lowerCamelCase      =   word[0]
			for char in word[1:]:
						pairs.add((prev_char, char)							)
						lowerCamelCase      =   char

			lowerCamelCase      =   set(snake_case__							)
			return pairs






class        __magic_name__				(			UpperCAmelCase__							):






				'''simple docstring'''
				__UpperCamelCase         =		VOCAB_FILES_NAMES
				__UpperCamelCase         =		PRETRAINED_VOCAB_FILES_MAP
				__UpperCamelCase         =		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				__UpperCamelCase         =		CONTROL_CODES





				def __init__(			self						,					_a						,					_a						,					_a="<unk>"						,					**_a					):





							"""simple docstring"""
							super().__init__(unk_token=_a						,					**_a					)

							with open(_a						,					encoding="""utf-8"""					) as vocab_handle:
										lowerCamelCase      =   json.load(_a					)
							lowerCamelCase      =   {v: k for k, v in self.encoder.items()}
							with open(_a						,					encoding="""utf-8"""					) as merges_handle:
										lowerCamelCase      =   merges_handle.read().split("""\n"""					)[1:-1]
							lowerCamelCase      =   [tuple(merge.split()					) for merge in merges]
							lowerCamelCase      =   dict(zip(_a						,					range(len(_a					)					)					)					)
							lowerCamelCase      =   {}





				@property
				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							return len(self.encoder					)





				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							return dict(self.encoder						,					**self.added_tokens_encoder					)





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							if token in self.cache:
										return self.cache[token]
							lowerCamelCase      =   tuple(_a					)
							lowerCamelCase      =   tuple(list(word[:-1]					) + [word[-1] + """</w>"""]					)
							lowerCamelCase      =   get_pairs(_a					)

							if not pairs:
										return token

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

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





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							lowerCamelCase      =   []

							lowerCamelCase      =   re.findall(r"""\S+\n?"""						,					_a					)

							for token in words:
										split_tokens.extend(list(self.bpe(_a					).split(""" """					)					)					)
							return split_tokens





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							return self.encoder.get(_a						,					self.encoder.get(self.unk_token					)					)





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							return self.decoder.get(_a						,					self.unk_token					)





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							lowerCamelCase      =   """ """.join(_a					).replace("""@@ """						,					""""""					).strip()
							return out_string





				def   _lowerCAmelCase							(			self						,					_a						,					_a = None					):





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

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

							lowerCamelCase      =   0
							with open(_a						,					"""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 _a					: 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(_a					) + """\n"""					)
													index += 1

							return vocab_file, merge_file

				# def decode(self, token_ids, skip_special_tokens=False, clean_up_tokenization_spaces=True):
				#     filtered_tokens = ' '.join(self.convert_ids_to_tokens(token_ids, skip_special_tokens=skip_special_tokens))
				#     tokens_generated_so_far = re.sub('(@@ )', '', string=filtered_tokens)
				#     tokens_generated_so_far = re.sub('(@@ ?$)', '', string=tokens_generated_so_far)
				#     return ''.join(tokens_generated_so_far)



