'''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


lowercase_  		=			'''block_records.npy'''


lowercase_  		=			logging.get_logger(__name__)



def 				UpperCamelCase__      (      a__      ,   a__   ):


						'''simple docstring'''






						import tensorflow.compat.va as tf

						_lowerCAmelCase					    =tf.data.TFRecordDataset(a__      ,   buffer_size=5_1_2 * 1_0_2_4 * 1_0_2_4   )
						_lowerCAmelCase					    =blocks_dataset.batch(a__      ,   drop_remainder=a__   )
						_lowerCAmelCase					    =next(blocks_dataset.take(1   ).as_numpy_iterator()   )

						return np_record
class 	SCREAMING_SNAKE_CASE      :







		"""simple docstring"""






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

								_lowerCAmelCase					    =Builder(db=__A		,    num_neighbors=__A		,    distance_measure='dot_product'					)
								_lowerCAmelCase					    =builder.tree(
								    num_leaves=__A		,    num_leaves_to_search=__A		,    training_sample_size=__A					)
								_lowerCAmelCase					    =builder.score_ah(dimensions_per_block=__A					)

								_lowerCAmelCase					    =builder.build()






		def 		UpperCamelCase__							( self		,    __A					)			->  Optional[Any]:
								_lowerCAmelCase    ,					_lowerCAmelCase					    =self.searcher.search_batched(question_projection.detach().cpu()					)
								return retrieved_block_ids.astype('int64'					)

class 	SCREAMING_SNAKE_CASE      :







		"""simple docstring"""






		def __init__( self		,    __A		,    __A					)			->  Optional[Any]:
								super().__init__()
								_lowerCAmelCase					    =block_records
								_lowerCAmelCase					    =tokenizer






		def __call__( self		,    __A		,    __A		,    __A		,    __A=None		,    __A="pt"					)			->  Union[str, Any]:
								_lowerCAmelCase					    =np.take(self.block_records		,    indices=__A		,    axis=0					)

								_lowerCAmelCase					    =self.tokenizer.decode(question_input_ids[0]		,    skip_special_tokens=__A					)

								_lowerCAmelCase					    =[]
								_lowerCAmelCase					    =[]
								for retrieved_block in retrieved_blocks:
														text.append(__A					)
														text_pair.append(retrieved_block.decode()					)

								_lowerCAmelCase					    =self.tokenizer(
								    __A		,    __A		,    padding=__A		,    truncation=__A		,    return_special_tokens_mask=__A		,    max_length=__A					)
								_lowerCAmelCase					    =concat_inputs.convert_to_tensors(__A					)

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






		@classmethod
		def 		UpperCamelCase__							( cls		,    __A		,    *__A		,    **__A					)			->  Tuple:
								if os.path.isdir(__A					):
														_lowerCAmelCase					    =os.path.join(__A		,    _REALM_BLOCK_RECORDS_FILENAME					)
								else:
														_lowerCAmelCase					    =hf_hub_download(
														    repo_id=__A		,    filename=_REALM_BLOCK_RECORDS_FILENAME		,    **__A					)
								_lowerCAmelCase					    =np.load(__A		,    allow_pickle=__A					)

								_lowerCAmelCase					    =AutoTokenizer.from_pretrained(__A		,    *__A		,    **__A					)

								return cls(__A		,    __A					)






		def 		UpperCamelCase__							( self		,    __A					)			->  str:
								# save block records
								np.save(os.path.join(__A		,    _REALM_BLOCK_RECORDS_FILENAME					)		,    self.block_records					)
								# save tokenizer
								self.tokenizer.save_pretrained(__A					)






		def 		UpperCamelCase__							( self		,    __A		,    __A					)			->  Optional[Any]:
								_lowerCAmelCase					    =[]
								_lowerCAmelCase					    =[]
								_lowerCAmelCase					    =[]
								_lowerCAmelCase					    =0

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

														if len(start_pos[-1]					) == 0:
																				has_answers.append(__A					)
														else:
																				has_answers.append(__A					)
																				if len(start_pos[-1]					) > max_answers:
																										_lowerCAmelCase					    =len(start_pos[-1]					)

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






'''simple docstring'''

from ..utils import DummyObject, requires_backends
class 	SCREAMING_SNAKE_CASE      (    metaclass=__lowercase):







		"""simple docstring"""
		lowercase :   List[Any]							=				['tensorflow_text']






		def __init__( self		,    *__A		,    **__A					)			->  List[Any]:
								requires_backends(self		,    ['tensorflow_text']					)





