import os
from shutil import copyfile
from typing import Optional, Tuple

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging


if is_sentencepiece_available():
			from .tokenization_reformer import ReformerTokenizer
else:
			_SCREAMING_SNAKE_CASE:    List[str]     						=							None


_SCREAMING_SNAKE_CASE:    Union[str, Any]     						=							logging.get_logger(__name__)


_SCREAMING_SNAKE_CASE:    Tuple     						=							'''▁'''

_SCREAMING_SNAKE_CASE:    Dict     						=							{'''vocab_file''': '''spiece.model''', '''tokenizer_file''': '''tokenizer.json'''}

_SCREAMING_SNAKE_CASE:    Optional[int]     						=							{
    '''vocab_file''': {
        '''google/reformer-crime-and-punishment''': (
            '''https://huggingface.co/google/reformer-crime-and-punishment/resolve/main/spiece.model'''
        )
    },
    '''tokenizer_file''': {
        '''google/reformer-crime-and-punishment''': (
            '''https://huggingface.co/google/reformer-crime-and-punishment/resolve/main/tokenizer.json'''
        )
    },
}

_SCREAMING_SNAKE_CASE:    str     						=							{
    '''google/reformer-crime-and-punishment''': 524_288,
}

class 						_SCREAMING_SNAKE_CASE				(  _lowercase				):






		"""simple docstring"""







		A_    			=   VOCAB_FILES_NAMES
		A_    			=   PRETRAINED_VOCAB_FILES_MAP
		A_    			=   PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
		A_    			=   ['''input_ids''', '''attention_mask''']
		A_    			=   ReformerTokenizer



		def __init__(      self     :					Union[str, Any]	,       __SCREAMING_SNAKE_CASE     :					Union[str, Any]=None	,       __SCREAMING_SNAKE_CASE     :					Union[str, Any]=None	,       __SCREAMING_SNAKE_CASE     :					Any="</s>"	,       __SCREAMING_SNAKE_CASE     :					Union[str, Any]="<unk>"	,       __SCREAMING_SNAKE_CASE     :					List[Any]=[]	,       **__SCREAMING_SNAKE_CASE     :					Optional[Any]	,       ):






					"""simple docstring"""
					super().__init__(
					    __SCREAMING_SNAKE_CASE	,       tokenizer_file=__SCREAMING_SNAKE_CASE	,       eos_token=__SCREAMING_SNAKE_CASE	,       unk_token=__SCREAMING_SNAKE_CASE	,       additional_special_tokens=__SCREAMING_SNAKE_CASE	,       **__SCREAMING_SNAKE_CASE	,       )

					_snake_case														=   vocab_file
					_snake_case														=   False if not self.vocab_file else True







		def 					_a					(      self     :					List[str]	,       __SCREAMING_SNAKE_CASE     :					str	,       __SCREAMING_SNAKE_CASE     :					Optional[str] = None ):






					"""simple docstring"""
					if not self.can_save_slow_tokenizer:
								raise ValueError(
								    "Your fast tokenizer does not have the necessary information to save the vocabulary for a slow "
								    "tokenizer." )

					if not os.path.isdir(__SCREAMING_SNAKE_CASE ):
								logger.error(F'Vocabulary path ({save_directory}) should be a directory' )
								return
					_snake_case														=   os.path.join(
					    __SCREAMING_SNAKE_CASE	,       (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"] )

					if os.path.abspath(self.vocab_file ) != os.path.abspath(__SCREAMING_SNAKE_CASE ):
								copyfile(self.vocab_file	,       __SCREAMING_SNAKE_CASE )

					return (out_vocab_file,)


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






		"""simple docstring"""







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






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



		def 					_a					(      self     :					Tuple ):






					"""simple docstring"""
					_snake_case														=   ids_tensor([self.batch_size, self.seq_length]	,       self.vocab_size )

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

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

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

					_snake_case														=   self.get_config()

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



		def 					_a					(      self     :					List[Any] ):






					"""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 					_a					(      self     :					List[Any]	,       __SCREAMING_SNAKE_CASE     :					Union[str, Any]	,       __SCREAMING_SNAKE_CASE     :					List[str]	,       __SCREAMING_SNAKE_CASE     :					List[str]	,       __SCREAMING_SNAKE_CASE     :					List[Any]	,       __SCREAMING_SNAKE_CASE     :					Any	,       __SCREAMING_SNAKE_CASE     :					List[Any]	,       __SCREAMING_SNAKE_CASE     :					List[str] ):






					"""simple docstring"""
					_snake_case														=   DPRContextEncoder(config=__SCREAMING_SNAKE_CASE )
					model.to(__SCREAMING_SNAKE_CASE )
					model.eval()
					_snake_case														=   model(__SCREAMING_SNAKE_CASE	,       attention_mask=__SCREAMING_SNAKE_CASE	,       token_type_ids=__SCREAMING_SNAKE_CASE )
					_snake_case														=   model(__SCREAMING_SNAKE_CASE	,       token_type_ids=__SCREAMING_SNAKE_CASE )
					_snake_case														=   model(__SCREAMING_SNAKE_CASE )
					self.parent.assertEqual(result.pooler_output.shape	,       (self.batch_size, self.projection_dim or self.hidden_size) )



		def 					_a					(      self     :					str	,       __SCREAMING_SNAKE_CASE     :					Optional[Any]	,       __SCREAMING_SNAKE_CASE     :					List[str]	,       __SCREAMING_SNAKE_CASE     :					List[str]	,       __SCREAMING_SNAKE_CASE     :					List[str]	,       __SCREAMING_SNAKE_CASE     :					Dict	,       __SCREAMING_SNAKE_CASE     :					Optional[Any]	,       __SCREAMING_SNAKE_CASE     :					str ):






					"""simple docstring"""
					_snake_case														=   DPRQuestionEncoder(config=__SCREAMING_SNAKE_CASE )
					model.to(__SCREAMING_SNAKE_CASE )
					model.eval()
					_snake_case														=   model(__SCREAMING_SNAKE_CASE	,       attention_mask=__SCREAMING_SNAKE_CASE	,       token_type_ids=__SCREAMING_SNAKE_CASE )
					_snake_case														=   model(__SCREAMING_SNAKE_CASE	,       token_type_ids=__SCREAMING_SNAKE_CASE )
					_snake_case														=   model(__SCREAMING_SNAKE_CASE )
					self.parent.assertEqual(result.pooler_output.shape	,       (self.batch_size, self.projection_dim or self.hidden_size) )



		def 					_a					(      self     :					List[str]	,       __SCREAMING_SNAKE_CASE     :					int	,       __SCREAMING_SNAKE_CASE     :					List[Any]	,       __SCREAMING_SNAKE_CASE     :					int	,       __SCREAMING_SNAKE_CASE     :					Union[str, Any]	,       __SCREAMING_SNAKE_CASE     :					Any	,       __SCREAMING_SNAKE_CASE     :					Optional[int]	,       __SCREAMING_SNAKE_CASE     :					Any ):






					"""simple docstring"""
					_snake_case														=   DPRReader(config=__SCREAMING_SNAKE_CASE )
					model.to(__SCREAMING_SNAKE_CASE )
					model.eval()
					_snake_case														=   model(
					    __SCREAMING_SNAKE_CASE	,       attention_mask=__SCREAMING_SNAKE_CASE	,       )

					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 					_a					(      self     :					Any ):






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

@require_torch
class 						_SCREAMING_SNAKE_CASE				(  _lowercase   ,  _lowercase   ,  unittest.TestCase				):






		"""simple docstring"""







		A_    			=   (
		    (
		        DPRContextEncoder,
		        DPRQuestionEncoder,
		        DPRReader,
		    )
		    if is_torch_available()
		    else ()
		)
		A_    			=   {'''feature-extraction''': DPRQuestionEncoder} if is_torch_available() else {}

		A_    			=   False
		A_    			=   False  # why?
		A_    			=   False
		A_    			=   False



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






					"""simple docstring"""
					_snake_case														=   DPRModelTester(self )
					_snake_case														=   ConfigTester(self	,       config_class=__SCREAMING_SNAKE_CASE	,       hidden_size=37 )



		def 					_a					(      self     :					int ):






					"""simple docstring"""
					self.config_tester.run_common_tests()



		def 					_a					(      self     :					List[str] ):






					"""simple docstring"""
					_snake_case														=   self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_context_encoder(*__SCREAMING_SNAKE_CASE )



		def 					_a					(      self     :					List[str] ):






					"""simple docstring"""
					_snake_case														=   self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_question_encoder(*__SCREAMING_SNAKE_CASE )



		def 					_a					(      self     :					Optional[Any] ):






					"""simple docstring"""
					_snake_case														=   self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_reader(*__SCREAMING_SNAKE_CASE )



		def 					_a					(      self     :					Optional[Any] ):






					"""simple docstring"""
					_snake_case														=   self.model_tester.prepare_config_and_inputs()[0]

					_snake_case														=   DPRQuestionEncoder(config=__SCREAMING_SNAKE_CASE )
					model.to(__SCREAMING_SNAKE_CASE )
					model.eval()

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

					self.assertIsNotNone(__SCREAMING_SNAKE_CASE )







		@slow
		def 					_a					(      self     :					List[str] ):






					"""simple docstring"""
					for model_name in DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
								_snake_case														=   DPRContextEncoder.from_pretrained(__SCREAMING_SNAKE_CASE )
								self.assertIsNotNone(__SCREAMING_SNAKE_CASE )

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

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

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

@require_torch
class 						_SCREAMING_SNAKE_CASE				(  unittest.TestCase				):






		"""simple docstring"""







		@slow
		def 					_a					(      self     :					Dict ):






					"""simple docstring"""
					_snake_case														=   DPRQuestionEncoder.from_pretrained("facebook/dpr-question_encoder-single-nq-base"	,       return_dict=__SCREAMING_SNAKE_CASE )
					model.to(__SCREAMING_SNAKE_CASE )

					_snake_case														=   torch.tensor(
					    [[101, 7_592, 1_010, 2_003, 2_026, 3_899, 10_140, 1_029, 102]]	,       dtype=torch.long	,       device=__SCREAMING_SNAKE_CASE )  # [CLS] hello, is my dog cute? [SEP]
					_snake_case														=   model(__SCREAMING_SNAKE_CASE )[0]  # embedding shape = (1, 768)
					# compare the actual values for a slice.
					_snake_case														=   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=__SCREAMING_SNAKE_CASE	,       )
					self.assertTrue(torch.allclose(output[:, :10]	,       __SCREAMING_SNAKE_CASE	,       atol=1e-4 ) )







		@slow
		def 					_a					(      self     :					Optional[int] ):






					"""simple docstring"""
					_snake_case														=   DPRReaderTokenizer.from_pretrained("facebook/dpr-reader-single-nq-base" )
					_snake_case														=   DPRReader.from_pretrained("facebook/dpr-reader-single-nq-base" )
					model.to(__SCREAMING_SNAKE_CASE )

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

					_snake_case														=   model(**__SCREAMING_SNAKE_CASE )

					# compare the actual values for a slice.
					_snake_case														=   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=__SCREAMING_SNAKE_CASE	,       )

					_snake_case														=   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=__SCREAMING_SNAKE_CASE	,       )
					self.assertTrue(torch.allclose(outputs.start_logits[:, :10]	,       __SCREAMING_SNAKE_CASE	,       atol=1e-4 ) )
					self.assertTrue(torch.allclose(outputs.end_logits[:, :10]	,       __SCREAMING_SNAKE_CASE	,       atol=1e-4 ) )

