



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


snake_case_								=			"""block_records.npy"""


snake_case_								=			logging.get_logger(__name__)

def 					_lowerCAmelCase (  lowercase_						,      lowercase_		):
						import tensorflow.compat.va as tf

						UpperCAmelCase			 =					tf.data.TFRecordDataset(lowercase_						,      buffer_size=512 * 1024 * 1024		)
						UpperCAmelCase			 =					blocks_dataset.batch(lowercase_						,      drop_remainder=lowercase_		)
						UpperCAmelCase			 =					next(blocks_dataset.take(1		).as_numpy_iterator()		)

						return np_record







class  A_     :



						"""simple docstring"""






						def __init__(   self					:Optional[Any]  ,    lowercase_					:List[Any]  ,    lowercase_					:Dict  ,    lowercase_					:List[str]=2  ,    lowercase_					:Optional[int]=10_00  ,    lowercase_					:List[str]=1_00  ,    lowercase_					:Tuple=10_00_00  ,    )							->     Any:
												from scann.scann_ops.py.scann_ops_pybind import builder as Builder

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

												UpperCAmelCase			 =					builder.build()






						def       UpperCAmelCase__	(   self					:Dict  ,    lowercase_					:Union[str, Any]   )							->     Dict:
												UpperCAmelCase ,       UpperCAmelCase			 =					self.searcher.search_batched(question_projection.detach().cpu()   )
												return retrieved_block_ids.astype('int64'   )







class  A_     :



						"""simple docstring"""






						def __init__(   self					:Optional[int]  ,    lowercase_					:Optional[Any]  ,    lowercase_					:int   )							->     List[Any]:
												super().__init__()
												UpperCAmelCase			 =					block_records
												UpperCAmelCase			 =					tokenizer






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

												UpperCAmelCase			 =					self.tokenizer.decode(question_input_ids[0]  ,    skip_special_tokens=lowercase_   )

												UpperCAmelCase			 =					[]
												UpperCAmelCase			 =					[]
												for retrieved_block in retrieved_blocks:
																		text.append(lowercase_   )
																		text_pair.append(retrieved_block.decode()   )

												UpperCAmelCase			 =					self.tokenizer(
												    lowercase_  ,    lowercase_  ,    padding=lowercase_  ,    truncation=lowercase_  ,    return_special_tokens_mask=lowercase_  ,    max_length=lowercase_   )
												UpperCAmelCase			 =					concat_inputs.convert_to_tensors(lowercase_   )

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






						@classmethod
						def       UpperCAmelCase__	(   cls					:Any  ,    lowercase_					:Optional[Union[str, os.PathLike]]  ,    *lowercase_					:Optional[int]  ,    **lowercase_					:Optional[Any]   )							->     int:
												if os.path.isdir(lowercase_   ):
																		UpperCAmelCase			 =					os.path.join(lowercase_  ,    _REALM_BLOCK_RECORDS_FILENAME   )
												else:
																		UpperCAmelCase			 =					hf_hub_download(
																		    repo_id=lowercase_  ,    filename=_REALM_BLOCK_RECORDS_FILENAME  ,    **lowercase_   )
												UpperCAmelCase			 =					np.load(lowercase_  ,    allow_pickle=lowercase_   )

												UpperCAmelCase			 =					AutoTokenizer.from_pretrained(lowercase_  ,    *lowercase_  ,    **lowercase_   )

												return cls(lowercase_  ,    lowercase_   )






						def       UpperCAmelCase__	(   self					:Any  ,    lowercase_					:Tuple   )							->     Tuple:
												# save block records
												np.save(os.path.join(lowercase_  ,    _REALM_BLOCK_RECORDS_FILENAME   )  ,    self.block_records   )
												# save tokenizer
												self.tokenizer.save_pretrained(lowercase_   )






						def       UpperCAmelCase__	(   self					:Union[str, Any]  ,    lowercase_					:List[str]  ,    lowercase_					:Optional[Any]   )							->     Union[str, Any]:
												UpperCAmelCase			 =					[]
												UpperCAmelCase			 =					[]
												UpperCAmelCase			 =					[]
												UpperCAmelCase			 =					0

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

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

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












'''simple docstring'''


from __future__ import annotations




def       A							(      _UpperCAmelCase   :				list					)  ->				int | float:


  '''simple docstring'''




  __lowerCAmelCase		:    List[Any]			     =  sorted(_UpperCAmelCase					)
  __lowerCAmelCase		:    List[str]			     =  len(_UpperCAmelCase					)
  __lowerCAmelCase		:    Any			     =  length >> 1
  return (
      (sorted_list[mid_index] + sorted_list[mid_index - 1]) / 2
      if length % 2 == 0
      else sorted_list[mid_index]
  )




def       A							(      )  ->				Union[str, Any]:


  '''simple docstring'''




  import doctest

  doctest.testmod()


if __name__ == "__main__":
    main()





