




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


lowercase_		=							'''block_records.npy'''


lowercase_		=							logging.get_logger(__name__)







def 			lowerCAmelCase  (     UpperCAmelCase,     UpperCAmelCase	)  ->np.ndarray:






					"""simple docstring"""






					import tensorflow.compat.va as tf

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

					return np_record





class 		A__    :






	def __init__(     self    ,					lowerCamelCase    ,					lowerCamelCase    ,					lowerCamelCase=2    ,					lowerCamelCase=1000    ,					lowerCamelCase=100    ,					lowerCamelCase=100000    ,					)				->       List[Any]:




						"""simple docstring"""






						from scann.scann_ops.py.scann_ops_pybind import builder as Builder

						__magic_name__ :							str		       =							Builder(db=lowerCamelCase    ,					num_neighbors=lowerCamelCase    ,					distance_measure='''dot_product'''						)
						__magic_name__ :							Any		       =							builder.tree(
						    num_leaves=lowerCamelCase    ,					num_leaves_to_search=lowerCamelCase    ,					training_sample_size=lowerCamelCase						)
						__magic_name__ :							Any		       =							builder.score_ah(dimensions_per_block=lowerCamelCase						)

						__magic_name__ :							Tuple		       =							builder.build()






	def 			lowercase				(     self    ,					lowerCamelCase						)				->       List[str]:




						"""simple docstring"""






						__magic_name__       ,			__magic_name__ :							Optional[Any]		       =							self.searcher.search_batched(question_projection.detach().cpu()						)
						return retrieved_block_ids.astype('''int64'''						)





class 		A__    :






	def __init__(     self    ,					lowerCamelCase    ,					lowerCamelCase						)				->       Tuple:




						"""simple docstring"""






						super().__init__()
						__magic_name__ :							Tuple		       =							block_records
						__magic_name__ :							Union[str, Any]		       =							tokenizer






	def __call__(     self    ,					lowerCamelCase    ,					lowerCamelCase    ,					lowerCamelCase    ,					lowerCamelCase=None    ,					lowerCamelCase="pt"						)				->       Any:




						"""simple docstring"""






						__magic_name__ :							str		       =							np.take(self.block_records    ,					indices=lowerCamelCase    ,					axis=0						)

						__magic_name__ :							List[str]		       =							self.tokenizer.decode(question_input_ids[0]    ,					skip_special_tokens=lowerCamelCase						)

						__magic_name__ :							List[str]		       =							[]
						__magic_name__ :							Optional[Any]		       =							[]
						for retrieved_block in retrieved_blocks:
											text.append(lowerCamelCase						)
											text_pair.append(retrieved_block.decode()						)

						__magic_name__ :							int		       =							self.tokenizer(
						    lowerCamelCase    ,					lowerCamelCase    ,					padding=lowerCamelCase    ,					truncation=lowerCamelCase    ,					return_special_tokens_mask=lowerCamelCase    ,					max_length=lowerCamelCase						)
						__magic_name__ :							Optional[Any]		       =							concat_inputs.convert_to_tensors(lowerCamelCase						)

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






	@classmethod
	def 			lowercase				(     cls    ,					lowerCamelCase    ,					*lowerCamelCase    ,					**lowerCamelCase						)				->       Any:




						"""simple docstring"""






						if os.path.isdir(lowerCamelCase						):
											__magic_name__ :							List[Any]		       =							os.path.join(lowerCamelCase    ,					_REALM_BLOCK_RECORDS_FILENAME						)
						else:
											__magic_name__ :							List[Any]		       =							hf_hub_download(
											    repo_id=lowerCamelCase    ,					filename=_REALM_BLOCK_RECORDS_FILENAME    ,					**lowerCamelCase						)
						__magic_name__ :							List[str]		       =							np.load(lowerCamelCase    ,					allow_pickle=lowerCamelCase						)

						__magic_name__ :							Tuple		       =							AutoTokenizer.from_pretrained(lowerCamelCase    ,					*lowerCamelCase    ,					**lowerCamelCase						)

						return cls(lowerCamelCase    ,					lowerCamelCase						)






	def 			lowercase				(     self    ,					lowerCamelCase						)				->       Dict:




						"""simple docstring"""






						np.save(os.path.join(lowerCamelCase    ,					_REALM_BLOCK_RECORDS_FILENAME						)    ,					self.block_records						)
						# save tokenizer
						self.tokenizer.save_pretrained(lowerCamelCase						)






	def 			lowercase				(     self    ,					lowerCamelCase    ,					lowerCamelCase						)				->       Dict:




						"""simple docstring"""






						__magic_name__ :							int		       =							[]
						__magic_name__ :							List[str]		       =							[]
						__magic_name__ :							Optional[int]		       =							[]
						__magic_name__ :							Dict		       =							0

						for input_id in concat_inputs.input_ids:
											__magic_name__ :							Any		       =							input_id.tolist()
											# Check answers between two [SEP] tokens
											__magic_name__ :							List[Any]		       =							input_id_list.index(self.tokenizer.sep_token_id						)
											__magic_name__ :							int		       =							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    ,					lowerCamelCase						):
																					if answer[0] == input_id_list[idx]:
																										if input_id_list[idx : idx + len(lowerCamelCase						)] == answer:
																															start_pos[-1].append(lowerCamelCase						)
																															end_pos[-1].append(idx + len(lowerCamelCase						) - 1						)

											if len(start_pos[-1]						) == 0:
																has_answers.append(lowerCamelCase						)
											else:
																has_answers.append(lowerCamelCase						)
																if len(start_pos[-1]						) > max_answers:
																					__magic_name__ :							Tuple		       =							len(start_pos[-1]						)

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









from ..utils import requires_backends


lowercase_		=							None





class 		A__    :






	def __init__(     self    ,					*lowerCamelCase    ,					**lowerCamelCase						)				->       Optional[Any]:




						"""simple docstring"""






						requires_backends(self    ,					['''detectron2''']						)






	@classmethod
	def 			lowercase				(     cls    ,					*lowerCamelCase    ,					**lowerCamelCase						)				->       List[Any]:




						"""simple docstring"""






						requires_backends(cls    ,					['''detectron2''']						)



