
"""simple docstring"""




import collections
from typing import List, Optional, Union

from ...tokenization_utils_base import BatchEncoding
from ...utils import TensorType, add_end_docstrings, add_start_docstrings, logging
from ..bert.tokenization_bert import BertTokenizer


SCREAMING_SNAKE_CASE							:int										=    logging.get_logger(__name__)

SCREAMING_SNAKE_CASE							:Union[str, Any]										=    {'vocab_file': 'vocab.txt', 'tokenizer_file': 'tokenizer.json'}

SCREAMING_SNAKE_CASE							:Any										=    {
    'vocab_file': {
        'facebook/dpr-ctx_encoder-single-nq-base': (
            'https://huggingface.co/facebook/dpr-ctx_encoder-single-nq-base/resolve/main/vocab.txt'
        ),
        'facebook/dpr-ctx_encoder-multiset-base': (
            'https://huggingface.co/facebook/dpr-ctx_encoder-multiset-base/resolve/main/vocab.txt'
        ),
    },
    'tokenizer_file': {
        'facebook/dpr-ctx_encoder-single-nq-base': (
            'https://huggingface.co/facebook/dpr-ctx_encoder-single-nq-base/resolve/main/tokenizer.json'
        ),
        'facebook/dpr-ctx_encoder-multiset-base': (
            'https://huggingface.co/facebook/dpr-ctx_encoder-multiset-base/resolve/main/tokenizer.json'
        ),
    },
}
SCREAMING_SNAKE_CASE							:Any										=    {
    'vocab_file': {
        'facebook/dpr-question_encoder-single-nq-base': (
            'https://huggingface.co/facebook/dpr-question_encoder-single-nq-base/resolve/main/vocab.txt'
        ),
        'facebook/dpr-question_encoder-multiset-base': (
            'https://huggingface.co/facebook/dpr-question_encoder-multiset-base/resolve/main/vocab.txt'
        ),
    },
    'tokenizer_file': {
        'facebook/dpr-question_encoder-single-nq-base': (
            'https://huggingface.co/facebook/dpr-question_encoder-single-nq-base/resolve/main/tokenizer.json'
        ),
        'facebook/dpr-question_encoder-multiset-base': (
            'https://huggingface.co/facebook/dpr-question_encoder-multiset-base/resolve/main/tokenizer.json'
        ),
    },
}
SCREAMING_SNAKE_CASE							:Dict										=    {
    'vocab_file': {
        'facebook/dpr-reader-single-nq-base': (
            'https://huggingface.co/facebook/dpr-reader-single-nq-base/resolve/main/vocab.txt'
        ),
        'facebook/dpr-reader-multiset-base': (
            'https://huggingface.co/facebook/dpr-reader-multiset-base/resolve/main/vocab.txt'
        ),
    },
    'tokenizer_file': {
        'facebook/dpr-reader-single-nq-base': (
            'https://huggingface.co/facebook/dpr-reader-single-nq-base/resolve/main/tokenizer.json'
        ),
        'facebook/dpr-reader-multiset-base': (
            'https://huggingface.co/facebook/dpr-reader-multiset-base/resolve/main/tokenizer.json'
        ),
    },
}

SCREAMING_SNAKE_CASE							:List[str]										=    {
    'facebook/dpr-ctx_encoder-single-nq-base': 512,
    'facebook/dpr-ctx_encoder-multiset-base': 512,
}
SCREAMING_SNAKE_CASE							:Tuple										=    {
    'facebook/dpr-question_encoder-single-nq-base': 512,
    'facebook/dpr-question_encoder-multiset-base': 512,
}
SCREAMING_SNAKE_CASE							:List[str]										=    {
    'facebook/dpr-reader-single-nq-base': 512,
    'facebook/dpr-reader-multiset-base': 512,
}


