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  __lowerCAmelCase	:



 '''simple docstring'''

 def __init__(      self							:		int							,      lowerCamelCase							:		int							,      lowerCamelCase							:		int=1_3							,      lowerCamelCase							:		Tuple=7							,      lowerCamelCase							:		Optional[Any]=False							,      lowerCamelCase							:		Optional[Any]=True							,      lowerCamelCase							:		Dict=True							,      lowerCamelCase							:		str=True							,      lowerCamelCase							:		int=9_9							,      lowerCamelCase							:		Optional[Any]=3_2							,      lowerCamelCase							:		Any=5							,      lowerCamelCase							:		Any=4							,      lowerCamelCase							:		Dict=3_7							,      lowerCamelCase							:		List[Any]="gelu"							,      lowerCamelCase							:		Optional[Any]=0.1							,      lowerCamelCase							:		int=0.1							,      lowerCamelCase							:		Optional[int]=5_1_2							,      lowerCamelCase							:		int=1_6							,      lowerCamelCase							:		List[Any]=2							,      lowerCamelCase							:		int=0.0_2							,      lowerCamelCase							:		Optional[int]=3							,      lowerCamelCase							:		Any=4							,      lowerCamelCase							:		Tuple=None							,      lowerCamelCase							:		int=0							,      )			->				Union[str, Any]:
        _UpperCamelCase  :							Union[str, Any]	    =	parent
        _UpperCamelCase  :							Optional[Any]	    =	batch_size
        _UpperCamelCase  :							List[str]	    =	seq_length
        _UpperCamelCase  :							str	    =	is_training
        _UpperCamelCase  :							Optional[Any]	    =	use_input_mask
        _UpperCamelCase  :							Optional[Any]	    =	use_token_type_ids
        _UpperCamelCase  :							Tuple	    =	use_labels
        _UpperCamelCase  :							int	    =	vocab_size
        _UpperCamelCase  :							Any	    =	hidden_size
        _UpperCamelCase  :							str	    =	num_hidden_layers
        _UpperCamelCase  :							Optional[Any]	    =	num_attention_heads
        _UpperCamelCase  :							Optional[Any]	    =	intermediate_size
        _UpperCamelCase  :							Any	    =	hidden_act
        _UpperCamelCase  :							List[str]	    =	hidden_dropout_prob
        _UpperCamelCase  :							Dict	    =	attention_probs_dropout_prob
        _UpperCamelCase  :							Optional[Any]	    =	max_position_embeddings
        _UpperCamelCase  :							Tuple	    =	type_vocab_size
        _UpperCamelCase  :							Optional[Any]	    =	type_sequence_label_size
        _UpperCamelCase  :							str	    =	initializer_range
        _UpperCamelCase  :							Union[str, Any]	    =	num_labels
        _UpperCamelCase  :							Optional[int]	    =	num_choices
        _UpperCamelCase  :							str	    =	scope
        _UpperCamelCase  :							int	    =	projection_dim

 def        __snake_case       (      self							:		Optional[Any]	)			->				Union[str, Any]:
        _UpperCamelCase  :							Union[str, Any]	    =	ids_tensor([self.batch_size, self.seq_length]							,      self.vocab_size	)

        _UpperCamelCase  :							List[Any]	    =	None
        if self.use_input_mask:
               _UpperCamelCase  :							Optional[int]	    =	random_attention_mask([self.batch_size, self.seq_length]	)

        _UpperCamelCase  :							Optional[int]	    =	None
        if self.use_token_type_ids:
               _UpperCamelCase  :							Union[str, Any]	    =	ids_tensor([self.batch_size, self.seq_length]							,      self.type_vocab_size	)

        _UpperCamelCase  :							Union[str, Any]	    =	None
        _UpperCamelCase  :							List[str]	    =	None
        _UpperCamelCase  :							Tuple	    =	None
        if self.use_labels:
               _UpperCamelCase  :							List[Any]	    =	ids_tensor([self.batch_size]							,      self.type_sequence_label_size	)
               _UpperCamelCase  :							Dict	    =	ids_tensor([self.batch_size, self.seq_length]							,      self.num_labels	)
               _UpperCamelCase  :							Dict	    =	ids_tensor([self.batch_size]							,      self.num_choices	)

        _UpperCamelCase  :							List[str]	    =	self.get_config()

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

 def        __snake_case       (      self							:		Optional[int]	)			->				Optional[int]:
        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        __snake_case       (      self							:		Tuple							,      lowerCamelCase							:		Optional[Any]							,      lowerCamelCase							:		Dict							,      lowerCamelCase							:		Optional[int]							,      lowerCamelCase							:		List[str]							,      lowerCamelCase							:		Optional[int]							,      lowerCamelCase							:		Optional[int]							,      lowerCamelCase							:		str	)			->				Dict:
        _UpperCamelCase  :							Optional[int]	    =	DPRContextEncoder(config=lowerCamelCase	)
        model.to(lowerCamelCase	)
        model.eval()
        _UpperCamelCase  :							Optional[int]	    =	model(lowerCamelCase							,      attention_mask=lowerCamelCase							,      token_type_ids=lowerCamelCase	)
        _UpperCamelCase  :							Any	    =	model(lowerCamelCase							,      token_type_ids=lowerCamelCase	)
        _UpperCamelCase  :							Dict	    =	model(lowerCamelCase	)
        self.parent.assertEqual(result.pooler_output.shape							,      (self.batch_size, self.projection_dim or self.hidden_size)	)

 def        __snake_case       (      self							:		Dict							,      lowerCamelCase							:		Optional[Any]							,      lowerCamelCase							:		Union[str, Any]							,      lowerCamelCase							:		Any							,      lowerCamelCase							:		Dict							,      lowerCamelCase							:		Tuple							,      lowerCamelCase							:		Union[str, Any]							,      lowerCamelCase							:		Optional[Any]	)			->				Union[str, Any]:
        _UpperCamelCase  :							Optional[Any]	    =	DPRQuestionEncoder(config=lowerCamelCase	)
        model.to(lowerCamelCase	)
        model.eval()
        _UpperCamelCase  :							Any	    =	model(lowerCamelCase							,      attention_mask=lowerCamelCase							,      token_type_ids=lowerCamelCase	)
        _UpperCamelCase  :							Union[str, Any]	    =	model(lowerCamelCase							,      token_type_ids=lowerCamelCase	)
        _UpperCamelCase  :							int	    =	model(lowerCamelCase	)
        self.parent.assertEqual(result.pooler_output.shape							,      (self.batch_size, self.projection_dim or self.hidden_size)	)

 def        __snake_case       (      self							:		Optional[int]							,      lowerCamelCase							:		str							,      lowerCamelCase							:		Dict							,      lowerCamelCase							:		Union[str, Any]							,      lowerCamelCase							:		List[str]							,      lowerCamelCase							:		int							,      lowerCamelCase							:		Any							,      lowerCamelCase							:		List[str]	)			->				Any:
        _UpperCamelCase  :							Tuple	    =	DPRReader(config=lowerCamelCase	)
        model.to(lowerCamelCase	)
        model.eval()
        _UpperCamelCase  :							Union[str, Any]	    =	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        __snake_case       (      self							:		List[Any]	)			->				str:
        _UpperCamelCase  :							Optional[Any]	    =	self.prepare_config_and_inputs()
        (
            (
            _UpperCamelCase
        )    ,   (
            _UpperCamelCase
        )    ,   (
            _UpperCamelCase
        )    ,   (
            _UpperCamelCase
        )    ,   (
            _UpperCamelCase
        )    ,   (
            _UpperCamelCase
        )    ,   (
            _UpperCamelCase
        )    ,   
        )  :							Union[str, Any]	    =	config_and_inputs
        _UpperCamelCase  :							Dict	    =	{'input_ids': input_ids}
        return config, inputs_dict







