
'''simple docstring'''

import tempfile
import unittest

from transformers import DPRConfig, is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
		import torch

		from transformers import DPRContextEncoder, DPRQuestionEncoder, DPRReader, DPRReaderTokenizer
		from transformers.models.dpr.modeling_dpr import (
		    DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST,
		    DPR_QUESTION_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST,
		    DPR_READER_PRETRAINED_MODEL_ARCHIVE_LIST,
		)






class       A   :





	def __init__(    self	:				List[str]					,							__magic_name__	:				Optional[Any]					,							__magic_name__	:				int=13					,							__magic_name__	:				Optional[Any]=7					,							__magic_name__	:				Dict=False					,							__magic_name__	:				str=True					,							__magic_name__	:				Tuple=True					,							__magic_name__	:				int=True					,							__magic_name__	:				Optional[Any]=99					,							__magic_name__	:				Optional[int]=32					,							__magic_name__	:				Optional[int]=5					,							__magic_name__	:				Optional[int]=4					,							__magic_name__	:				int=37					,							__magic_name__	:				Tuple="gelu"					,							__magic_name__	:				Any=0.1					,							__magic_name__	:				Optional[Any]=0.1					,							__magic_name__	:				List[str]=512					,							__magic_name__	:				Any=16					,							__magic_name__	:				Optional[int]=2					,							__magic_name__	:				Union[str, Any]=0.02					,							__magic_name__	:				str=3					,							__magic_name__	:				List[str]=4					,							__magic_name__	:				Union[str, Any]=None					,							__magic_name__	:				Any=0					,							):


				"""simple docstring"""

				lowerCAmelCase__									=							parent
				lowerCAmelCase__									=							batch_size
				lowerCAmelCase__									=							seq_length
				lowerCAmelCase__									=							is_training
				lowerCAmelCase__									=							use_input_mask
				lowerCAmelCase__									=							use_token_type_ids
				lowerCAmelCase__									=							use_labels
				lowerCAmelCase__									=							vocab_size
				lowerCAmelCase__									=							hidden_size
				lowerCAmelCase__									=							num_hidden_layers
				lowerCAmelCase__									=							num_attention_heads
				lowerCAmelCase__									=							intermediate_size
				lowerCAmelCase__									=							hidden_act
				lowerCAmelCase__									=							hidden_dropout_prob
				lowerCAmelCase__									=							attention_probs_dropout_prob
				lowerCAmelCase__									=							max_position_embeddings
				lowerCAmelCase__									=							type_vocab_size
				lowerCAmelCase__									=							type_sequence_label_size
				lowerCAmelCase__									=							initializer_range
				lowerCAmelCase__									=							num_labels
				lowerCAmelCase__									=							num_choices
				lowerCAmelCase__									=							scope
				lowerCAmelCase__									=							projection_dim





	def 							__SCREAMING_SNAKE_CASE     (    self	:				Optional[int]							):


				"""simple docstring"""

				lowerCAmelCase__									=							ids_tensor([self.batch_size, self.seq_length]					,							self.vocab_size							)

				lowerCAmelCase__									=							None
				if self.use_input_mask:
							lowerCAmelCase__									=							random_attention_mask([self.batch_size, self.seq_length]							)

				lowerCAmelCase__									=							None
				if self.use_token_type_ids:
							lowerCAmelCase__									=							ids_tensor([self.batch_size, self.seq_length]					,							self.type_vocab_size							)

				lowerCAmelCase__									=							None
				lowerCAmelCase__									=							None
				lowerCAmelCase__									=							None
				if self.use_labels:
							lowerCAmelCase__									=							ids_tensor([self.batch_size]					,							self.type_sequence_label_size							)
							lowerCAmelCase__									=							ids_tensor([self.batch_size, self.seq_length]					,							self.num_labels							)
							lowerCAmelCase__									=							ids_tensor([self.batch_size]					,							self.num_choices							)

				lowerCAmelCase__									=							self.get_config()

				return config, input_ids, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels





	def 							__SCREAMING_SNAKE_CASE     (    self	:				int							):


				"""simple docstring"""

				return DPRConfig(
				    projection_dim=self.projection_dim					,							vocab_size=self.vocab_size					,							hidden_size=self.hidden_size					,							num_hidden_layers=self.num_hidden_layers					,							num_attention_heads=self.num_attention_heads					,							intermediate_size=self.intermediate_size					,							hidden_act=self.hidden_act					,							hidden_dropout_prob=self.hidden_dropout_prob					,							attention_probs_dropout_prob=self.attention_probs_dropout_prob					,							max_position_embeddings=self.max_position_embeddings					,							type_vocab_size=self.type_vocab_size					,							initializer_range=self.initializer_range					,							)





	def 							__SCREAMING_SNAKE_CASE     (    self	:				Tuple					,							__magic_name__	:				Optional[int]					,							__magic_name__	:				Tuple					,							__magic_name__	:				Dict					,							__magic_name__	:				List[Any]					,							__magic_name__	:				int					,							__magic_name__	:				Any					,							__magic_name__	:				Optional[int]							):


				"""simple docstring"""

				lowerCAmelCase__									=							DPRContextEncoder(config=_lowerCamelCase							)
				model.to(_lowerCamelCase							)
				model.eval()
				lowerCAmelCase__									=							model(_lowerCamelCase					,							attention_mask=_lowerCamelCase					,							token_type_ids=_lowerCamelCase							)
				lowerCAmelCase__									=							model(_lowerCamelCase					,							token_type_ids=_lowerCamelCase							)
				lowerCAmelCase__									=							model(_lowerCamelCase							)
				self.parent.assertEqual(result.pooler_output.shape					,							(self.batch_size, self.projection_dim or self.hidden_size)							)





	def 							__SCREAMING_SNAKE_CASE     (    self	:				str					,							__magic_name__	:				Union[str, Any]					,							__magic_name__	:				str					,							__magic_name__	:				Tuple					,							__magic_name__	:				int					,							__magic_name__	:				Optional[Any]					,							__magic_name__	:				Optional[int]					,							__magic_name__	:				Tuple							):


				"""simple docstring"""

				lowerCAmelCase__									=							DPRQuestionEncoder(config=_lowerCamelCase							)
				model.to(_lowerCamelCase							)
				model.eval()
				lowerCAmelCase__									=							model(_lowerCamelCase					,							attention_mask=_lowerCamelCase					,							token_type_ids=_lowerCamelCase							)
				lowerCAmelCase__									=							model(_lowerCamelCase					,							token_type_ids=_lowerCamelCase							)
				lowerCAmelCase__									=							model(_lowerCamelCase							)
				self.parent.assertEqual(result.pooler_output.shape					,							(self.batch_size, self.projection_dim or self.hidden_size)							)





	def 							__SCREAMING_SNAKE_CASE     (    self	:				Optional[Any]					,							__magic_name__	:				Union[str, Any]					,							__magic_name__	:				Tuple					,							__magic_name__	:				Any					,							__magic_name__	:				Optional[int]					,							__magic_name__	:				str					,							__magic_name__	:				Any					,							__magic_name__	:				Union[str, Any]							):


				"""simple docstring"""

				lowerCAmelCase__									=							DPRReader(config=_lowerCamelCase							)
				model.to(_lowerCamelCase							)
				model.eval()
				lowerCAmelCase__									=							model(
				    _lowerCamelCase					,							attention_mask=_lowerCamelCase					,							)

				self.parent.assertEqual(result.start_logits.shape					,							(self.batch_size, self.seq_length)							)
				self.parent.assertEqual(result.end_logits.shape					,							(self.batch_size, self.seq_length)							)
				self.parent.assertEqual(result.relevance_logits.shape					,							(self.batch_size,)							)







	def 							__SCREAMING_SNAKE_CASE     (    self	:				Tuple							):


				"""simple docstring"""

				lowerCAmelCase__									=							self.prepare_config_and_inputs()
				(
				    (
				    lowerCAmelCase__						
				)		,(
				    lowerCAmelCase__						
				)		,(
				    lowerCAmelCase__						
				)		,(
				    lowerCAmelCase__						
				)		,(
				    lowerCAmelCase__						
				)		,(
				    lowerCAmelCase__						
				)		,(
				    lowerCAmelCase__						
				)		,						
				)									=							config_and_inputs
				lowerCAmelCase__									=							{"input_ids": input_ids}
				return config, inputs_dict