SCREAMING_SNAKE_CASE							:Optional[Any]										=    {
    'facebook/dpr-ctx_encoder-single-nq-base': {'do_lower_case': True},
    'facebook/dpr-ctx_encoder-multiset-base': {'do_lower_case': True},
}
SCREAMING_SNAKE_CASE							:List[Any]										=    {
    'facebook/dpr-question_encoder-single-nq-base': {'do_lower_case': True},
    'facebook/dpr-question_encoder-multiset-base': {'do_lower_case': True},
}
SCREAMING_SNAKE_CASE							:Optional[Any]										=    {
    'facebook/dpr-reader-single-nq-base': {'do_lower_case': True},
    'facebook/dpr-reader-multiset-base': {'do_lower_case': True},
}





class 							UpperCAmelCase				(  __lowerCAmelCase  ):




	'''simple docstring'''





	snake_case_    =					VOCAB_FILES_NAMES
	snake_case_    =					CONTEXT_ENCODER_PRETRAINED_VOCAB_FILES_MAP
	snake_case_    =					CONTEXT_ENCODER_PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	snake_case_    =					CONTEXT_ENCODER_PRETRAINED_INIT_CONFIGURATION





class 							UpperCAmelCase				(  __lowerCAmelCase  ):




	'''simple docstring'''





	snake_case_    =					VOCAB_FILES_NAMES
	snake_case_    =					QUESTION_ENCODER_PRETRAINED_VOCAB_FILES_MAP
	snake_case_    =					QUESTION_ENCODER_PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	snake_case_    =					QUESTION_ENCODER_PRETRAINED_INIT_CONFIGURATION





SCREAMING_SNAKE_CASE							:List[Any]										=    collections.namedtuple(
    'DPRSpanPrediction', ['span_score', 'relevance_score', 'doc_id', 'start_index', 'end_index', 'text']
)

SCREAMING_SNAKE_CASE							:Dict										=    collections.namedtuple('DPRReaderOutput', ['start_logits', 'end_logits', 'relevance_logits'])


SCREAMING_SNAKE_CASE							:Union[str, Any]										=    R'\n    Return a dictionary with the token ids of the input strings and other information to give to `.decode_best_spans`.\n    It converts the strings of a question and different passages (title and text) in a sequence of IDs (integers),\n    using the tokenizer and vocabulary. The resulting `input_ids` is a matrix of size `(n_passages, sequence_length)`\n    with the format:\n\n    ```\n    [CLS] <question token ids> [SEP] <titles ids> [SEP] <texts ids>\n    ```\n\n    Args:\n        questions (`str` or `List[str]`):\n            The questions to be encoded. You can specify one question for many passages. In this case, the question\n            will be duplicated like `[questions] * n_passages`. Otherwise you have to specify as many questions as in\n            `titles` or `texts`.\n        titles (`str` or `List[str]`):\n            The passages titles to be encoded. This can be a string or a list of strings if there are several passages.\n        texts (`str` or `List[str]`):\n            The passages texts to be encoded. This can be a string or a list of strings if there are several passages.\n        padding (`bool`, `str` or [`~utils.PaddingStrategy`], *optional*, defaults to `False`):\n            Activates and controls padding. Accepts the following values:\n\n            - `True` or `\'longest\'`: Pad to the longest sequence in the batch (or no padding if only a single sequence\n              if provided).\n            - `\'max_length\'`: Pad to a maximum length specified with the argument `max_length` or to the maximum\n              acceptable input length for the model if that argument is not provided.\n            - `False` or `\'do_not_pad\'` (default): No padding (i.e., can output a batch with sequences of different\n              lengths).\n        truncation (`bool`, `str` or [`~tokenization_utils_base.TruncationStrategy`], *optional*, defaults to `False`):\n            Activates and controls truncation. Accepts the following values:\n\n            - `True` or `\'longest_first\'`: Truncate to a maximum length specified with the argument `max_length` or to\n              the maximum acceptable input length for the model if that argument is not provided. This will truncate\n              token by token, removing a token from the longest sequence in the pair if a pair of sequences (or a batch\n              of pairs) is provided.\n            - `\'only_first\'`: Truncate to a maximum length specified with the argument `max_length` or to the maximum\n              acceptable input length for the model if that argument is not provided. This will only truncate the first\n              sequence of a pair if a pair of sequences (or a batch of pairs) is provided.\n            - `\'only_second\'`: Truncate to a maximum length specified with the argument `max_length` or to the maximum\n              acceptable input length for the model if that argument is not provided. This will only truncate the\n              second sequence of a pair if a pair of sequences (or a batch of pairs) is provided.\n            - `False` or `\'do_not_truncate\'` (default): No truncation (i.e., can output batch with sequence lengths\n              greater than the model maximum admissible input size).\n        max_length (`int`, *optional*):\n                Controls the maximum length to use by one of the truncation/padding parameters.\n\n                If left unset or set to `None`, this will use the predefined model maximum length if a maximum length\n                is required by one of the truncation/padding parameters. If the model has no specific maximum input\n                length (like XLNet) truncation/padding to a maximum length will be deactivated.\n        return_tensors (`str` or [`~utils.TensorType`], *optional*):\n                If set, will return tensors instead of list of python integers. Acceptable values are:\n\n                - `\'tf\'`: Return TensorFlow `tf.constant` objects.\n                - `\'pt\'`: Return PyTorch `torch.Tensor` objects.\n                - `\'np\'`: Return Numpy `np.ndarray` objects.\n        return_attention_mask (`bool`, *optional*):\n            Whether or not to return the attention mask. If not set, will return the attention mask according to the\n            specific tokenizer\'s default, defined by the `return_outputs` attribute.\n\n            [What are attention masks?](../glossary#attention-mask)\n\n    Returns:\n        `Dict[str, List[List[int]]]`: A dictionary with the following keys:\n\n        - `input_ids`: List of token ids to be fed to a model.\n        - `attention_mask`: List of indices specifying which tokens should be attended to by the model.\n    '





