





"""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 							snake_case							(			__UpperCAmelCase ):

   '''simple docstring'''



   def       A_					(							self       :      int ):


     '''simple docstring'''




     __UpperCAmelCase    :						str       =						tempfile.mkdtemp()
     __UpperCAmelCase    :						List[Any]       =						5

     # Realm tok
     __UpperCAmelCase    :						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''',
     ]
     __UpperCAmelCase    :						Optional[int]       =						os.path.join(self.tmpdirname						,				'''realm_tokenizer''' )
     os.makedirs(__lowercase						,				exist_ok=__lowercase )
     __UpperCAmelCase    :						List[Any]       =						os.path.join(__lowercase						,				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] ) )

     __UpperCAmelCase    :						Tuple       =						os.path.join(self.tmpdirname						,				'''realm_block_records''' )
     os.makedirs(__lowercase						,				exist_ok=__lowercase )

   def       A_					(							self       :      Union[str, Any] ):


     '''simple docstring'''




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

   def       A_					(							self       :      Optional[int] ):


     '''simple docstring'''




     shutil.rmtree(self.tmpdirname )

   def       A_					(							self       :      Union[str, Any] ):


     '''simple docstring'''




     __UpperCAmelCase    :						Any       =						RealmConfig(num_block_records=self.num_block_records )
     return config

   def       A_					(							self       :      int ):


     '''simple docstring'''




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

   def       A_					(							self       :      List[Any] ):


     '''simple docstring'''




     __UpperCAmelCase    :						Dict       =						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=__lowercase						,				)
     return block_records

   def       A_					(							self       :      List[Any] ):


     '''simple docstring'''




     __UpperCAmelCase    :						Dict       =						RealmRetriever(
         block_records=self.get_dummy_block_records()						,				tokenizer=self.get_tokenizer()						,				)
     return retriever

   def       A_					(							self       :      Optional[Any] ):


     '''simple docstring'''




     __UpperCAmelCase    :						Dict       =						self.get_config()
     __UpperCAmelCase    :						int       =						self.get_dummy_retriever()
     __UpperCAmelCase    :						Optional[int]       =						retriever.tokenizer

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

     __UpperCAmelCase    :						Tuple       =						retriever(
         __lowercase						,				__lowercase						,				answer_ids=__lowercase						,				max_length=__lowercase						,				return_tensors='''np''' )

     self.assertEqual(len(__lowercase )						,				2 )
     self.assertEqual(len(__lowercase )						,				2 )
     self.assertEqual(len(__lowercase )						,				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       A_					(							self       :      Any ):


     '''simple docstring'''




     __UpperCAmelCase    :						List[Any]       =						self.get_config()
     __UpperCAmelCase    :						int       =						self.get_dummy_retriever()
     __UpperCAmelCase    :						int       =						retriever.tokenizer

     __UpperCAmelCase    :						List[str]       =						np.array([0, 3, 5]						,				dtype='''long''' )
     __UpperCAmelCase    :						int       =						tokenizer(['''Test question'''] ).input_ids
     __UpperCAmelCase    :						Tuple       =						tokenizer(
         ['''the fourth''', '''longer longer''']						,				add_special_tokens=__lowercase						,				return_token_type_ids=__lowercase						,				return_attention_mask=__lowercase						,				).input_ids
     __UpperCAmelCase    :						Tuple       =						config.reader_seq_len

     __UpperCAmelCase    :						Dict       =						retriever(
         __lowercase						,				__lowercase						,				answer_ids=__lowercase						,				max_length=__lowercase						,				return_tensors='''np''' )

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







   def       A_					(							self       :      Any ):


     '''simple docstring'''




     __UpperCAmelCase    :						Any       =						self.get_dummy_retriever()
     retriever.save_pretrained(os.path.join(self.tmpdirname						,				'''realm_block_records''' ) )

     # Test local path
     __UpperCAmelCase    :						List[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:
       __UpperCAmelCase    :						Dict       =						os.path.join(
           os.path.join(self.tmpdirname						,				'''realm_block_records''' )						,				_REALM_BLOCK_RECORDS_FILENAME )
       __UpperCAmelCase    :						Optional[int]       =						RealmRetriever.from_pretrained('''google/realm-cc-news-pretrained-openqa''' )

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






"""simple docstring"""





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


lowercase__				:int				   =       logging.get_logger(__name__)

lowercase__				:List[str]				   =       {
    'EleutherAI/gpt-neox-20b': 'https://huggingface.co/EleutherAI/gpt-neox-20b/resolve/main/config.json',
    # See all GPTNeoX models at https://huggingface.co/models?filter=gpt_neox
}




class 							snake_case							(			__UpperCAmelCase ):

      '''simple docstring'''



      _A     :       List[Any]         =       'gpt_neox'

      def __init__(							self       :      Union[str, Any]						,				__lowercase       :      Any=50_432						,				__lowercase       :      int=6_144						,				__lowercase       :      List[str]=44						,				__lowercase       :      List[Any]=64						,				__lowercase       :      int=24_576						,				__lowercase       :      List[str]="gelu"						,				__lowercase       :      Optional[int]=0.2_5						,				__lowercase       :      Any=10_000						,				__lowercase       :      Union[str, Any]=0.0						,				__lowercase       :      List[str]=0.0						,				__lowercase       :      Union[str, Any]=0.1						,				__lowercase       :      Optional[int]=2_048						,				__lowercase       :      Union[str, Any]=0.0_2						,				__lowercase       :      Optional[int]=1e-5						,				__lowercase       :      str=True						,				__lowercase       :      str=0						,				__lowercase       :      Tuple=2						,				__lowercase       :      Tuple=False						,				__lowercase       :      int=True						,				__lowercase       :      Any=None						,				**__lowercase       :      List[str]						,				):


        '''simple docstring'''




        super().__init__(bos_token_id=__lowercase						,				eos_token_id=__lowercase						,				**__lowercase )
        __UpperCAmelCase    :						int       =						vocab_size
        __UpperCAmelCase    :						Union[str, Any]       =						max_position_embeddings
        __UpperCAmelCase    :						List[str]       =						hidden_size
        __UpperCAmelCase    :						Tuple       =						num_hidden_layers
        __UpperCAmelCase    :						Any       =						num_attention_heads
        __UpperCAmelCase    :						Union[str, Any]       =						intermediate_size
        __UpperCAmelCase    :						str       =						hidden_act
        __UpperCAmelCase    :						str       =						rotary_pct
        __UpperCAmelCase    :						str       =						rotary_emb_base
        __UpperCAmelCase    :						Dict       =						attention_dropout
        __UpperCAmelCase    :						str       =						hidden_dropout
        __UpperCAmelCase    :						Any       =						classifier_dropout
        __UpperCAmelCase    :						Tuple       =						initializer_range
        __UpperCAmelCase    :						Any       =						layer_norm_eps
        __UpperCAmelCase    :						Optional[int]       =						use_cache
        __UpperCAmelCase    :						List[Any]       =						tie_word_embeddings
        __UpperCAmelCase    :						Any       =						use_parallel_residual
        __UpperCAmelCase    :						List[str]       =						rope_scaling
        self._rope_scaling_validation()

        if self.hidden_size % self.num_attention_heads != 0:
          raise ValueError(
              '''The hidden size is not divisble by the number of attention heads! Make sure to update them!''' )







      def       A_					(							self       :      int ):


        '''simple docstring'''




        if self.rope_scaling is None:
          return

        if not isinstance(self.rope_scaling						,				__lowercase ) or len(self.rope_scaling ) != 2:
          raise ValueError(
              '''`rope_scaling` must be a dictionary with with two fields, `name` and `factor`, '''
              f'''got {self.rope_scaling}''' )
        __UpperCAmelCase    :						Union[str, Any]       =						self.rope_scaling.get('''type'''						,				__lowercase )
        __UpperCAmelCase    :						List[Any]       =						self.rope_scaling.get('''factor'''						,				__lowercase )
        if rope_scaling_type is None or rope_scaling_type not in ["linear", "dynamic"]:
          raise ValueError(
              f'''`rope_scaling`\'s name field must be one of [\'linear\', \'dynamic\'], got {rope_scaling_type}''' )
        if rope_scaling_factor is None or not isinstance(__lowercase						,				__lowercase ) or rope_scaling_factor <= 1.0:
          raise ValueError(f'''`rope_scaling`\'s factor field must be an float > 1, got {rope_scaling_factor}''' )