

from __future__ import annotations

import unittest

from transformers import is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
     import numpy
     import tensorflow as tf

     from transformers import (
         TF_DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST,
         TF_DPR_QUESTION_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST,
         TF_DPR_READER_PRETRAINED_MODEL_ARCHIVE_LIST,
         BertConfig,
         DPRConfig,
         TFDPRContextEncoder,
         TFDPRQuestionEncoder,
         TFDPRReader,
     )


class 			A						:







    """simple docstring"""





    def __init__(				self			:	str,lowercase_			:	int,lowercase_			:	Optional[Any]=1_3,lowercase_			:	Optional[int]=7,lowercase_			:	str=True,lowercase_			:	Optional[int]=True,lowercase_			:	Dict=True,lowercase_			:	str=True,lowercase_			:	Union[str, Any]=9_9,lowercase_			:	Union[str, Any]=3_2,lowercase_			:	Optional[Any]=2,lowercase_			:	int=4,lowercase_			:	str=3_7,lowercase_			:	int="gelu",lowercase_			:	Dict=0.1,lowercase_			:	Union[str, Any]=0.1,lowercase_			:	Optional[int]=5_1_2,lowercase_			:	Union[str, Any]=1_6,lowercase_			:	Any=2,lowercase_			:	Dict=0.02,lowercase_			:	int=3,lowercase_			:	Dict=4,lowercase_			:	str=None,lowercase_			:	Optional[int]=0,)->   Optional[int]:





        '''simple docstring'''







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


    def 						snake_case__					(				self			:	List[str]		)->   Tuple:





        '''simple docstring'''







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

        A__   				=       None
        if self.use_input_mask:
            # follow test_modeling_tf_ctrl.py
            A__   				=       random_attention_mask([self.batch_size, self.seq_length]		)

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

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

        A__   				=       BertConfig(
            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,is_decoder=lowercase_,initializer_range=self.initializer_range,)
        A__   				=       DPRConfig(projection_dim=self.projection_dim,**config.to_dict()		)

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


    def 						snake_case__					(				self			:	List[Any],lowercase_			:	Optional[int],lowercase_			:	Optional[Any],lowercase_			:	str,lowercase_			:	int,lowercase_			:	Tuple,lowercase_			:	List[Any],lowercase_			:	Dict		)->   List[str]:





        '''simple docstring'''







        A__   				=       TFDPRContextEncoder(config=lowercase_		)
        A__   				=       model(lowercase_,attention_mask=lowercase_,token_type_ids=lowercase_		)
        A__   				=       model(lowercase_,token_type_ids=lowercase_		)
        A__   				=       model(lowercase_		)
        self.parent.assertEqual(result.pooler_output.shape,(self.batch_size, self.projection_dim or self.hidden_size)		)


    def 						snake_case__					(				self			:	Optional[int],lowercase_			:	Optional[Any],lowercase_			:	Tuple,lowercase_			:	Any,lowercase_			:	Any,lowercase_			:	Union[str, Any],lowercase_			:	Union[str, Any],lowercase_			:	str		)->   str:





        '''simple docstring'''







        A__   				=       TFDPRQuestionEncoder(config=lowercase_		)
        A__   				=       model(lowercase_,attention_mask=lowercase_,token_type_ids=lowercase_		)
        A__   				=       model(lowercase_,token_type_ids=lowercase_		)
        A__   				=       model(lowercase_		)
        self.parent.assertEqual(result.pooler_output.shape,(self.batch_size, self.projection_dim or self.hidden_size)		)


    def 						snake_case__					(				self			:	int,lowercase_			:	List[Any],lowercase_			:	Optional[Any],lowercase_			:	Any,lowercase_			:	Any,lowercase_			:	List[Any],lowercase_			:	Union[str, Any],lowercase_			:	Tuple		)->   List[Any]:





        '''simple docstring'''







        A__   				=       TFDPRReader(config=lowercase_		)
        A__   				=       model(lowercase_,attention_mask=lowercase_		)

        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			:	int		)->   Union[str, Any]:





        '''simple docstring'''







        A__   				=       self.prepare_config_and_inputs()
        (
            (
            A__
        )				,						(
            A__
        )				,						(
            A__
        )				,						(
            A__
        )				,						(
            A__
        )				,						(
            A__
        )				,						(
            A__
        )				,						
        )   				=       config_and_inputs
        A__   				=       {'input_ids': input_ids}
        return config, inputs_dict