@add_start_docstrings(__lowerCAmelCase  )
class 							UpperCAmelCase				:




	'''simple docstring'''





	def __call__(	self :			List[Any]							,A :			List[str]							,A :			str = None							,A :			List[str] = None							,A :			int = False							,A :			Any = False							,A :			Dict = None							,A :			List[Any] = None							,A :			Dict = None							,**A :			List[Any]							,):
					if titles is None and texts is None:
									return super().__call__(
									    lowerCAmelCase_							,padding=lowerCAmelCase_							,truncation=lowerCAmelCase_							,max_length=lowerCAmelCase_							,return_tensors=lowerCAmelCase_							,return_attention_mask=lowerCAmelCase_							,**lowerCAmelCase_							,)
					elif titles is None or texts is None:
									__A             =			titles if texts is None else texts
									return super().__call__(
									    lowerCAmelCase_							,lowerCAmelCase_							,padding=lowerCAmelCase_							,truncation=lowerCAmelCase_							,max_length=lowerCAmelCase_							,return_tensors=lowerCAmelCase_							,return_attention_mask=lowerCAmelCase_							,**lowerCAmelCase_							,)
					__A             =			titles if not isinstance(lowerCAmelCase_							,lowerCAmelCase_			) else [titles]
					__A             =			texts if not isinstance(lowerCAmelCase_							,lowerCAmelCase_			) else [texts]
					__A             =			len(lowerCAmelCase_			)
					__A             =			questions if not isinstance(lowerCAmelCase_							,lowerCAmelCase_			) else [questions] * n_passages
					if len(lowerCAmelCase_			) != len(lowerCAmelCase_			):
									raise ValueError(
									    f'''There should be as many titles than texts but got {len(lowerCAmelCase_			)} titles and {len(lowerCAmelCase_			)} texts.'''			)
					__A             =			super().__call__(lowerCAmelCase_							,lowerCAmelCase_							,padding=lowerCAmelCase_							,truncation=lowerCAmelCase_			)["input_ids"]
					__A             =			super().__call__(lowerCAmelCase_							,add_special_tokens=lowerCAmelCase_							,padding=lowerCAmelCase_							,truncation=lowerCAmelCase_			)["input_ids"]
					__A             =			{
					    "input_ids": [
					        (encoded_question_and_title + encoded_text)[:max_length]
					        if max_length is not None and truncation
					        else encoded_question_and_title + encoded_text
					        for encoded_question_and_title, encoded_text in zip(lowerCAmelCase_							,lowerCAmelCase_			)
					    ]
					}
					if return_attention_mask is not False:
									__A             =			[]
									for input_ids in encoded_inputs["input_ids"]:
													attention_mask.append([int(input_id != self.pad_token_id			) for input_id in input_ids]			)
									__A             =			attention_mask
					return self.pad(lowerCAmelCase_							,padding=lowerCAmelCase_							,max_length=lowerCAmelCase_							,return_tensors=lowerCAmelCase_			)






	def     UpperCamelCase_  (	self :			Tuple							,A :			Tuple							,A :			Dict							,A :			int = 16							,A :			Optional[int] = 64							,A :			List[str] = 4							,):
					__A             =			reader_input["input_ids"]
					__A      ,							__A      ,							__A             =			reader_output[:3]
					__A             =			len(lowerCAmelCase_			)
					__A             =			sorted(range(lowerCAmelCase_			)							,reverse=lowerCAmelCase_							,key=relevance_logits.__getitem__			)
					__A             =			[]
					for doc_id in sorted_docs:
									__A             =			list(input_ids[doc_id]			)
									# assuming question & title information is at the beginning of the sequence
									__A             =			sequence_ids.index(self.sep_token_id							,2			) + 1  # second sep id
									if sequence_ids[-1] == self.pad_token_id:
													__A             =			sequence_ids.index(self.pad_token_id			)
									else:
													__A             =			len(lowerCAmelCase_			)

									__A             =			self._get_best_spans(
									    start_logits=start_logits[doc_id][passage_offset:sequence_len]							,end_logits=end_logits[doc_id][passage_offset:sequence_len]							,max_answer_length=lowerCAmelCase_							,top_spans=lowerCAmelCase_							,)
									for start_index, end_index in best_spans:
													start_index += passage_offset
													end_index += passage_offset
													nbest_spans_predictions.append(
													    DPRSpanPrediction(
													        span_score=start_logits[doc_id][start_index] + end_logits[doc_id][end_index]							,relevance_score=relevance_logits[doc_id]							,doc_id=lowerCAmelCase_							,start_index=lowerCAmelCase_							,end_index=lowerCAmelCase_							,text=self.decode(sequence_ids[start_index : end_index + 1]			)							,)			)
									if len(lowerCAmelCase_			) >= num_spans:
													break
					return nbest_spans_predictions[:num_spans]







	def     UpperCamelCase_  (	self :			List[Any]							,A :			int							,A :			Union[str, Any]							,A :			Union[str, Any]							,A :			Optional[int]							,):
					__A             =			[]
					for start_index, start_score in enumerate(lowerCAmelCase_			):
									for answer_length, end_score in enumerate(end_logits[start_index : start_index + max_answer_length]			):
													scores.append(((start_index, start_index + answer_length), start_score + end_score)			)
					__A             =			sorted(lowerCAmelCase_							,key=lambda A			: x[1]							,reverse=lowerCAmelCase_			)
					__A             =			[]
					for (start_index, end_index), score in scores:
									if start_index > end_index:
													raise ValueError(f'''Wrong span indices: [{start_index}:{end_index}]'''			)
									__A             =			end_index - start_index + 1
									if length > max_answer_length:
													raise ValueError(f'''Span is too long: {length} > {max_answer_length}'''			)
									if any(
									    start_index <= prev_start_index <= prev_end_index <= end_index
									    or prev_start_index <= start_index <= end_index <= prev_end_index
									    for (prev_start_index, prev_end_index) in chosen_span_intervals			):
													continue
									chosen_span_intervals.append((start_index, end_index)			)

									if len(lowerCAmelCase_			) == top_spans:
													break
					return chosen_span_intervals





