


"""simple docstring"""

from abc import ABC, abstractmethod
from argparse import ArgumentParser





class    lowerCamelCase__						(		_a    ):
      @staticmethod
      @abstractmethod
      def SCREAMING_SNAKE_CASE_	(		A_	:	ArgumentParser							):






       '''simple docstring'''


       raise NotImplementedError()






      @abstractmethod
      def SCREAMING_SNAKE_CASE_	(		self	:	Optional[int]							):






       '''simple docstring'''


       raise NotImplementedError()




"""simple docstring"""

import os
import shutil
import tempfile
from unittest import TestCase
from unittest.mock import patch

import numpy as np
from datasets import Dataset

from transformers.models.realm.configuration_realm import RealmConfig
from transformers.models.realm.retrieval_realm import _REALM_BLOCK_RECORDS_FILENAME, RealmRetriever
from transformers.models.realm.tokenization_realm import VOCAB_FILES_NAMES, RealmTokenizer





class    lowerCamelCase__						(		_a    ):
      def SCREAMING_SNAKE_CASE_	(		self	:	Optional[Any]							):






       '''simple docstring'''


       __lowercase									=	tempfile.mkdtemp()
       __lowercase									=	5

       # Realm tok
       __lowercase									=	[
           """[UNK]""",
           """[CLS]""",
           """[SEP]""",
           """[PAD]""",
           """[MASK]""",
           """test""",
           """question""",
           """this""",
           """is""",
           """the""",
           """first""",
           """second""",
           """third""",
           """fourth""",
           """fifth""",
           """record""",
           """want""",
           """##want""",
           """##ed""",
           """wa""",
           """un""",
           """runn""",
           """##ing""",
           """,""",
           """low""",
           """lowest""",
       ]
       __lowercase									=	os.path.join(self.tmpdirname      ,       """realm_tokenizer"""							)
       os.makedirs(A_      ,       exist_ok=A_							)
       __lowercase									=	os.path.join(A_      ,       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]							)							)

       __lowercase									=	os.path.join(self.tmpdirname      ,       """realm_block_records"""							)
       os.makedirs(A_      ,       exist_ok=A_							)
      def SCREAMING_SNAKE_CASE_	(		self	:	Optional[int]							):






       '''simple docstring'''


       return RealmTokenizer.from_pretrained(os.path.join(self.tmpdirname      ,       """realm_tokenizer"""							)							)
      def SCREAMING_SNAKE_CASE_	(		self	:	Optional[int]							):






       '''simple docstring'''


       shutil.rmtree(self.tmpdirname							)
      def SCREAMING_SNAKE_CASE_	(		self	:	int							):






       '''simple docstring'''


       __lowercase									=	RealmConfig(num_block_records=self.num_block_records							)
       return config
      def SCREAMING_SNAKE_CASE_	(		self	:	Any							):






       '''simple docstring'''


       __lowercase									=	Dataset.from_dict(
           {
               """id""": ["""0""", """1"""],
               """question""": ["""foo""", """bar"""],
               """answers""": [["""Foo""", """Bar"""], ["""Bar"""]],
           }							)
       return dataset
      def SCREAMING_SNAKE_CASE_	(		self	:	Optional[Any]							):






       '''simple docstring'''


       __lowercase									=	np.array(
           [
               B"""This is the first record""",
               B"""This is the second record""",
               B"""This is the third record""",
               B"""This is the fourth record""",
               B"""This is the fifth record""",
               B"""This is a longer longer longer record""",
           ]      ,       dtype=A_      ,       )
       return block_records
      def SCREAMING_SNAKE_CASE_	(		self	:	Union[str, Any]							):






       '''simple docstring'''


       __lowercase									=	RealmRetriever(
           block_records=self.get_dummy_block_records()      ,       tokenizer=self.get_tokenizer()      ,       )
       return retriever
      def SCREAMING_SNAKE_CASE_	(		self	:	Dict							):






       '''simple docstring'''


       __lowercase									=	self.get_config()
       __lowercase									=	self.get_dummy_retriever()
       __lowercase									=	retriever.tokenizer

       __lowercase									=	np.array([0, 3]      ,       dtype="""long"""							)
       __lowercase									=	tokenizer(["""Test question"""]							).input_ids
       __lowercase									=	tokenizer(
           ["""the fourth"""]      ,       add_special_tokens=A_      ,       return_token_type_ids=A_      ,       return_attention_mask=A_      ,       ).input_ids
       __lowercase									=	config.reader_seq_len

       __lowercase							,		__lowercase							,		__lowercase							,		__lowercase									=	retriever(
           A_      ,       A_      ,       answer_ids=A_      ,       max_length=A_      ,       return_tensors="""np"""							)

       self.assertEqual(len(A_							)      ,       2							)
       self.assertEqual(len(A_							)      ,       2							)
       self.assertEqual(len(A_							)      ,       2							)
       self.assertEqual(concat_inputs.input_ids.shape      ,       (2, 1_0)							)
       self.assertEqual(concat_inputs.attention_mask.shape      ,       (2, 1_0)							)
       self.assertEqual(concat_inputs.token_type_ids.shape      ,       (2, 1_0)							)
       self.assertEqual(concat_inputs.special_tokens_mask.shape      ,       (2, 1_0)							)
       self.assertEqual(
           tokenizer.convert_ids_to_tokens(concat_inputs.input_ids[0]							)      ,       ["""[CLS]""", """test""", """question""", """[SEP]""", """this""", """is""", """the""", """first""", """record""", """[SEP]"""]      ,       )
       self.assertEqual(
           tokenizer.convert_ids_to_tokens(concat_inputs.input_ids[1]							)      ,       ["""[CLS]""", """test""", """question""", """[SEP]""", """this""", """is""", """the""", """fourth""", """record""", """[SEP]"""]      ,       )
      def SCREAMING_SNAKE_CASE_	(		self	:	str							):






       '''simple docstring'''


       __lowercase									=	self.get_config()
       __lowercase									=	self.get_dummy_retriever()
       __lowercase									=	retriever.tokenizer

       __lowercase									=	np.array([0, 3, 5]      ,       dtype="""long"""							)
       __lowercase									=	tokenizer(["""Test question"""]							).input_ids
       __lowercase									=	tokenizer(
           ["""the fourth""", """longer longer"""]      ,       add_special_tokens=A_      ,       return_token_type_ids=A_      ,       return_attention_mask=A_      ,       ).input_ids
       __lowercase									=	config.reader_seq_len

       __lowercase							,		__lowercase							,		__lowercase							,		__lowercase									=	retriever(
           A_      ,       A_      ,       answer_ids=A_      ,       max_length=A_      ,       return_tensors="""np"""							)

       self.assertEqual([False, True, True]      ,       A_							)
       self.assertEqual([[-1, -1, -1], [6, -1, -1], [6, 7, 8]]      ,       A_							)
       self.assertEqual([[-1, -1, -1], [7, -1, -1], [7, 8, 9]]      ,       A_							)






      def SCREAMING_SNAKE_CASE_	(		self	:	Optional[Any]							):






       '''simple docstring'''


       __lowercase									=	self.get_dummy_retriever()
       retriever.save_pretrained(os.path.join(self.tmpdirname      ,       """realm_block_records"""							)							)

       # Test local path
       __lowercase									=	retriever.from_pretrained(os.path.join(self.tmpdirname      ,       """realm_block_records"""							)							)
       self.assertEqual(retriever.block_records[0]      ,       B"""This is the first record"""							)

       # Test mocked remote path
       with patch("""transformers.models.realm.retrieval_realm.hf_hub_download"""							) as mock_hf_hub_download:
        __lowercase									=	os.path.join(
            os.path.join(self.tmpdirname      ,       """realm_block_records"""							)      ,       _REALM_BLOCK_RECORDS_FILENAME							)
        __lowercase									=	RealmRetriever.from_pretrained("""google/realm-cc-news-pretrained-openqa"""							)

       self.assertEqual(retriever.block_records[0]      ,       B"""This is the first record"""							)