@require_torch
class  __lowerCAmelCase	(							a_,						a_,						unittest.TestCase    ):



 '''simple docstring'''
 _lowerCamelCase    :						Union[str, Any]					     =			(
     (
         DPRContextEncoder,
         DPRQuestionEncoder,
         DPRReader,
     )
     if is_torch_available()
     else ()
 )
 _lowerCamelCase    :						Dict					     =			{'feature-extraction': DPRQuestionEncoder} if is_torch_available() else {}

 _lowerCamelCase    :						Union[str, Any]					     =			False
 _lowerCamelCase    :						List[Any]					     =			False  # why?
 _lowerCamelCase    :						Optional[Any]					     =			False
 _lowerCamelCase    :						List[Any]					     =			False

 def        __snake_case       (      self							:		Tuple	)			->				Optional[Any]:
        _UpperCamelCase  :							Union[str, Any]	    =	DPRModelTester(self	)
        _UpperCamelCase  :							Optional[int]	    =	ConfigTester(self							,      config_class=lowerCamelCase							,      hidden_size=3_7	)

 def        __snake_case       (      self							:		Tuple	)			->				List[str]:
        self.config_tester.run_common_tests()

 def        __snake_case       (      self							:		Dict	)			->				str:
        _UpperCamelCase  :							Union[str, Any]	    =	self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_context_encoder(*lowerCamelCase	)

 def        __snake_case       (      self							:		Optional[int]	)			->				str:
        _UpperCamelCase  :							Dict	    =	self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_question_encoder(*lowerCamelCase	)

 def        __snake_case       (      self							:		Union[str, Any]	)			->				Tuple:
        _UpperCamelCase  :							Union[str, Any]	    =	self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_reader(*lowerCamelCase	)

 def        __snake_case       (      self							:		Union[str, Any]	)			->				Dict:
        _UpperCamelCase  :							List[Any]	    =	self.model_tester.prepare_config_and_inputs()[0]

        _UpperCamelCase  :							Dict	    =	DPRQuestionEncoder(config=lowerCamelCase	)
        model.to(lowerCamelCase	)
        model.eval()

        with tempfile.TemporaryDirectory() as tmp_dirname:
               model.save_pretrained(lowerCamelCase	)
               _UpperCamelCase  :							List[Any]	    =	DPRQuestionEncoder.from_pretrained(lowerCamelCase							,      projection_dim=5_1_2	)

        self.assertIsNotNone(lowerCamelCase	)

 @slow
 def        __snake_case       (      self							:		int	)			->				int:
        for model_name in DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
               _UpperCamelCase  :							Dict	    =	DPRContextEncoder.from_pretrained(lowerCamelCase	)
               self.assertIsNotNone(lowerCamelCase	)

        for model_name in DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
               _UpperCamelCase  :							str	    =	DPRContextEncoder.from_pretrained(lowerCamelCase	)
               self.assertIsNotNone(lowerCamelCase	)

        for model_name in DPR_QUESTION_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
               _UpperCamelCase  :							Any	    =	DPRQuestionEncoder.from_pretrained(lowerCamelCase	)
               self.assertIsNotNone(lowerCamelCase	)

        for model_name in DPR_READER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
               _UpperCamelCase  :							List[str]	    =	DPRReader.from_pretrained(lowerCamelCase	)
               self.assertIsNotNone(lowerCamelCase	)







