





'''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 __a (   _snake_case			):





   def      UpperCAmelCase__     (					self				:			str    ):

      '''simple docstring'''

      __SCREAMING_SNAKE_CASE      							=       tempfile.mkdtemp()
      __SCREAMING_SNAKE_CASE      							=       5

      # Realm tok
      __SCREAMING_SNAKE_CASE      							=       [
          """[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""",
      ]
      __SCREAMING_SNAKE_CASE      							=       os.path.join(self.tmpdirname				,"""realm_tokenizer"""    )
      os.makedirs(lowerCamelCase				,exist_ok=lowerCamelCase    )
      __SCREAMING_SNAKE_CASE      							=       os.path.join(lowerCamelCase				,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]    )    )

      __SCREAMING_SNAKE_CASE      							=       os.path.join(self.tmpdirname				,"""realm_block_records"""    )
      os.makedirs(lowerCamelCase				,exist_ok=lowerCamelCase    )





   def      UpperCAmelCase__     (					self				:			Any    ):

      '''simple docstring'''

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





   def      UpperCAmelCase__     (					self				:			int    ):

      '''simple docstring'''

      shutil.rmtree(self.tmpdirname    )





   def      UpperCAmelCase__     (					self				:			Optional[Any]    ):

      '''simple docstring'''

      __SCREAMING_SNAKE_CASE      							=       RealmConfig(num_block_records=self.num_block_records    )
      return config





   def      UpperCAmelCase__     (					self				:			Any    ):

      '''simple docstring'''

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





   def      UpperCAmelCase__     (					self				:			Optional[int]    ):

      '''simple docstring'''

      __SCREAMING_SNAKE_CASE      							=       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=lowerCamelCase				,)
      return block_records





   def      UpperCAmelCase__     (					self				:			List[Any]    ):

      '''simple docstring'''

      __SCREAMING_SNAKE_CASE      							=       RealmRetriever(
          block_records=self.get_dummy_block_records()				,tokenizer=self.get_tokenizer()				,)
      return retriever





   def      UpperCAmelCase__     (					self				:			List[Any]    ):

      '''simple docstring'''

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

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

      __SCREAMING_SNAKE_CASE							,     __SCREAMING_SNAKE_CASE							,     __SCREAMING_SNAKE_CASE							,     __SCREAMING_SNAKE_CASE      							=       retriever(
          lowerCamelCase				,lowerCamelCase				,answer_ids=lowerCamelCase				,max_length=lowerCamelCase				,return_tensors="""np"""    )

      self.assertEqual(len(lowerCamelCase    )				,2    )
      self.assertEqual(len(lowerCamelCase    )				,2    )
      self.assertEqual(len(lowerCamelCase    )				,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      UpperCAmelCase__     (					self				:			List[str]    ):

      '''simple docstring'''

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

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

      __SCREAMING_SNAKE_CASE							,     __SCREAMING_SNAKE_CASE							,     __SCREAMING_SNAKE_CASE							,     __SCREAMING_SNAKE_CASE      							=       retriever(
          lowerCamelCase				,lowerCamelCase				,answer_ids=lowerCamelCase				,max_length=lowerCamelCase				,return_tensors="""np"""    )

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





   def      UpperCAmelCase__     (					self				:			int    ):

      '''simple docstring'''

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

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

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












'''simple docstring'''

import math
from enum import Enum
from typing import Optional, Union

from torch.optim import Optimizer
from torch.optim.lr_scheduler import LambdaLR

from .utils import logging


a        =						logging.get_logger(__name__)







class __a (   _snake_case			):
   __UpperCamelCase   : int	      =	'linear'
   __UpperCamelCase   : Tuple	      =	'cosine'
   __UpperCamelCase   : Tuple	      =	'cosine_with_restarts'
   __UpperCamelCase   : List[Any]	      =	'polynomial'
   __UpperCamelCase   : Optional[Any]	      =	'constant'
   __UpperCamelCase   : Optional[int]	      =	'constant_with_warmup'
   __UpperCamelCase   : List[Any]	      =	'piecewise_constant'





def 						__magic_name__       (							__UpperCAmelCase							,	__UpperCAmelCase = -1      )						->					int:

   '''simple docstring'''







   return LambdaLR(__UpperCAmelCase							,	lambda __UpperCAmelCase      : 1							,	last_epoch=__UpperCAmelCase      )

def 						__magic_name__       (							__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase = -1      )						->					List[Any]:

   '''simple docstring'''







   def lr_lambda(__UpperCAmelCase      ):
      if current_step < num_warmup_steps:
         return float(__UpperCAmelCase      ) / float(max(1.0							,	__UpperCAmelCase      )      )
      return 1.0

   return LambdaLR(__UpperCAmelCase							,	__UpperCAmelCase							,	last_epoch=__UpperCAmelCase      )

def 						__magic_name__       (							__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase = -1      )						->					int:

   '''simple docstring'''







   __SCREAMING_SNAKE_CASE      							=       {}
   __SCREAMING_SNAKE_CASE      							=       step_rules.split(""","""      )
   for rule_str in rule_list[:-1]:
      __SCREAMING_SNAKE_CASE							,     __SCREAMING_SNAKE_CASE      							=       rule_str.split(""":"""      )
      __SCREAMING_SNAKE_CASE      							=       int(__UpperCAmelCase      )
      __SCREAMING_SNAKE_CASE      							=       float(__UpperCAmelCase      )
      __SCREAMING_SNAKE_CASE      							=       value
   __SCREAMING_SNAKE_CASE      							=       float(rule_list[-1]      )

   def create_rules_function(__UpperCAmelCase							,	__UpperCAmelCase      ):
      def rule_func(__UpperCAmelCase      ) -> float:
         __SCREAMING_SNAKE_CASE      							=       sorted(rules_dict.keys()      )
         for i, sorted_step in enumerate(__UpperCAmelCase      ):
            if steps < sorted_step:
               return rules_dict[sorted_steps[i]]
         return last_lr_multiple

      return rule_func

   __SCREAMING_SNAKE_CASE      							=       create_rules_function(__UpperCAmelCase							,	__UpperCAmelCase      )

   return LambdaLR(__UpperCAmelCase							,	__UpperCAmelCase							,	last_epoch=__UpperCAmelCase      )

def 						__magic_name__       (							__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase=-1      )						->					int:

   '''simple docstring'''







   def lr_lambda(__UpperCAmelCase      ):
      if current_step < num_warmup_steps:
         return float(__UpperCAmelCase      ) / float(max(1							,	__UpperCAmelCase      )      )
      return max(
          0.0							,	float(num_training_steps - current_step      ) / float(max(1							,	num_training_steps - num_warmup_steps      )      )      )

   return LambdaLR(__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase      )

def 						__magic_name__       (							__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase = 0.5							,	__UpperCAmelCase = -1      )						->					Dict:

   '''simple docstring'''







   def lr_lambda(__UpperCAmelCase      ):
      if current_step < num_warmup_steps:
         return float(__UpperCAmelCase      ) / float(max(1							,	__UpperCAmelCase      )      )
      __SCREAMING_SNAKE_CASE      							=       float(current_step - num_warmup_steps      ) / float(max(1							,	num_training_steps - num_warmup_steps      )      )
      return max(0.0							,	0.5 * (1.0 + math.cos(math.pi * float(__UpperCAmelCase      ) * 2.0 * progress      ))      )

   return LambdaLR(__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase      )

def 						__magic_name__       (							__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase = 1							,	__UpperCAmelCase = -1      )						->					Tuple:

   '''simple docstring'''







   def lr_lambda(__UpperCAmelCase      ):
      if current_step < num_warmup_steps:
         return float(__UpperCAmelCase      ) / float(max(1							,	__UpperCAmelCase      )      )
      __SCREAMING_SNAKE_CASE      							=       float(current_step - num_warmup_steps      ) / float(max(1							,	num_training_steps - num_warmup_steps      )      )
      if progress >= 1.0:
         return 0.0
      return max(0.0							,	0.5 * (1.0 + math.cos(math.pi * ((float(__UpperCAmelCase      ) * progress) % 1.0)      ))      )

   return LambdaLR(__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase      )

def 						__magic_name__       (							__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase=1e-7							,	__UpperCAmelCase=1.0							,	__UpperCAmelCase=-1      )						->					Tuple:

   '''simple docstring'''







   __SCREAMING_SNAKE_CASE      							=       optimizer.defaults["""lr"""]
   if not (lr_init > lr_end):
      raise ValueError(f"""lr_end ({lr_end}) must be be smaller than initial lr ({lr_init})"""      )

   def lr_lambda(__UpperCAmelCase      ):
      if current_step < num_warmup_steps:
         return float(__UpperCAmelCase      ) / float(max(1							,	__UpperCAmelCase      )      )
      elif current_step > num_training_steps:
         return lr_end / lr_init  # as LambdaLR multiplies by lr_init
      else:
         __SCREAMING_SNAKE_CASE      							=       lr_init - lr_end
         __SCREAMING_SNAKE_CASE      							=       num_training_steps - num_warmup_steps
         __SCREAMING_SNAKE_CASE      							=       1 - (current_step - num_warmup_steps) / decay_steps
         __SCREAMING_SNAKE_CASE      							=       lr_range * pct_remaining**power + lr_end
         return decay / lr_init  # as LambdaLR multiplies by lr_init

   return LambdaLR(__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase      )


a        =						{
    SchedulerType.LINEAR: get_linear_schedule_with_warmup,
    SchedulerType.COSINE: get_cosine_schedule_with_warmup,
    SchedulerType.COSINE_WITH_RESTARTS: get_cosine_with_hard_restarts_schedule_with_warmup,
    SchedulerType.POLYNOMIAL: get_polynomial_decay_schedule_with_warmup,
    SchedulerType.CONSTANT: get_constant_schedule,
    SchedulerType.CONSTANT_WITH_WARMUP: get_constant_schedule_with_warmup,
    SchedulerType.PIECEWISE_CONSTANT: get_piecewise_constant_schedule,
}







def 						__magic_name__       (							__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase = None							,	__UpperCAmelCase = None							,	__UpperCAmelCase = None							,	__UpperCAmelCase = 1							,	__UpperCAmelCase = 1.0							,	__UpperCAmelCase = -1							,	)						->					str:

   '''simple docstring'''







   __SCREAMING_SNAKE_CASE      							=       SchedulerType(__UpperCAmelCase      )
   __SCREAMING_SNAKE_CASE      							=       TYPE_TO_SCHEDULER_FUNCTION[name]
   if name == SchedulerType.CONSTANT:
      return schedule_func(__UpperCAmelCase							,	last_epoch=__UpperCAmelCase      )

   if name == SchedulerType.PIECEWISE_CONSTANT:
      return schedule_func(__UpperCAmelCase							,	step_rules=__UpperCAmelCase							,	last_epoch=__UpperCAmelCase      )

   # All other schedulers require `num_warmup_steps`
   if num_warmup_steps is None:
      raise ValueError(f"""{name} requires `num_warmup_steps`, please provide that argument."""      )

   if name == SchedulerType.CONSTANT_WITH_WARMUP:
      return schedule_func(__UpperCAmelCase							,	num_warmup_steps=__UpperCAmelCase							,	last_epoch=__UpperCAmelCase      )

   # All other schedulers require `num_training_steps`
   if num_training_steps is None:
      raise ValueError(f"""{name} requires `num_training_steps`, please provide that argument."""      )

   if name == SchedulerType.COSINE_WITH_RESTARTS:
      return schedule_func(
          __UpperCAmelCase							,	num_warmup_steps=__UpperCAmelCase							,	num_training_steps=__UpperCAmelCase							,	num_cycles=__UpperCAmelCase							,	last_epoch=__UpperCAmelCase							,	)

   if name == SchedulerType.POLYNOMIAL:
      return schedule_func(
          __UpperCAmelCase							,	num_warmup_steps=__UpperCAmelCase							,	num_training_steps=__UpperCAmelCase							,	power=__UpperCAmelCase							,	last_epoch=__UpperCAmelCase							,	)

   return schedule_func(
       __UpperCAmelCase							,	num_warmup_steps=__UpperCAmelCase							,	num_training_steps=__UpperCAmelCase							,	last_epoch=__UpperCAmelCase      )