@require_torch
class       A   (		lowerCAmelCase__     ,					lowerCAmelCase__     ,					unittest.TestCase			):
	snake_case__					:Union[str, Any]         =   (
	    (
	        DPRContextEncoder,
	        DPRQuestionEncoder,
	        DPRReader,
	    )
	    if is_torch_available()
	    else ()
	)
	snake_case__					:Optional[Any]         =   {"feature-extraction": DPRQuestionEncoder} if is_torch_available() else {}

	snake_case__					:int         =   False
	snake_case__					:Dict         =   False  # why?
	snake_case__					:Union[str, Any]         =   False
	snake_case__					:Tuple         =   False





	def 							__SCREAMING_SNAKE_CASE     (    self	:				List[str]							):


				"""simple docstring"""

				lowerCAmelCase__									=							DPRModelTester(self							)
				lowerCAmelCase__									=							ConfigTester(self					,							config_class=_lowerCamelCase					,							hidden_size=37							)





	def 							__SCREAMING_SNAKE_CASE     (    self	:				Any							):


				"""simple docstring"""

				self.config_tester.run_common_tests()





	def 							__SCREAMING_SNAKE_CASE     (    self	:				str							):


				"""simple docstring"""

				lowerCAmelCase__									=							self.model_tester.prepare_config_and_inputs()
				self.model_tester.create_and_check_context_encoder(*_lowerCamelCase							)





	def 							__SCREAMING_SNAKE_CASE     (    self	:				Any							):


				"""simple docstring"""

				lowerCAmelCase__									=							self.model_tester.prepare_config_and_inputs()
				self.model_tester.create_and_check_question_encoder(*_lowerCamelCase							)





	def 							__SCREAMING_SNAKE_CASE     (    self	:				Optional[Any]							):


				"""simple docstring"""

				lowerCAmelCase__									=							self.model_tester.prepare_config_and_inputs()
				self.model_tester.create_and_check_reader(*_lowerCamelCase							)





	def 							__SCREAMING_SNAKE_CASE     (    self	:				str							):


				"""simple docstring"""

				lowerCAmelCase__									=							self.model_tester.prepare_config_and_inputs()[0]

				lowerCAmelCase__									=							DPRQuestionEncoder(config=_lowerCamelCase							)
				model.to(_lowerCamelCase							)
				model.eval()

				with tempfile.TemporaryDirectory() as tmp_dirname:
							model.save_pretrained(_lowerCamelCase							)
							lowerCAmelCase__									=							DPRQuestionEncoder.from_pretrained(_lowerCamelCase					,							projection_dim=512							)

				self.assertIsNotNone(_lowerCamelCase							)







	@slow
	def 							__SCREAMING_SNAKE_CASE     (    self	:				Any							):


				"""simple docstring"""

				for model_name in DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
							lowerCAmelCase__									=							DPRContextEncoder.from_pretrained(_lowerCamelCase							)
							self.assertIsNotNone(_lowerCamelCase							)

				for model_name in DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
							lowerCAmelCase__									=							DPRContextEncoder.from_pretrained(_lowerCamelCase							)
							self.assertIsNotNone(_lowerCamelCase							)

				for model_name in DPR_QUESTION_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
							lowerCAmelCase__									=							DPRQuestionEncoder.from_pretrained(_lowerCamelCase							)
							self.assertIsNotNone(_lowerCamelCase							)

				for model_name in DPR_READER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
							lowerCAmelCase__									=							DPRReader.from_pretrained(_lowerCamelCase							)
							self.assertIsNotNone(_lowerCamelCase							)