@require_torch
class  __lowerCAmelCase	(							unittest.TestCase    ):



 '''simple docstring'''

 @slow
 def        __snake_case       (      self							:		Tuple	)			->				List[str]:
        _UpperCamelCase  :							str	    =	DPRQuestionEncoder.from_pretrained('facebook/dpr-question_encoder-single-nq-base'							,      return_dict=lowerCamelCase	)
        model.to(lowerCamelCase	)

        _UpperCamelCase  :							Optional[Any]	    =	torch.tensor(
            [[1_0_1, 7_5_9_2, 1_0_1_0, 2_0_0_3, 2_0_2_6, 3_8_9_9, 1_0_1_4_0, 1_0_2_9, 1_0_2]]							,      dtype=torch.long							,      device=lowerCamelCase	)  # [CLS] hello, is my dog cute? [SEP]
        _UpperCamelCase  :							Union[str, Any]	    =	model(lowerCamelCase	)[0]  # embedding shape = (1, 768)
        # compare the actual values for a slice.
        _UpperCamelCase  :							Tuple	    =	torch.tensor(
            [
                [
                    0.0_3_2_3_6_2_5_3,
                    0.1_2_7_5_3_3_3_5,
                    0.1_6_8_1_8_5_0_9,
                    0.0_0_2_7_9_7_8_6,
                    0.3_8_9_6_9_3_3,
                    0.2_4_2_6_4_9_4_5,
                    0.2_1_7_8_9_7_1,
                    -0.0_2_3_3_5_2_2_7,
                    -0.0_8_4_8_1_9_5_9,
                    -0.1_4_3_2_4_1_1_7,
                ]
            ]							,      dtype=torch.float							,      device=lowerCamelCase							,      )
        self.assertTrue(torch.allclose(output[:, :1_0]							,      lowerCamelCase							,      atol=1e-4	)	)

 @slow
 def        __snake_case       (      self							:		int	)			->				Optional[int]:
        _UpperCamelCase  :							Optional[int]	    =	DPRReaderTokenizer.from_pretrained('facebook/dpr-reader-single-nq-base'	)
        _UpperCamelCase  :							Optional[Any]	    =	DPRReader.from_pretrained('facebook/dpr-reader-single-nq-base'	)
        model.to(lowerCamelCase	)

        _UpperCamelCase  :							Tuple	    =	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	)

        _UpperCamelCase  :							int	    =	model(**lowerCamelCase	)

        # compare the actual values for a slice.
        _UpperCamelCase  :							Optional[int]	    =	torch.tensor(
            [[-1_0.3_0_0_5, -1_0.7_7_6_5, -1_1.4_8_7_2, -1_1.6_8_4_1, -1_1.9_3_1_2, -1_0.3_0_0_2, -9.8_5_4_4, -1_1.7_3_7_8, -1_2.0_8_2_1, -1_0.2_9_7_5]]							,      dtype=torch.float							,      device=lowerCamelCase							,      )

        _UpperCamelCase  :							List[Any]	    =	torch.tensor(
            [[-1_1.0_6_8_4, -1_1.7_0_4_1, -1_1.5_3_9_7, -1_0.3_4_6_5, -1_0.8_7_9_1, -6.8_4_4_3, -1_1.9_9_5_9, -1_1.0_3_6_4, -1_0.0_0_9_6, -6.8_4_0_5]]							,      dtype=torch.float							,      device=lowerCamelCase							,      )
        self.assertTrue(torch.allclose(outputs.start_logits[:, :1_0]							,      lowerCamelCase							,      atol=1e-4	)	)
        self.assertTrue(torch.allclose(outputs.end_logits[:, :1_0]							,      lowerCamelCase							,      atol=1e-4	)	)





