
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():
   _A   =			"pt"
elif is_tf_available():
   _A   =			"tf"
else:
   _A   =			"jax"







class 					lowerCamelCase		(  A_							,			unittest.TestCase  ):
     UpperCAmelCase__    :							Any									=		PerceiverTokenizer
     UpperCAmelCase__    :							Union[str, Any]									=		False





     def 		UpperCAmelCase(self				:   int       )						->						Dict:
         super().setUp()
         snake_case								= PerceiverTokenizer()
         tokenizer.save_pretrained(self.tmpdirname       )





     @cached_property
     def 		UpperCAmelCase(self				:   Union[str, Any]       )						->						Union[str, Any]:
         return PerceiverTokenizer.from_pretrained("deepmind/language-perceiver"       )





     def 		UpperCAmelCase(self				:   str    ,   **_A				:   str       )						->						PerceiverTokenizer:
         return self.tokenizer_class.from_pretrained(self.tmpdirname    ,   **_A       )





     def 		UpperCAmelCase(self				:   str    ,   _A				:   Union[str, Any]    ,   _A				:   List[Any]=False    ,   _A				:   List[str]=2_0    ,   _A				:   str=5       )						->						Tuple[str, list]:
         # 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.

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

         snake_case								= list(filter(lambda _A       : re.match(r"^[ a-zA-Z]+$"    ,   t[1]       )    ,   _A       )       )
         snake_case								= 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:
             snake_case								= toks[:max_length]
         if min_length is not None and len(_A       ) < min_length and len(_A       ) > 0:
             while len(_A       ) < min_length:
                 snake_case								= toks + toks
        # toks_str = [t[1] for t in toks]
         snake_case								= [t[0] for t in toks]

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





     def 		UpperCAmelCase(self				:   Any       )						->						Tuple:
         snake_case								= self.perceiver_tokenizer
         snake_case								= "Unicode €."
         snake_case								= tokenizer(_A       )
         snake_case								= [4, 9_1, 1_1_6, 1_1_1, 1_0_5, 1_1_7, 1_0_6, 1_0_7, 3_8, 2_3_2, 1_3_6, 1_7_8, 5_2, 5]
         self.assertEqual(encoded["input_ids"]    ,   _A       )

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

         snake_case								= tokenizer("e è é ê ë"       )
         snake_case								= [4, 1_0_7, 3_8, 2_0_1, 1_7_4, 3_8, 2_0_1, 1_7_5, 3_8, 2_0_1, 1_7_6, 3_8, 2_0_1, 1_7_7, 5]
         self.assertEqual(encoded["input_ids"]    ,   _A       )
         # decoding
         snake_case								= 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 		UpperCAmelCase(self				:   Optional[Any]       )						->						Tuple:
         snake_case								= self.perceiver_tokenizer
         snake_case								= ["A long paragraph for summarization.", "Another paragraph for summarization."]
         # fmt: off
         snake_case								= [4, 7_1, 3_8, 1_1_4, 1_1_7, 1_1_6, 1_0_9, 3_8, 1_1_8, 1_0_3, 1_2_0, 1_0_3, 1_0_9, 1_2_0, 1_0_3, 1_1_8, 1_1_0, 3_8, 1_0_8, 1_1_7, 1_2_0, 3_8, 1_2_1, 1_2_3, 1_1_5, 1_1_5, 1_0_3, 1_2_0, 1_1_1, 1_2_8, 1_0_3, 1_2_2, 1_1_1, 1_1_7, 1_1_6, 5_2, 5, 0]
         # fmt: on
         snake_case								= tokenizer(_A    ,   padding=_A    ,   return_tensors=_A       )
         self.assertIsInstance(_A    ,   _A       )

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

         self.assertListEqual(_A    ,   _A       )

         self.assertEqual((2, 3_8)    ,   batch.input_ids.shape       )
         self.assertEqual((2, 3_8)    ,   batch.attention_mask.shape       )





     def 		UpperCAmelCase(self				:   Dict       )						->						Optional[int]:
         snake_case								= self.perceiver_tokenizer
         snake_case								= ["A long paragraph for summarization.", "Another paragraph for summarization."]
         snake_case								= 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 		UpperCAmelCase(self				:   Tuple       )						->						List[str]:
         snake_case								= self.perceiver_tokenizer
         snake_case								= [
             "Summary of the text.",
             "Another summary.",
         ]
         snake_case								= tokenizer(
             text_target=_A    ,   max_length=3_2    ,   padding="max_length"    ,   truncation=_A    ,   return_tensors=_A       )
         self.assertEqual(3_2    ,   targets["input_ids"].shape[1]       )





     def 		UpperCAmelCase(self				:   List[str]       )						->						Dict:
         # safety check on max_len default value so we are sure the test works
         snake_case								= self.get_tokenizers()
         for tokenizer in tokenizers:
             with self.subTest(f'{tokenizer.__class__.__name__}'       ):
                 self.assertNotEqual(tokenizer.model_max_length    ,   4_2       )

        # Now let's start the test
         snake_case								= 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
                 snake_case								= tempfile.mkdtemp()

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

                 snake_case								= tokenizer.__class__.from_pretrained(_A       )
                 snake_case								= after_tokenizer.encode(_A    ,   add_special_tokens=_A       )
                 self.assertListEqual(_A    ,   _A       )

                 shutil.rmtree(_A       )

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

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

                 snake_case								= tokenizer.__class__.from_pretrained(_A       )
                 snake_case								= 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    ,   4_2       )

                 snake_case								= tokenizer.__class__.from_pretrained(_A    ,   model_max_length=4_3       )
                 self.assertEqual(tokenizer.model_max_length    ,   4_3       )

                 shutil.rmtree(_A       )





     def 		UpperCAmelCase(self				:   Dict       )						->						Optional[Any]:
         snake_case								= []
         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:
                     snake_case								= json.load(_A       )

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

                 snake_case								= [f'<extra_id_{i}>' for i in range(1_2_5       )]

                 snake_case								= added_tokens_extra_ids + [
                     "an_additional_special_token"
                 ]
                 snake_case								= 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
                 snake_case								= 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
                 snake_case								= added_tokens_extra_ids + [AddedToken("a_new_additional_special_token"    ,   lstrip=_A       )]
                 snake_case								= 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 		UpperCAmelCase(self				:   List[str]       )						->						List[str]:
         snake_case								= self.perceiver_tokenizer
         self.assertEqual(tokenizer.decode([1_7_8]       )    ,   "�"       )





     def 		UpperCAmelCase(self				:   Dict       )						->						Union[str, Any]:
         pass





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





     def 		UpperCAmelCase(self				:   List[Any]       )						->						str:
         pass





     def 		UpperCAmelCase(self				:   str       )						->						Optional[Any]:
         pass





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

                 self.assertIsInstance(_A    ,   _A       )