@add_end_docstrings(__lowerCAmelCase  )
class 							UpperCAmelCase				(  __lowerCAmelCase      ,      __lowerCAmelCase  ):




	'''simple docstring'''





	snake_case_    =					VOCAB_FILES_NAMES
	snake_case_    =					READER_PRETRAINED_VOCAB_FILES_MAP
	snake_case_    =					READER_PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	snake_case_    =					READER_PRETRAINED_INIT_CONFIGURATION
	snake_case_    =					['''input_ids''', '''attention_mask''']


from typing import Optional

import numpy as np
import torch
from torch import nn
from transformers import GPTaConfig, GPTaLMHeadModel
from transformers.modeling_utils import ModuleUtilsMixin

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin





class 							UpperCAmelCase				(  __SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE  ):




	'''simple docstring'''





	snake_case_    =					[R"h\.\d+\.attn\.bias", R"h\.\d+\.attn\.masked_bias"]






	@register_to_config
	def __init__(	self :			Tuple							,A :			int							,A :			int							,A :			Optional[int] = None							,A :			int = 5_02_57							,A :			int = 10_24							,A :			int = 7_68							,A :			int = 12							,A :			int = 12							,A :			Optional[int] = None							,A :			str = "gelu_new"							,A :			float = 0.1							,A :			float = 0.1							,A :			float = 0.1							,A :			float = 1E-5							,A :			float = 0.02							,A :			bool = True							,A :			bool = True							,A :			bool = False							,A :			bool = False							,):
					super().__init__()

					__A             =			prefix_length

					if prefix_inner_dim != n_embd and prefix_hidden_dim is None:
									raise ValueError(
									    f'''`prefix_hidden_dim` cannot be `None` when `prefix_inner_dim`: {prefix_hidden_dim} and'''
									    f''' `n_embd`: {n_embd} are not equal.'''			)

					__A             =			prefix_inner_dim
					__A             =			prefix_hidden_dim

					__A             =			(
					    nn.Linear(self.prefix_inner_dim							,self.prefix_hidden_dim			)
					    if self.prefix_hidden_dim is not None
					    else nn.Identity()
					)
					__A             =			(
					    nn.Linear(self.prefix_hidden_dim							,A			) if self.prefix_hidden_dim is not None else nn.Identity()
					)

					__A             =			GPTaConfig(
					    vocab_size=A							,n_positions=A							,n_embd=A							,n_layer=A							,n_head=A							,n_inner=A							,activation_function=A							,resid_pdrop=A							,embd_pdrop=A							,attn_pdrop=A							,layer_norm_epsilon=A							,initializer_range=A							,scale_attn_weights=A							,use_cache=A							,scale_attn_by_inverse_layer_idx=A							,reorder_and_upcast_attn=A							,)
					__A             =			GPTaLMHeadModel(A			)






	def     UpperCamelCase_  (	self :			Union[str, Any]							,A :			torch.Tensor							,A :			torch.Tensor							,A :			Optional[torch.Tensor] = None							,A :			Optional[torch.Tensor] = None							,):
					__A             =			self.transformer.transformer.wte(A			)
					__A             =			self.encode_prefix(A			)
					__A             =			self.decode_prefix(A			)
					__A             =			torch.cat((prefix_embeds, embedding_text)							,dim=1			)

					if labels is not None:
									__A             =			self.get_dummy_token(input_ids.shape[0]							,input_ids.device			)
									__A             =			torch.cat((dummy_token, input_ids)							,dim=1			)
					__A             =			self.transformer(inputs_embeds=A							,labels=A							,attention_mask=A			)
					if self.prefix_hidden_dim is not None:
									return out, hidden
					else:
									return out






	def     UpperCamelCase_  (	self :			int							,A :			int							,A :			torch.device			):
					return torch.zeros(A							,self.prefix_length							,dtype=torch.intaa							,device=A			)






	def     UpperCamelCase_  (	self :			Optional[Any]							,A :			List[Any]			):
					return self.encode_prefix(A			)






	@torch.no_grad()
	def     UpperCamelCase_  (	self :			str							,A :			List[str]							,A :			List[Any]							,A :			List[Any]			):
					__A             =			torch.split(A							,1							,dim=0			)
					__A             =			[]
					__A             =			[]
					for feature in features:
									__A             =			self.decode_prefix(feature.to(A			)			)  # back to the clip feature
									# Only support beam search for now
									__A      ,							__A             =			self.generate_beam(
									    input_embeds=A							,device=A							,eos_token_id=A			)
									generated_tokens.append(output_tokens[0]			)
									generated_seq_lengths.append(seq_lengths[0]			)
					__A             =			torch.stack(A			)
					__A             =			torch.stack(A			)
					return generated_tokens, generated_seq_lengths







	@torch.no_grad()
	def     UpperCamelCase_  (	self :			Tuple							,A :			List[str]=None							,A :			Dict=None							,A :			Optional[int]=None							,A :			int = 5							,A :			int = 67							,A :			float = 1.0							,A :			Optional[int] = None							,):
					__A             =			eos_token_id
					__A             =			None
					__A             =			None
					__A             =			torch.ones(A							,device=A							,dtype=torch.int			)
					__A             =			torch.zeros(A							,device=A							,dtype=torch.bool			)

					if input_embeds is not None:
									__A             =			input_embeds
					else:
									__A             =			self.transformer.transformer.wte(A			)

					for i in range(A			):
									__A             =			self.transformer(inputs_embeds=A			)
									__A             =			outputs.logits
									__A             =			logits[:, -1, :] / (temperature if temperature > 0 else 1.0)
									__A             =			logits.softmax(-1			).log()

									if scores is None:
													__A      ,							__A             =			logits.topk(A							,-1			)
													__A             =			generated.expand(A							,*generated.shape[1:]			)
													__A      ,							__A             =			next_tokens.permute(1							,0			), scores.squeeze(0			)
													if tokens is None:
																	__A             =			next_tokens
													else:
																	__A             =			tokens.expand(A							,*tokens.shape[1:]			)
																	__A             =			torch.cat((tokens, next_tokens)							,dim=1			)
									else:
													__A             =			-float(np.inf			)
													__A             =			0
													__A             =			scores[:, None] + logits
													seq_lengths[~is_stopped] += 1
													__A             =			scores_sum / seq_lengths[:, None]
													__A      ,							__A             =			scores_sum_average.view(-1			).topk(A							,-1			)
													__A             =			next_tokens // scores_sum.shape[1]
													__A             =			seq_lengths[next_tokens_source]
													__A             =			next_tokens % scores_sum.shape[1]
													__A             =			next_tokens.unsqueeze(1			)
													__A             =			tokens[next_tokens_source]
													__A             =			torch.cat((tokens, next_tokens)							,dim=1			)
													__A             =			generated[next_tokens_source]
													__A             =			scores_sum_average * seq_lengths
													__A             =			is_stopped[next_tokens_source]

									__A             =			self.transformer.transformer.wte(next_tokens.squeeze()			).view(generated.shape[0]							,1							,-1			)
									__A             =			torch.cat((generated, next_token_embed)							,dim=1			)
									__A             =			is_stopped + next_tokens.eq(A			).squeeze()
									if is_stopped.all():
													break

					__A             =			scores / seq_lengths
					__A             =			scores.argsort(descending=A			)
					# tokens tensors are already padded to max_seq_length
					__A             =			[tokens[i] for i in order]
					__A             =			torch.stack(A							,dim=0			)
					__A             =			torch.tensor([seq_lengths[i] for i in order]							,dtype=seq_lengths.dtype			)
					return output_texts, seq_lengths