'''simple docstring'''







from ..utils import DummyObject, requires_backends


class       lowerCAmelCase			(					metaclass=__UpperCAmelCase      ):
		a				:       Any							=       ["""torch""", """transformers""", """invisible_watermark"""]




		def __init__(					self       ,    *UpperCamelCase       ,    **UpperCamelCase   ):
								requires_backends(self       ,    ["torch", "transformers", "invisible_watermark"]   )




		@classmethod
		def      lowercase							(					cls       ,    *UpperCamelCase       ,    **UpperCamelCase   ):
								requires_backends(cls       ,    ["torch", "transformers", "invisible_watermark"]   )




		@classmethod
		def      lowercase							(					cls       ,    *UpperCamelCase       ,    **UpperCamelCase   ):
								requires_backends(cls       ,    ["torch", "transformers", "invisible_watermark"]   )


class       lowerCAmelCase			(					metaclass=__UpperCAmelCase      ):
		a				:       Optional[Any]							=       ["""torch""", """transformers""", """invisible_watermark"""]




		def __init__(					self       ,    *UpperCamelCase       ,    **UpperCamelCase   ):
								requires_backends(self       ,    ["torch", "transformers", "invisible_watermark"]   )




		@classmethod
		def      lowercase							(					cls       ,    *UpperCamelCase       ,    **UpperCamelCase   ):
								requires_backends(cls       ,    ["torch", "transformers", "invisible_watermark"]   )




		@classmethod
		def      lowercase							(					cls       ,    *UpperCamelCase       ,    **UpperCamelCase   ):
								requires_backends(cls       ,    ["torch", "transformers", "invisible_watermark"]   )