@require_torch
class       A   (		unittest.TestCase			):





	@slow
	def 							__SCREAMING_SNAKE_CASE     (    self	:				Dict							):


				"""simple docstring"""

				lowerCAmelCase__									=							DPRQuestionEncoder.from_pretrained("facebook/dpr-question_encoder-single-nq-base"					,							return_dict=_lowerCamelCase							)
				model.to(_lowerCamelCase							)

				lowerCAmelCase__									=							torch.tensor(
				    [[101, 7592, 1010, 2003, 2026, 3899, 10140, 1029, 102]]					,							dtype=torch.long					,							device=_lowerCamelCase							)  # [CLS] hello, is my dog cute? [SEP]
				lowerCAmelCase__									=							model(_lowerCamelCase							)[0]  # embedding shape = (1, 768)
				# compare the actual values for a slice.
				lowerCAmelCase__									=							torch.tensor(
				    [
				        [
				            0.0323_6253,
				            0.1275_3335,
				            0.1681_8509,
				            0.0027_9786,
				            0.389_6933,
				            0.2426_4945,
				            0.217_8971,
				            -0.0233_5227,
				            -0.0848_1959,
				            -0.1432_4117,
				        ]
				    ]					,							dtype=torch.float					,							device=_lowerCamelCase					,							)
				self.assertTrue(torch.allclose(output[:, :10]					,							_lowerCamelCase					,							atol=1E-4							)							)







	@slow
	def 							__SCREAMING_SNAKE_CASE     (    self	:				str							):


				"""simple docstring"""

				lowerCAmelCase__									=							DPRReaderTokenizer.from_pretrained("facebook/dpr-reader-single-nq-base"							)
				lowerCAmelCase__									=							DPRReader.from_pretrained("facebook/dpr-reader-single-nq-base"							)
				model.to(_lowerCamelCase							)

				lowerCAmelCase__									=							tokenizer(
				    questions="What is love ?"					,							titles="Haddaway"					,							texts="What Is Love is a song recorded by the artist Haddaway"					,							padding=_lowerCamelCase					,							return_tensors="pt"					,							)
				encoded_inputs.to(_lowerCamelCase							)

				lowerCAmelCase__									=							model(**_lowerCamelCase							)

				# compare the actual values for a slice.
				lowerCAmelCase__									=							torch.tensor(
				    [[-10.3005, -10.7765, -11.4872, -11.6841, -11.9312, -10.3002, -9.8544, -11.7378, -12.0821, -10.2975]]					,							dtype=torch.float					,							device=_lowerCamelCase					,							)

				lowerCAmelCase__									=							torch.tensor(
				    [[-11.0684, -11.7041, -11.5397, -10.3465, -10.8791, -6.8443, -11.9959, -11.0364, -10.0096, -6.8405]]					,							dtype=torch.float					,							device=_lowerCamelCase					,							)
				self.assertTrue(torch.allclose(outputs.start_logits[:, :10]					,							_lowerCamelCase					,							atol=1E-4							)							)
				self.assertTrue(torch.allclose(outputs.end_logits[:, :10]					,							_lowerCamelCase					,							atol=1E-4							)							)












