
"""simple docstring"""

import os
from typing import Optional, Union

import numpy as np
from huggingface_hub import hf_hub_download

from ... import AutoTokenizer
from ...utils import logging


_A            = 'block_records.npy'


_A            = logging.get_logger(__name__)
def 		UpperCAmelCase					(		a_,						a_			):





		'''simple docstring'''


		import tensorflow.compat.va as tf

		lowerCamelCase     :			Dict		=    tf.data.TFRecordDataset(a_,						buffer_size=512 * 1024 * 1024			)
		lowerCamelCase     :			List[str]		=    blocks_dataset.batch(a_,						drop_remainder=a_			)
		lowerCamelCase     :			List[Any]		=    next(blocks_dataset.take(1			).as_numpy_iterator()			)

		return np_record






class        _lowercase   :




	def __init__( self		, UpperCAmelCase_		, UpperCAmelCase_		, UpperCAmelCase_=2		, UpperCAmelCase_=1000		, UpperCAmelCase_=100		, UpperCAmelCase_=100000		, )		->							int:
			from scann.scann_ops.py.scann_ops_pybind import builder as Builder

			lowerCamelCase     :			List[Any]		=    Builder(db=UpperCAmelCase_		, num_neighbors=UpperCAmelCase_		, distance_measure='dot_product'					)
			lowerCamelCase     :			str		=    builder.tree(
			    num_leaves=UpperCAmelCase_		, num_leaves_to_search=UpperCAmelCase_		, training_sample_size=UpperCAmelCase_					)
			lowerCamelCase     :			Union[str, Any]		=    builder.score_ah(dimensions_per_block=UpperCAmelCase_					)

			lowerCamelCase     :			str		=    builder.build()




	def 			_UpperCamelCase    ( self		, UpperCAmelCase_					)		->							int:
			lowerCamelCase	,    lowerCamelCase     :			Optional[int]		=    self.searcher.search_batched(question_projection.detach().cpu()					)
			return retrieved_block_ids.astype('int64'					)






class        _lowercase   :




	def __init__( self		, UpperCAmelCase_		, UpperCAmelCase_					)		->							List[Any]:
			super().__init__()
			lowerCamelCase     :			str		=    block_records
			lowerCamelCase     :			List[str]		=    tokenizer




	def __call__( self		, UpperCAmelCase_		, UpperCAmelCase_		, UpperCAmelCase_		, UpperCAmelCase_=None		, UpperCAmelCase_="pt"					)		->							Optional[Any]:
			lowerCamelCase     :			Optional[int]		=    np.take(self.block_records		, indices=UpperCAmelCase_		, axis=0					)

			lowerCamelCase     :			Any		=    self.tokenizer.decode(question_input_ids[0]		, skip_special_tokens=UpperCAmelCase_					)

			lowerCamelCase     :			int		=    []
			lowerCamelCase     :			Tuple		=    []
			for retrieved_block in retrieved_blocks:
					text.append(UpperCAmelCase_					)
					text_pair.append(retrieved_block.decode()					)

			lowerCamelCase     :			Dict		=    self.tokenizer(
			    UpperCAmelCase_		, UpperCAmelCase_		, padding=UpperCAmelCase_		, truncation=UpperCAmelCase_		, return_special_tokens_mask=UpperCAmelCase_		, max_length=UpperCAmelCase_					)
			lowerCamelCase     :			Union[str, Any]		=    concat_inputs.convert_to_tensors(UpperCAmelCase_					)

			if answer_ids is not None:
					return self.block_has_answer(UpperCAmelCase_		, UpperCAmelCase_					) + (concat_inputs_tensors,)
			else:
					return (None, None, None, concat_inputs_tensors)




	@classmethod
	def 			_UpperCamelCase    ( cls		, UpperCAmelCase_		, *UpperCAmelCase_		, **UpperCAmelCase_					)		->							Tuple:
			if os.path.isdir(UpperCAmelCase_					):
					lowerCamelCase     :			Optional[Any]		=    os.path.join(UpperCAmelCase_		, _REALM_BLOCK_RECORDS_FILENAME					)
			else:
					lowerCamelCase     :			int		=    hf_hub_download(
					    repo_id=UpperCAmelCase_		, filename=_REALM_BLOCK_RECORDS_FILENAME		, **UpperCAmelCase_					)
			lowerCamelCase     :			Optional[int]		=    np.load(UpperCAmelCase_		, allow_pickle=UpperCAmelCase_					)

			lowerCamelCase     :			int		=    AutoTokenizer.from_pretrained(UpperCAmelCase_		, *UpperCAmelCase_		, **UpperCAmelCase_					)

			return cls(UpperCAmelCase_		, UpperCAmelCase_					)




	def 			_UpperCamelCase    ( self		, UpperCAmelCase_					)		->							Optional[int]:
			# save block records
			np.save(os.path.join(UpperCAmelCase_		, _REALM_BLOCK_RECORDS_FILENAME					)		, self.block_records					)
			# save tokenizer
			self.tokenizer.save_pretrained(UpperCAmelCase_					)




	def 			_UpperCamelCase    ( self		, UpperCAmelCase_		, UpperCAmelCase_					)		->							Optional[int]:
			lowerCamelCase     :			Optional[Any]		=    []
			lowerCamelCase     :			Dict		=    []
			lowerCamelCase     :			str		=    []
			lowerCamelCase     :			List[Any]		=    0

			for input_id in concat_inputs.input_ids:
					lowerCamelCase     :			Union[str, Any]		=    input_id.tolist()
					# Check answers between two [SEP] tokens
					lowerCamelCase     :			List[Any]		=    input_id_list.index(self.tokenizer.sep_token_id					)
					lowerCamelCase     :			Tuple		=    first_sep_idx + 1 + input_id_list[first_sep_idx + 1 :].index(self.tokenizer.sep_token_id					)

					start_pos.append([]					)
					end_pos.append([]					)
					for answer in answer_ids:
							for idx in range(first_sep_idx + 1		, UpperCAmelCase_					):
									if answer[0] == input_id_list[idx]:
											if input_id_list[idx : idx + len(UpperCAmelCase_					)] == answer:
													start_pos[-1].append(UpperCAmelCase_					)
													end_pos[-1].append(idx + len(UpperCAmelCase_					) - 1					)

					if len(start_pos[-1]					) == 0:
							has_answers.append(UpperCAmelCase_					)
					else:
							has_answers.append(UpperCAmelCase_					)
							if len(start_pos[-1]					) > max_answers:
									lowerCamelCase     :			List[Any]		=    len(start_pos[-1]					)

        # Pad -1 to max_answers
			for start_pos_, end_pos_ in zip(UpperCAmelCase_		, UpperCAmelCase_					):
					if len(start_pos_					) < max_answers:
							lowerCamelCase     :			str		=    [-1] * (max_answers - len(start_pos_					))
							start_pos_ += padded
							end_pos_ += padded
			return has_answers, start_pos, end_pos








