





'''simple docstring'''




import enum
import shutil
import sys


lowercase__	,							lowercase__    :      List[Any]						=    shutil.get_terminal_size()

lowercase__    :      Any						=    {'''UP''': '''A''', '''DOWN''': '''B''', '''RIGHT''': '''C''', '''LEFT''': '''D'''}


class  SCREAMING_SNAKE_CASE    (enum.Enum							):
     lowerCAmelCase					=       0
     lowerCAmelCase					=       1






def        _lowerCAmelCase			(  __snake_case  : Tuple ,							__snake_case  : List[str]=""			)     ->     List[str]:
     sys.stdout.write(str(__snake_case			) + end			)
     sys.stdout.flush()






def        _lowerCAmelCase			(  __snake_case  : Tuple ,							__snake_case  : Dict ,							__snake_case  : Tuple=""			)     ->     List[str]:
     forceWrite(f'\u001b[{color}m{content}\u001b[0m' ,							__snake_case			)






def        _lowerCAmelCase			(  )     ->     Union[str, Any]:
     forceWrite('\r'			)






def        _lowerCAmelCase			(  __snake_case  : int ,							__snake_case  : str			)     ->     Dict:
     forceWrite(f'\033[{num_lines}{CURSOR_TO_CHAR[direction.upper()]}'			)






def        _lowerCAmelCase			(  )     ->     Dict:
     forceWrite(' ' * TERMINAL_WIDTH			)
     reset_cursor()







def        _lowerCAmelCase			(  )     ->     str:
     reset_cursor()
     forceWrite('-' * TERMINAL_WIDTH			)






'''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  SCREAMING_SNAKE_CASE    (a__							):


     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          __A						: Any      =       tempfile.mkdtemp()
          __A						: str      =       5

          # Realm tok
          __A						: Union[str, Any]      =       [
              '[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',
          ]
          __A						: Union[str, Any]      =       os.path.join(self.tmpdirname						,	'realm_tokenizer')
          os.makedirs(_UpperCAmelCase						,	exist_ok=_UpperCAmelCase)
          __A						: Tuple      =       os.path.join(_UpperCAmelCase						,	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]))

          __A						: int      =       os.path.join(self.tmpdirname						,	'realm_block_records')
          os.makedirs(_UpperCAmelCase						,	exist_ok=_UpperCAmelCase)


     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          return RealmTokenizer.from_pretrained(os.path.join(self.tmpdirname						,	'realm_tokenizer'))


     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          shutil.rmtree(self.tmpdirname)


     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          __A						: Union[str, Any]      =       RealmConfig(num_block_records=self.num_block_records)
          return config


     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          __A						: Dict      =       Dataset.from_dict(
              {
                  'id': ['0', '1'],
                  'question': ['foo', 'bar'],
                  'answers': [['Foo', 'Bar'], ['Bar']],
              })
          return dataset


     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          __A						: Any      =       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=_UpperCAmelCase						,	)
          return block_records


     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          __A						: Optional[Any]      =       RealmRetriever(
              block_records=self.get_dummy_block_records()						,	tokenizer=self.get_tokenizer()						,	)
          return retriever


     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          __A						: List[Any]      =       self.get_config()
          __A						: str      =       self.get_dummy_retriever()
          __A						: List[str]      =       retriever.tokenizer

          __A						: Dict      =       np.array([0, 3]						,	dtype='long')
          __A						: Dict      =       tokenizer(['Test question']).input_ids
          __A						: Optional[Any]      =       tokenizer(
              ['the fourth']						,	add_special_tokens=_UpperCAmelCase						,	return_token_type_ids=_UpperCAmelCase						,	return_attention_mask=_UpperCAmelCase						,	).input_ids
          __A						: str      =       config.reader_seq_len

          __A ,__A ,__A ,__A						: List[str]      =       retriever(
              _UpperCAmelCase						,	_UpperCAmelCase						,	answer_ids=_UpperCAmelCase						,	max_length=_UpperCAmelCase						,	return_tensors='np')

          self.assertEqual(len(_UpperCAmelCase)						,	2)
          self.assertEqual(len(_UpperCAmelCase)						,	2)
          self.assertEqual(len(_UpperCAmelCase)						,	2)
          self.assertEqual(concat_inputs.input_ids.shape						,	(2, 10))
          self.assertEqual(concat_inputs.attention_mask.shape						,	(2, 10))
          self.assertEqual(concat_inputs.token_type_ids.shape						,	(2, 10))
          self.assertEqual(concat_inputs.special_tokens_mask.shape						,	(2, 10))
          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):

          '''simple docstring'''






          __A						: int      =       self.get_config()
          __A						: Any      =       self.get_dummy_retriever()
          __A						: str      =       retriever.tokenizer

          __A						: Dict      =       np.array([0, 3, 5]						,	dtype='long')
          __A						: Tuple      =       tokenizer(['Test question']).input_ids
          __A						: Union[str, Any]      =       tokenizer(
              ['the fourth', 'longer longer']						,	add_special_tokens=_UpperCAmelCase						,	return_token_type_ids=_UpperCAmelCase						,	return_attention_mask=_UpperCAmelCase						,	).input_ids
          __A						: Dict      =       config.reader_seq_len

          __A ,__A ,__A ,__A						: str      =       retriever(
              _UpperCAmelCase						,	_UpperCAmelCase						,	answer_ids=_UpperCAmelCase						,	max_length=_UpperCAmelCase						,	return_tensors='np')

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


     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          __A						: Optional[Any]      =       self.get_dummy_retriever()
          retriever.save_pretrained(os.path.join(self.tmpdirname						,	'realm_block_records'))

          # Test local path
          __A						: str      =       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:
               __A						: int      =       os.path.join(
                   os.path.join(self.tmpdirname						,	'realm_block_records')						,	_REALM_BLOCK_RECORDS_FILENAME)
               __A						: Tuple      =       RealmRetriever.from_pretrained('google/realm-cc-news-pretrained-openqa')

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