@require_tf
class 			A						(							_UpperCAmelCase   ,							_UpperCAmelCase   ,							unittest.TestCase			):







    """simple docstring"""





    lowerCamelCase		     =							(
        (
            TFDPRContextEncoder,
            TFDPRQuestionEncoder,
            TFDPRReader,
        )
        if is_tf_available()
        else ()
    )
    lowerCamelCase		     =							{'feature-extraction': TFDPRQuestionEncoder} if is_tf_available() else {}

    lowerCamelCase		     =							False
    lowerCamelCase		     =							False
    lowerCamelCase		     =							False
    lowerCamelCase		     =							False
    lowerCamelCase		     =							False


    def 						snake_case__					(				self			:	str		)->   List[Any]:





        '''simple docstring'''







        A__   				=       TFDPRModelTester(self		)
        A__   				=       ConfigTester(self,config_class=lowercase_,hidden_size=3_7		)


    def 						snake_case__					(				self			:	Dict		)->   Any:





        '''simple docstring'''







        self.config_tester.run_common_tests()


    def 						snake_case__					(				self			:	Tuple		)->   Tuple:





        '''simple docstring'''







        A__   				=       self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_dpr_context_encoder(*lowercase_		)


    def 						snake_case__					(				self			:	Optional[int]		)->   Union[str, Any]:





        '''simple docstring'''







        A__   				=       self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_dpr_question_encoder(*lowercase_		)


    def 						snake_case__					(				self			:	List[Any]		)->   int:





        '''simple docstring'''







        A__   				=       self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_dpr_reader(*lowercase_		)


    @slow
    def 						snake_case__					(				self			:	Any		)->   Any:





        '''simple docstring'''







        for model_name in TF_DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
            A__   				=       TFDPRContextEncoder.from_pretrained(lowercase_		)
            self.assertIsNotNone(lowercase_		)

        for model_name in TF_DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
            A__   				=       TFDPRContextEncoder.from_pretrained(lowercase_		)
            self.assertIsNotNone(lowercase_		)

        for model_name in TF_DPR_QUESTION_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
            A__   				=       TFDPRQuestionEncoder.from_pretrained(lowercase_		)
            self.assertIsNotNone(lowercase_		)

        for model_name in TF_DPR_READER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
            A__   				=       TFDPRReader.from_pretrained(lowercase_		)
            self.assertIsNotNone(lowercase_		)




@require_tf
class 			A						(							unittest.TestCase			):







    """simple docstring"""





    @slow
    def 						snake_case__					(				self			:	int		)->   Dict:





        '''simple docstring'''







        A__   				=       TFDPRQuestionEncoder.from_pretrained('facebook/dpr-question_encoder-single-nq-base'		)

        A__   				=       tf.constant(
            [[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]]		)  # [CLS] hello, is my dog cute? [SEP]
        A__   				=       model(lowercase_		)[0]  # embedding shape = (1, 768)
        # compare the actual values for a slice.
        A__   				=       tf.constant(
            [
                [
                    0.03_236_253,
                    0.12_753_335,
                    0.16_818_509,
                    0.00_279_786,
                    0.3_896_933,
                    0.24_264_945,
                    0.2_178_971,
                    -0.02_335_227,
                    -0.08_481_959,
                    -0.14_324_117,
                ]
            ]		)
        self.assertTrue(numpy.allclose(output[:, :1_0].numpy(),expected_slice.numpy(),atol=1E-4		)		)




import os
import tempfile
import unittest
from pathlib import Path

from transformers import AutoConfig, is_torch_available
from transformers.testing_utils import require_torch, torch_device


if is_torch_available():
     from transformers import PyTorchBenchmark, PyTorchBenchmarkArguments