"""simple docstring"""

from transformers import (
    DPRContextEncoderTokenizer,
    DPRContextEncoderTokenizerFast,
    DPRQuestionEncoderTokenizer,
    DPRQuestionEncoderTokenizerFast,
    DPRReaderOutput,
    DPRReaderTokenizer,
    DPRReaderTokenizerFast,
)
from transformers.testing_utils import require_tokenizers, slow
from transformers.tokenization_utils_base import BatchEncoding

from ..bert.test_tokenization_bert import BertTokenizationTest






@require_tokenizers
class        _lowercase   (				__UpperCAmelCase		):
	lowercase_        =	DPRContextEncoderTokenizer
	lowercase_        =	DPRContextEncoderTokenizerFast
	lowercase_        =	True






@require_tokenizers
class        _lowercase   (				__UpperCAmelCase		):
	lowercase_        =	DPRQuestionEncoderTokenizer
	lowercase_        =	DPRQuestionEncoderTokenizerFast
	lowercase_        =	True






@require_tokenizers
class        _lowercase   (				__UpperCAmelCase		):
	lowercase_        =	DPRReaderTokenizer
	lowercase_        =	DPRReaderTokenizerFast
	lowercase_        =	True




	@slow
	def 			_UpperCamelCase    ( self					)		->							Any:
			lowerCamelCase     :			str		=    self.tokenizer_class.from_pretrained('bert-base-uncased'					)

			lowerCamelCase     :			int		=    tokenizer.encode('question sequence'		, add_special_tokens=UpperCAmelCase_					)
			lowerCamelCase     :			List[Any]		=    tokenizer.encode('title sequence'		, add_special_tokens=UpperCAmelCase_					)
			lowerCamelCase     :			Optional[int]		=    tokenizer.encode('text sequence ' * 4		, add_special_tokens=UpperCAmelCase_					)
			lowerCamelCase     :			Union[str, Any]		=    [[101] + text_a + [102] + text_a + [102] + text_a]
			lowerCamelCase     :			List[str]		=    BatchEncoding({'input_ids': input_ids}					)

			lowerCamelCase     :			Optional[int]		=    [[0] * len(input_ids[0]					)]
			lowerCamelCase     :			str		=    [[0] * len(input_ids[0]					)]
			lowerCamelCase     :			List[Any]		=    [0]
			lowerCamelCase     :			Tuple		=    DPRReaderOutput(UpperCAmelCase_		, UpperCAmelCase_		, UpperCAmelCase_					)

			lowerCamelCase	,    lowerCamelCase     :			Optional[int]		=    8, 9
			lowerCamelCase     :			Optional[int]		=    10
			lowerCamelCase     :			int		=    10
			lowerCamelCase     :			List[Any]		=    tokenizer.decode_best_spans(UpperCAmelCase_		, UpperCAmelCase_					)
			self.assertEqual(predicted_spans[0].start_index		, UpperCAmelCase_					)
			self.assertEqual(predicted_spans[0].end_index		, UpperCAmelCase_					)
			self.assertEqual(predicted_spans[0].doc_id		, 0					)




	@slow
	def 			_UpperCamelCase    ( self					)		->							List[Any]:
			lowerCamelCase     :			Union[str, Any]		=    self.tokenizer_class.from_pretrained('bert-base-uncased'					)

			lowerCamelCase     :			List[str]		=    tokenizer.encode('question sequence'		, add_special_tokens=UpperCAmelCase_					)
			lowerCamelCase     :			List[str]		=    tokenizer.encode('title sequence'		, add_special_tokens=UpperCAmelCase_					)
			lowerCamelCase     :			List[Any]		=    tokenizer.encode('text sequence'		, add_special_tokens=UpperCAmelCase_					)
			lowerCamelCase     :			List[Any]		=    [101] + text_a + [102] + text_a + [102] + text_a
			lowerCamelCase     :			Optional[Any]		=    tokenizer(questions=['question sequence']		, titles=['title sequence']		, texts=['text sequence']					)
			self.assertIn('input_ids'		, UpperCAmelCase_					)
			self.assertIn('attention_mask'		, UpperCAmelCase_					)
			self.assertListEqual(encoded_input['input_ids'][0]		, UpperCAmelCase_					)