'''simple docstring'''



class 							__lowercase				:







	'''simple docstring'''

	def __init__(self   ,_lowerCamelCase       )							->							Optional[Any]:







				'''simple docstring'''





				__lowercase   =							val
				# Number of nodes in left subtree
				__lowercase   =							0
				__lowercase   =							None
				__lowercase   =							None
				__lowercase   =							None


	def   _UpperCAmelCase   (self   ,_lowerCamelCase       )							->							Tuple:







				'''simple docstring'''





				assert self.left_tree_size == other.left_tree_size, "Unequal Sizes of Blocks"

				if self.val < other.val:
							__lowercase   =							self.right
							__lowercase   =							None
							if self.right:
										__lowercase   =							other
							__lowercase   =							other
							__lowercase   =							self.left_tree_size * 2 + 1
							return self
				else:
							__lowercase   =							other.right
							__lowercase   =							None
							if other.right:
										__lowercase   =							self
							__lowercase   =							self
							__lowercase   =							other.left_tree_size * 2 + 1
							return other






class 							__lowercase				:







	'''simple docstring'''

	def __init__(self   ,_lowerCamelCase=None   ,_lowerCamelCase=None   ,_lowerCamelCase=0       )							->							int:







				'''simple docstring'''





				__lowercase   =							heap_size
				__lowercase   =							bottom_root
				__lowercase   =							min_node


	def   _UpperCAmelCase   (self   ,_lowerCamelCase       )							->							Tuple:







				'''simple docstring'''





				if other.size == 0:
							return None
				if self.size == 0:
							__lowercase   =							other.size
							__lowercase   =							other.bottom_root
							__lowercase   =							other.min_node
							return None
				# Update size
				__lowercase   =							self.size + other.size

				# Update min.node
				if self.min_node.val > other.min_node.val:
							__lowercase   =							other.min_node
				# Merge

				# Order roots by left_subtree_size
				__lowercase   =							[]
				__lowercase  ,						__lowercase   =							self.bottom_root, other.bottom_root
				while i or j:
							if i and ((not j) or i.left_tree_size < j.left_tree_size):
										combined_roots_list.append((i, True)       )
										__lowercase   =							i.parent
							else:
										combined_roots_list.append((j, False)       )
										__lowercase   =							j.parent
        # Insert links between them
				for i in range(len(_lowerCamelCase       ) - 1       ):
							if combined_roots_list[i][1] != combined_roots_list[i + 1][1]:
										__lowercase   =							combined_roots_list[i + 1][0]
										__lowercase   =							combined_roots_list[i][0]
        # Consecutively merge roots with same left_tree_size
				__lowercase   =							combined_roots_list[0][0]
				while i.parent:
							if (
							    (i.left_tree_size == i.parent.left_tree_size) and (not i.parent.parent)
							) or (
							    i.left_tree_size == i.parent.left_tree_size
							    and i.left_tree_size != i.parent.parent.left_tree_size
							):
										# Neighbouring Nodes
										__lowercase   =							i.left
										__lowercase   =							i.parent.parent

										# Merging trees
										__lowercase   =							i.merge_trees(i.parent       )

										# Updating links
										__lowercase   =							previous_node
										__lowercase   =							next_node
										if previous_node:
													__lowercase   =							i
										if next_node:
													__lowercase   =							i
							else:
										__lowercase   =							i.parent
        # Updating self.bottom_root
				while i.left:
							__lowercase   =							i.left
				__lowercase   =							i

				# Update other
				__lowercase   =							self.size
				__lowercase   =							self.bottom_root
				__lowercase   =							self.min_node

				# Return the merged heap
				return self


	def   _UpperCAmelCase   (self   ,_lowerCamelCase       )							->							Dict:







				'''simple docstring'''





				if self.size == 0:
							__lowercase   =							Node(_lowerCamelCase       )
							__lowercase   =							1
							__lowercase   =							self.bottom_root
				else:
							# Create new node
							__lowercase   =							Node(_lowerCamelCase       )

							# Update size
							self.size += 1

							# update min_node
							if val < self.min_node.val:
										__lowercase   =							new_node
							# Put new_node as a bottom_root in heap
							__lowercase   =							new_node
							__lowercase   =							self.bottom_root
							__lowercase   =							new_node

							# Consecutively merge roots with same left_tree_size
							while (
							    self.bottom_root.parent
							    and self.bottom_root.left_tree_size
							    == self.bottom_root.parent.left_tree_size
							):
										# Next node
										__lowercase   =							self.bottom_root.parent.parent

										# Merge
										__lowercase   =							self.bottom_root.merge_trees(self.bottom_root.parent       )

										# Update Links
										__lowercase   =							next_node
										__lowercase   =							None
										if next_node:
													__lowercase   =							self.bottom_root


	def   _UpperCAmelCase   (self       )							->							Union[str, Any]:







				'''simple docstring'''





				return self.min_node.val


	def   _UpperCAmelCase   (self       )							->							List[Any]:







				'''simple docstring'''





				return self.size == 0


	def   _UpperCAmelCase   (self       )							->							Union[str, Any]:







				'''simple docstring'''





				__lowercase   =							self.min_node.val

				# Last element in heap corner case
				if self.size == 1:
							# Update size
							__lowercase   =							0

							# Update bottom root
							__lowercase   =							None

							# Update min_node
							__lowercase   =							None

							return min_value
				# No right subtree corner case
				# The structure of the tree implies that this should be the bottom root
				# and there is at least one other root
				if self.min_node.right is None:
							# Update size
							self.size -= 1

							# Update bottom root
							__lowercase   =							self.bottom_root.parent
							__lowercase   =							None

							# Update min_node
							__lowercase   =							self.bottom_root
							__lowercase   =							self.bottom_root.parent
							while i:
										if i.val < self.min_node.val:
													__lowercase   =							i
										__lowercase   =							i.parent
							return min_value
				# General case
				# Find the BinomialHeap of the right subtree of min_node
				__lowercase   =							self.min_node.right
				__lowercase   =							None
				__lowercase   =							bottom_of_new
				__lowercase   =							1

				# Size, min_node and bottom_root
				while bottom_of_new.left:
							__lowercase   =							size_of_new * 2 + 1
							__lowercase   =							bottom_of_new.left
							if bottom_of_new.val < min_of_new.val:
										__lowercase   =							bottom_of_new
        # Corner case of single root on top left path
				if (not self.min_node.left) and (not self.min_node.parent):
							__lowercase   =							size_of_new
							__lowercase   =							bottom_of_new
							__lowercase   =							min_of_new
							# print("Single root, multiple nodes case")
							return min_value
				# Remaining cases
				# Construct heap of right subtree
				__lowercase   =							BinomialHeap(
				    bottom_root=_lowerCamelCase   ,min_node=_lowerCamelCase   ,heap_size=_lowerCamelCase       )

				# Update size
				__lowercase   =							self.size - 1 - size_of_new

				# Neighbour nodes
				__lowercase   =							self.min_node.left
				__lowercase   =							self.min_node.parent

				# Initialize new bottom_root and min_node
				__lowercase   =							previous_node or next_node
				__lowercase   =							next_node

				# Update links of previous_node and search below for new min_node and
				# bottom_root
				if previous_node:
							__lowercase   =							next_node

							# Update bottom_root and search for min_node below
							__lowercase   =							previous_node
							__lowercase   =							previous_node
							while self.bottom_root.left:
										__lowercase   =							self.bottom_root.left
										if self.bottom_root.val < self.min_node.val:
													__lowercase   =							self.bottom_root
				if next_node:
							__lowercase   =							previous_node

							# Search for new min_node above min_node
							__lowercase   =							next_node
							while i:
										if i.val < self.min_node.val:
													__lowercase   =							i
										__lowercase   =							i.parent
        # Merge heaps
				self.merge_heaps(_lowerCamelCase       )

				return min_value


	def   _UpperCAmelCase   (self       )							->							str:







				'''simple docstring'''





				__lowercase   =							self.bottom_root
				while top_root.parent:
							__lowercase   =							top_root.parent
				# preorder
				__lowercase   =							[]
				self.__traversal(_lowerCamelCase   ,_lowerCamelCase       )
				return heap_pre_order


	def   _UpperCAmelCase   (self   ,_lowerCamelCase   ,_lowerCamelCase   ,_lowerCamelCase=0       )							->							str:







				'''simple docstring'''





				if curr_node:
							preorder.append((curr_node.val, level)       )
							self.__traversal(curr_node.left   ,_lowerCamelCase   ,level + 1       )
							self.__traversal(curr_node.right   ,_lowerCamelCase   ,level + 1       )
				else:
							preorder.append(('''#''', level)       )


	def __str__(self       )							->							List[str]:







				'''simple docstring'''





				if self.is_empty():
							return ""
				__lowercase   =							self.pre_order()

				return "\n".join(('''-''' * level + str(_lowerCamelCase       )) for value, level in preorder_heap       )


# Unit Tests
if __name__ == "__main__":
	import doctest

	doctest.testmod()