@require_torch
class 			A						(							unittest.TestCase			):







    """simple docstring"""





    def 						snake_case__					(				self			:	List[Any],lowercase_			:	str		)->   List[Any]:





        '''simple docstring'''







        for model_result in results.values():
            for batch_size, sequence_length in zip(model_result['bs'],model_result['ss']		):
                A__   				=       model_result['result'][batch_size][sequence_length]
                self.assertIsNotNone(lowercase_		)


    def 						snake_case__					(				self			:	Dict		)->   List[str]:





        '''simple docstring'''







        A__   				=       'sshleifer/tiny-gpt2'
        A__   				=       PyTorchBenchmarkArguments(
            models=[MODEL_ID],training=lowercase_,inference=lowercase_,sequence_lengths=[8],batch_sizes=[1],multi_process=lowercase_,)
        A__   				=       PyTorchBenchmark(lowercase_		)
        A__   				=       benchmark.run()
        self.check_results_dict_not_empty(results.time_inference_result		)
        self.check_results_dict_not_empty(results.memory_inference_result		)


    def 						snake_case__					(				self			:	Dict		)->   List[str]:





        '''simple docstring'''







        A__   				=       'sgugger/tiny-distilbert-classification'
        A__   				=       PyTorchBenchmarkArguments(
            models=[MODEL_ID],training=lowercase_,inference=lowercase_,sequence_lengths=[8],batch_sizes=[1],multi_process=lowercase_,only_pretrain_model=lowercase_,)
        A__   				=       PyTorchBenchmark(lowercase_		)
        A__   				=       benchmark.run()
        self.check_results_dict_not_empty(results.time_inference_result		)
        self.check_results_dict_not_empty(results.memory_inference_result		)


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





        '''simple docstring'''







        A__   				=       'sshleifer/tiny-gpt2'
        A__   				=       PyTorchBenchmarkArguments(
            models=[MODEL_ID],training=lowercase_,inference=lowercase_,torchscript=lowercase_,sequence_lengths=[8],batch_sizes=[1],multi_process=lowercase_,)
        A__   				=       PyTorchBenchmark(lowercase_		)
        A__   				=       benchmark.run()
        self.check_results_dict_not_empty(results.time_inference_result		)
        self.check_results_dict_not_empty(results.memory_inference_result		)


    @unittest.skipIf(torch_device == 'cpu','Cant do half precision'		)
    def 						snake_case__					(				self			:	Any		)->   Dict:





        '''simple docstring'''







        A__   				=       'sshleifer/tiny-gpt2'
        A__   				=       PyTorchBenchmarkArguments(
            models=[MODEL_ID],training=lowercase_,inference=lowercase_,fpaa=lowercase_,sequence_lengths=[8],batch_sizes=[1],multi_process=lowercase_,)
        A__   				=       PyTorchBenchmark(lowercase_		)
        A__   				=       benchmark.run()
        self.check_results_dict_not_empty(results.time_inference_result		)
        self.check_results_dict_not_empty(results.memory_inference_result		)


    def 						snake_case__					(				self			:	Any		)->   Optional[Any]:





        '''simple docstring'''







        A__   				=       'sshleifer/tiny-gpt2'
        A__   				=       AutoConfig.from_pretrained(lowercase_		)
        # set architectures equal to `None`
        A__   				=       None
        A__   				=       PyTorchBenchmarkArguments(
            models=[MODEL_ID],training=lowercase_,inference=lowercase_,sequence_lengths=[8],batch_sizes=[1],multi_process=lowercase_,)
        A__   				=       PyTorchBenchmark(lowercase_,configs=[config]		)
        A__   				=       benchmark.run()
        self.check_results_dict_not_empty(results.time_inference_result		)
        self.check_results_dict_not_empty(results.memory_inference_result		)


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





        '''simple docstring'''







        A__   				=       'sshleifer/tiny-gpt2'
        A__   				=       PyTorchBenchmarkArguments(
            models=[MODEL_ID],training=lowercase_,inference=lowercase_,sequence_lengths=[8],batch_sizes=[1],multi_process=lowercase_,)
        A__   				=       PyTorchBenchmark(lowercase_		)
        A__   				=       benchmark.run()
        self.check_results_dict_not_empty(results.time_train_result		)
        self.check_results_dict_not_empty(results.memory_train_result		)


    @unittest.skipIf(torch_device == 'cpu','Can\'t do half precision'		)
    def 						snake_case__					(				self			:	List[Any]		)->   Dict:





        '''simple docstring'''







        A__   				=       'sshleifer/tiny-gpt2'
        A__   				=       PyTorchBenchmarkArguments(
            models=[MODEL_ID],training=lowercase_,inference=lowercase_,sequence_lengths=[8],batch_sizes=[1],fpaa=lowercase_,multi_process=lowercase_,)
        A__   				=       PyTorchBenchmark(lowercase_		)
        A__   				=       benchmark.run()
        self.check_results_dict_not_empty(results.time_train_result		)
        self.check_results_dict_not_empty(results.memory_train_result		)


    def 						snake_case__					(				self			:	int		)->   Optional[int]:





        '''simple docstring'''







        A__   				=       'sshleifer/tiny-gpt2'
        A__   				=       AutoConfig.from_pretrained(lowercase_		)
        A__   				=       PyTorchBenchmarkArguments(
            models=[MODEL_ID],training=lowercase_,inference=lowercase_,sequence_lengths=[8],batch_sizes=[1],multi_process=lowercase_,)
        A__   				=       PyTorchBenchmark(lowercase_,configs=[config]		)
        A__   				=       benchmark.run()
        self.check_results_dict_not_empty(results.time_inference_result		)
        self.check_results_dict_not_empty(results.memory_inference_result		)


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





        '''simple docstring'''







        A__   				=       'sshleifer/tinier_bart'
        A__   				=       AutoConfig.from_pretrained(lowercase_		)
        A__   				=       PyTorchBenchmarkArguments(
            models=[MODEL_ID],training=lowercase_,inference=lowercase_,sequence_lengths=[8],batch_sizes=[1],multi_process=lowercase_,)
        A__   				=       PyTorchBenchmark(lowercase_,configs=[config]		)
        A__   				=       benchmark.run()
        self.check_results_dict_not_empty(results.time_inference_result		)
        self.check_results_dict_not_empty(results.memory_inference_result		)


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





        '''simple docstring'''







        A__   				=       'sshleifer/tiny-gpt2'
        A__   				=       AutoConfig.from_pretrained(lowercase_		)
        A__   				=       PyTorchBenchmarkArguments(
            models=[MODEL_ID],training=lowercase_,inference=lowercase_,sequence_lengths=[8],batch_sizes=[1],multi_process=lowercase_,)
        A__   				=       PyTorchBenchmark(lowercase_,configs=[config]		)
        A__   				=       benchmark.run()
        self.check_results_dict_not_empty(results.time_train_result		)
        self.check_results_dict_not_empty(results.memory_train_result		)


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





        '''simple docstring'''







        A__   				=       'sshleifer/tinier_bart'
        A__   				=       AutoConfig.from_pretrained(lowercase_		)
        A__   				=       PyTorchBenchmarkArguments(
            models=[MODEL_ID],training=lowercase_,inference=lowercase_,sequence_lengths=[8],batch_sizes=[1],multi_process=lowercase_,)
        A__   				=       PyTorchBenchmark(lowercase_,configs=[config]		)
        A__   				=       benchmark.run()
        self.check_results_dict_not_empty(results.time_train_result		)
        self.check_results_dict_not_empty(results.memory_train_result		)


    def 						snake_case__					(				self			:	Optional[Any]		)->   Tuple:





        '''simple docstring'''







        A__   				=       'sshleifer/tiny-gpt2'
        with tempfile.TemporaryDirectory() as tmp_dir:
            A__   				=       PyTorchBenchmarkArguments(
                models=[MODEL_ID],training=lowercase_,inference=lowercase_,save_to_csv=lowercase_,sequence_lengths=[8],batch_sizes=[1],inference_time_csv_file=os.path.join(lowercase_,'inf_time.csv'		),train_memory_csv_file=os.path.join(lowercase_,'train_mem.csv'		),inference_memory_csv_file=os.path.join(lowercase_,'inf_mem.csv'		),train_time_csv_file=os.path.join(lowercase_,'train_time.csv'		),env_info_csv_file=os.path.join(lowercase_,'env.csv'		),multi_process=lowercase_,)
            A__   				=       PyTorchBenchmark(lowercase_		)
            benchmark.run()
            self.assertTrue(Path(os.path.join(lowercase_,'inf_time.csv'		)		).exists()		)
            self.assertTrue(Path(os.path.join(lowercase_,'train_time.csv'		)		).exists()		)
            self.assertTrue(Path(os.path.join(lowercase_,'inf_mem.csv'		)		).exists()		)
            self.assertTrue(Path(os.path.join(lowercase_,'train_mem.csv'		)		).exists()		)
            self.assertTrue(Path(os.path.join(lowercase_,'env.csv'		)		).exists()		)


    def 						snake_case__					(				self			:	Tuple		)->   str:





        '''simple docstring'''







        A__   				=       'sshleifer/tiny-gpt2'

        def _check_summary_is_not_empty(lowercase_			:	Optional[Any]		):
            self.assertTrue(hasattr(lowercase_,'sequential'		)		)
            self.assertTrue(hasattr(lowercase_,'cumulative'		)		)
            self.assertTrue(hasattr(lowercase_,'current'		)		)
            self.assertTrue(hasattr(lowercase_,'total'		)		)

        with tempfile.TemporaryDirectory() as tmp_dir:
            A__   				=       PyTorchBenchmarkArguments(
                models=[MODEL_ID],training=lowercase_,inference=lowercase_,sequence_lengths=[8],batch_sizes=[1],log_filename=os.path.join(lowercase_,'log.txt'		),log_print=lowercase_,trace_memory_line_by_line=lowercase_,multi_process=lowercase_,)
            A__   				=       PyTorchBenchmark(lowercase_		)
            A__   				=       benchmark.run()
            _check_summary_is_not_empty(result.inference_summary		)
            _check_summary_is_not_empty(result.train_summary		)
            self.assertTrue(Path(os.path.join(lowercase_,'log.txt'		)		).exists()		)