from ...configuration_utils import PretrainedConfig
from ...utils import logging


_A   =			logging.get_logger(__name__)

_A   =			{
    "google/realm-cc-news-pretrained-embedder": (
        "https://huggingface.co/google/realm-cc-news-pretrained-embedder/resolve/main/config.json"
    ),
    "google/realm-cc-news-pretrained-encoder": (
        "https://huggingface.co/google/realm-cc-news-pretrained-encoder/resolve/main/config.json"
    ),
    "google/realm-cc-news-pretrained-scorer": (
        "https://huggingface.co/google/realm-cc-news-pretrained-scorer/resolve/main/config.json"
    ),
    "google/realm-cc-news-pretrained-openqa": (
        "https://huggingface.co/google/realm-cc-news-pretrained-openqa/aresolve/main/config.json"
    ),
    "google/realm-orqa-nq-openqa": "https://huggingface.co/google/realm-orqa-nq-openqa/resolve/main/config.json",
    "google/realm-orqa-nq-reader": "https://huggingface.co/google/realm-orqa-nq-reader/resolve/main/config.json",
    "google/realm-orqa-wq-openqa": "https://huggingface.co/google/realm-orqa-wq-openqa/resolve/main/config.json",
    "google/realm-orqa-wq-reader": "https://huggingface.co/google/realm-orqa-wq-reader/resolve/main/config.json",
    # See all REALM models at https://huggingface.co/models?filter=realm
}







class 					lowerCamelCase		(  A_  ):
     UpperCAmelCase__    :							str									=		"realm"





     def __init__(self				:   Optional[int]    ,   _A				:   Optional[Any]=3_0_5_2_2    ,   _A				:   Tuple=7_6_8    ,   _A				:   List[str]=1_2_8    ,   _A				:   Optional[Any]=1_2    ,   _A				:   Dict=1_2    ,   _A				:   Tuple=8    ,   _A				:   Dict=3_0_7_2    ,   _A				:   Union[str, Any]="gelu_new"    ,   _A				:   Any=0.1    ,   _A				:   int=0.1    ,   _A				:   Union[str, Any]=5_1_2    ,   _A				:   List[str]=2    ,   _A				:   Any=0.02    ,   _A				:   int=1E-12    ,   _A				:   Tuple=2_5_6    ,   _A				:   Optional[Any]=1_0    ,   _A				:   Any=1E-3    ,   _A				:   int=5    ,   _A				:   int=3_2_0    ,   _A				:   Dict=1_3_3_5_3_7_1_8    ,   _A				:   Any=5_0_0_0    ,   _A				:   Union[str, Any]=1    ,   _A				:   Dict=0    ,   _A				:   int=2    ,   **_A				:   Union[str, Any]    ,   )						->						Optional[Any]:
         super().__init__(pad_token_id=_A    ,   bos_token_id=_A    ,   eos_token_id=_A    ,   **_A       )

         # Common config
         snake_case								= vocab_size
         snake_case								= max_position_embeddings
         snake_case								= hidden_size
         snake_case								= retriever_proj_size
         snake_case								= num_hidden_layers
         snake_case								= num_attention_heads
         snake_case								= num_candidates
         snake_case								= intermediate_size
         snake_case								= hidden_act
         snake_case								= hidden_dropout_prob
         snake_case								= attention_probs_dropout_prob
         snake_case								= initializer_range
         snake_case								= type_vocab_size
         snake_case								= layer_norm_eps

         # Reader config
         snake_case								= span_hidden_size
         snake_case								= max_span_width
         snake_case								= reader_layer_norm_eps
         snake_case								= reader_beam_size
         snake_case								= reader_seq_len

         # Retrieval config
         snake_case								= num_block_records
         snake_case								= searcher_beam_size

