


'''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__:   int    =	'''block_records.npy'''


A__:   Optional[int]    =	logging.get_logger(__name__)


def      SCREAMING_SNAKE_CASE_ (  _UpperCAmelCase     :     str   ,_UpperCAmelCase     :     int  )  ->      np.ndarray:
       import tensorflow.compat.va as tf

       _a   :			Union[str, Any]					 =tf.data.TFRecordDataset(_UpperCAmelCase   ,buffer_size=512 * 1024 * 1024  )
       _a   :			List[Any]					 =blocks_dataset.batch(_UpperCAmelCase   ,drop_remainder=_UpperCAmelCase  )
       _a   :			Optional[int]					 =next(blocks_dataset.take(1  ).as_numpy_iterator()  )

       return np_record







class A__	:






      def __init__(			self      :List[Any]					,						SCREAMING_SNAKE_CASE      :Optional[Any]					,						SCREAMING_SNAKE_CASE      :Union[str, Any]					,						SCREAMING_SNAKE_CASE      :Optional[int]=2					,						SCREAMING_SNAKE_CASE      :int=1_0_0_0					,						SCREAMING_SNAKE_CASE      :Optional[int]=1_0_0					,						SCREAMING_SNAKE_CASE      :Optional[int]=1_0_0_0_0_0					,						)	-> List[Any]:




             '''simple docstring'''
             from scann.scann_ops.py.scann_ops_pybind import builder as Builder

             _a   :			List[Any]					 =Builder(db=SCREAMING_SNAKE_CASE					,						num_neighbors=SCREAMING_SNAKE_CASE					,						distance_measure="""dot_product"""     )
             _a   :			str					 =builder.tree(
                 num_leaves=SCREAMING_SNAKE_CASE					,						num_leaves_to_search=SCREAMING_SNAKE_CASE					,						training_sample_size=SCREAMING_SNAKE_CASE     )
             _a   :			int					 =builder.score_ah(dimensions_per_block=SCREAMING_SNAKE_CASE     )

             _a   :			str					 =builder.build()






      def __UpperCAmelCase	(			self      :Optional[int]					,						SCREAMING_SNAKE_CASE      :List[str]     )	-> int:




             '''simple docstring'''
             _a   :			Union[str, Any]					 =self.searcher.search_batched(question_projection.detach().cpu()     )
             return retrieved_block_ids.astype("""int64"""     )







class A__	:






      def __init__(			self      :Any					,						SCREAMING_SNAKE_CASE      :Dict					,						SCREAMING_SNAKE_CASE      :int     )	-> Tuple:




             '''simple docstring'''
             super().__init__()
             _a   :			Union[str, Any]					 =block_records
             _a   :			int					 =tokenizer






      def __call__(			self      :Union[str, Any]					,						SCREAMING_SNAKE_CASE      :List[str]					,						SCREAMING_SNAKE_CASE      :Dict					,						SCREAMING_SNAKE_CASE      :Optional[int]					,						SCREAMING_SNAKE_CASE      :Optional[int]=None					,						SCREAMING_SNAKE_CASE      :int="pt"     )	-> Optional[int]:




             '''simple docstring'''
             _a   :			Any					 =np.take(self.block_records					,						indices=SCREAMING_SNAKE_CASE					,						axis=0     )

             _a   :			str					 =self.tokenizer.decode(question_input_ids[0]					,						skip_special_tokens=SCREAMING_SNAKE_CASE     )

             _a   :			int					 =[]
             _a   :			str					 =[]
             for retrieved_block in retrieved_blocks:
                    text.append(SCREAMING_SNAKE_CASE     )
                    text_pair.append(retrieved_block.decode()     )

             _a   :			List[str]					 =self.tokenizer(
                 SCREAMING_SNAKE_CASE					,						SCREAMING_SNAKE_CASE					,						padding=SCREAMING_SNAKE_CASE					,						truncation=SCREAMING_SNAKE_CASE					,						return_special_tokens_mask=SCREAMING_SNAKE_CASE					,						max_length=SCREAMING_SNAKE_CASE     )
             _a   :			Optional[int]					 =concat_inputs.convert_to_tensors(SCREAMING_SNAKE_CASE     )

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






      @classmethod
      def __UpperCAmelCase	(			cls      :List[Any]					,						SCREAMING_SNAKE_CASE      :Optional[Union[str, os.PathLike]]					,						*SCREAMING_SNAKE_CASE      :str					,						**SCREAMING_SNAKE_CASE      :Optional[Any]     )	-> Dict:




             '''simple docstring'''
             if os.path.isdir(SCREAMING_SNAKE_CASE     ):
                    _a   :			Any					 =os.path.join(SCREAMING_SNAKE_CASE					,						_REALM_BLOCK_RECORDS_FILENAME     )
             else:
                    _a   :			List[str]					 =hf_hub_download(
                        repo_id=SCREAMING_SNAKE_CASE					,						filename=_REALM_BLOCK_RECORDS_FILENAME					,						**SCREAMING_SNAKE_CASE     )
             _a   :			Optional[Any]					 =np.load(SCREAMING_SNAKE_CASE					,						allow_pickle=SCREAMING_SNAKE_CASE     )

             _a   :			Union[str, Any]					 =AutoTokenizer.from_pretrained(SCREAMING_SNAKE_CASE					,						*SCREAMING_SNAKE_CASE					,						**SCREAMING_SNAKE_CASE     )

             return cls(SCREAMING_SNAKE_CASE					,						SCREAMING_SNAKE_CASE     )






      def __UpperCAmelCase	(			self      :Union[str, Any]					,						SCREAMING_SNAKE_CASE      :Optional[int]     )	-> Tuple:




             '''simple docstring'''
             np.save(os.path.join(SCREAMING_SNAKE_CASE					,						_REALM_BLOCK_RECORDS_FILENAME     )					,						self.block_records     )
             # save tokenizer
             self.tokenizer.save_pretrained(SCREAMING_SNAKE_CASE     )






      def __UpperCAmelCase	(			self      :str					,						SCREAMING_SNAKE_CASE      :str					,						SCREAMING_SNAKE_CASE      :Dict     )	-> Any:




             '''simple docstring'''
             _a   :			Union[str, Any]					 =[]
             _a   :			str					 =[]
             _a   :			List[str]					 =[]
             _a   :			Any					 =0

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

                    if len(start_pos[-1]     ) == 0:
                           has_answers.append(SCREAMING_SNAKE_CASE     )
                    else:
                           has_answers.append(SCREAMING_SNAKE_CASE     )
                           if len(start_pos[-1]     ) > max_answers:
                                  _a   :			Optional[int]					 =len(start_pos[-1]     )

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






'''simple docstring'''







from typing import Dict, Optional, Tuple

import torch


def      SCREAMING_SNAKE_CASE_ (  _UpperCAmelCase     :     torch.Tensor  )  ->      torch.Tensor:
       _a   :			Tuple					 =boundaries[1] - boundaries[0]
       _a   :			Dict					 =boundaries + step / 2
       _a   :			List[Any]					 =torch.cat([bin_centers, (bin_centers[-1] + step).unsqueeze(-1  )]   ,dim=0  )
       return bin_centers


def      SCREAMING_SNAKE_CASE_ (  _UpperCAmelCase     :     torch.Tensor   ,_UpperCAmelCase     :     torch.Tensor   ,)  ->      Tuple[torch.Tensor, torch.Tensor]:
       _a   :			Any					 =_calculate_bin_centers(_UpperCAmelCase  )
       return (
           torch.sum(aligned_distance_error_probs * bin_centers   ,dim=-1  ),
           bin_centers[-1],
       )


def      SCREAMING_SNAKE_CASE_ (  _UpperCAmelCase     :     torch.Tensor   ,_UpperCAmelCase     :     int = 31   ,_UpperCAmelCase     :     int = 64   ,**_UpperCAmelCase     :     Dict   ,)  ->      Dict[str, torch.Tensor]:
       _a   :			Dict					 =torch.linspace(0   ,_UpperCAmelCase   ,steps=(no_bins - 1)   ,device=logits.device  )

       _a   :			int					 =torch.nn.functional.softmax(_UpperCAmelCase   ,dim=-1  )
       _a ,   _a   :			List[Any]					 =_calculate_expected_aligned_error(
           alignment_confidence_breaks=_UpperCAmelCase   ,aligned_distance_error_probs=_UpperCAmelCase   ,)

       return {
           "aligned_confidence_probs": aligned_confidence_probs,
           "predicted_aligned_error": predicted_aligned_error,
           "max_predicted_aligned_error": max_predicted_aligned_error,
       }


def      SCREAMING_SNAKE_CASE_ (  _UpperCAmelCase     :     torch.Tensor   ,_UpperCAmelCase     :     Optional[torch.Tensor] = None   ,_UpperCAmelCase     :     int = 31   ,_UpperCAmelCase     :     int = 64   ,_UpperCAmelCase     :     float = 1e-8   ,**_UpperCAmelCase     :     Optional[Any]   ,)  ->      torch.Tensor:
       if residue_weights is None:
              _a   :			Dict					 =logits.new_ones(logits.shape[-2]  )

       _a   :			str					 =torch.linspace(0   ,_UpperCAmelCase   ,steps=(no_bins - 1)   ,device=logits.device  )

       _a   :			Optional[int]					 =_calculate_bin_centers(_UpperCAmelCase  )
       torch.sum(_UpperCAmelCase  )
       _a   :			List[str]					 =logits.shape[-2]
       _a   :			List[Any]					 =max(_UpperCAmelCase   ,19  )

       _a   :			Union[str, Any]					 =1.2_4 * (clipped_n - 15) ** (1.0 / 3) - 1.8

       _a   :			Tuple					 =torch.nn.functional.softmax(_UpperCAmelCase   ,dim=-1  )

       _a   :			Dict					 =1.0 / (1 + (bin_centers**2) / (da**2))
       _a   :			int					 =torch.sum(probs * tm_per_bin   ,dim=-1  )

       _a   :			List[Any]					 =residue_weights / (eps + residue_weights.sum())
       _a   :			List[str]					 =torch.sum(predicted_tm_term * normed_residue_mask   ,dim=-1  )

       _a   :			int					 =per_alignment * residue_weights

       _a   :			List[Any]					 =(weighted == torch.max(_UpperCAmelCase  )).nonzero()[0]
       return per_alignment[tuple(_UpperCAmelCase  )]


