






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


__lowerCAmelCase							="block_records.npy"


__lowerCAmelCase							=logging.get_logger(__name__)

def 							a	(   _UpperCAmelCase					,  _UpperCAmelCase )				->				np.ndarray:






					"""simple docstring"""







					import tensorflow.compat.va as tf

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

					return np_record


class    _snake_case   :

					"""simple docstring"""





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

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

										a_									=   builder.build()




					def  __SCREAMING_SNAKE_CASE (     self   , UpperCAmelCase__     )      ->	List[str]:
										a_							,					a_									=   self.searcher.search_batched(question_projection.detach().cpu()     )
										return retrieved_block_ids.astype('int64'     )



class    _snake_case   :

					"""simple docstring"""





					def __init__(     self   , UpperCAmelCase__   , UpperCAmelCase__     )      ->	Any:
										super().__init__()
										a_									=   block_records
										a_									=   tokenizer




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

										a_									=   self.tokenizer.decode(question_input_ids[0]   , skip_special_tokens=UpperCAmelCase__     )

										a_									=   []
										a_									=   []
										for retrieved_block in retrieved_blocks:
															text.append(UpperCAmelCase__     )
															text_pair.append(retrieved_block.decode()     )

										a_									=   self.tokenizer(
										    UpperCAmelCase__   , UpperCAmelCase__   , padding=UpperCAmelCase__   , truncation=UpperCAmelCase__   , return_special_tokens_mask=UpperCAmelCase__   , max_length=UpperCAmelCase__     )
										a_									=   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  __SCREAMING_SNAKE_CASE (     cls   , UpperCAmelCase__   , *UpperCAmelCase__   , **UpperCAmelCase__     )      ->	str:
										if os.path.isdir(UpperCAmelCase__     ):
															a_									=   os.path.join(UpperCAmelCase__   , _REALM_BLOCK_RECORDS_FILENAME     )
										else:
															a_									=   hf_hub_download(
															    repo_id=UpperCAmelCase__   , filename=_REALM_BLOCK_RECORDS_FILENAME   , **UpperCAmelCase__     )
										a_									=   np.load(UpperCAmelCase__   , allow_pickle=UpperCAmelCase__     )

										a_									=   AutoTokenizer.from_pretrained(UpperCAmelCase__   , *UpperCAmelCase__   , **UpperCAmelCase__     )

										return cls(UpperCAmelCase__   , UpperCAmelCase__     )




					def  __SCREAMING_SNAKE_CASE (     self   , UpperCAmelCase__     )      ->	List[str]:
										# save block records
										np.save(os.path.join(UpperCAmelCase__   , _REALM_BLOCK_RECORDS_FILENAME     )   , self.block_records     )
										# save tokenizer
										self.tokenizer.save_pretrained(UpperCAmelCase__     )




					def  __SCREAMING_SNAKE_CASE (     self   , UpperCAmelCase__   , UpperCAmelCase__     )      ->	Optional[int]:
										a_									=   []
										a_									=   []
										a_									=   []
										a_									=   0

										for input_id in concat_inputs.input_ids:
															a_									=   input_id.tolist()
															# Check answers between two [SEP] tokens
															a_									=   input_id_list.index(self.tokenizer.sep_token_id     )
															a_									=   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:
																									a_									=   len(start_pos[-1]     )

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









'''simple docstring'''
import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
						import torch

if is_vision_available():
						from PIL import Image

						from transformers import ConvNextImageProcessor


class    _snake_case   (				unittest.TestCase					):

					"""simple docstring"""





					def __init__(     self   , UpperCAmelCase__   , UpperCAmelCase__=7   , UpperCAmelCase__=3   , UpperCAmelCase__=18   , UpperCAmelCase__=30   , UpperCAmelCase__=400   , UpperCAmelCase__=True   , UpperCAmelCase__=None   , UpperCAmelCase__=0.8_7_5   , UpperCAmelCase__=True   , UpperCAmelCase__=[0.5, 0.5, 0.5]   , UpperCAmelCase__=[0.5, 0.5, 0.5]   , )      ->	Optional[int]:
										a_									=   size if size is not None else {'shortest_edge': 20}
										a_									=   parent
										a_									=   batch_size
										a_									=   num_channels
										a_									=   image_size
										a_									=   min_resolution
										a_									=   max_resolution
										a_									=   do_resize
										a_									=   size
										a_									=   crop_pct
										a_									=   do_normalize
										a_									=   image_mean
										a_									=   image_std




					def  __SCREAMING_SNAKE_CASE (     self     )      ->	List[str]:
										return {
										    "image_mean": self.image_mean,
										    "image_std": self.image_std,
										    "do_normalize": self.do_normalize,
										    "do_resize": self.do_resize,
										    "size": self.size,
										    "crop_pct": self.crop_pct,
										}



@require_torch
@require_vision
class    _snake_case   (				snake_case     ,					unittest.TestCase					):

					"""simple docstring"""





					_UpperCamelCase		 =      ConvNextImageProcessor if is_vision_available() else None




					def  __SCREAMING_SNAKE_CASE (     self     )      ->	Tuple:
										a_									=   ConvNextImageProcessingTester(self     )




					@property
					def  __SCREAMING_SNAKE_CASE (     self     )      ->	Dict:
										return self.image_processor_tester.prepare_image_processor_dict()




					def  __SCREAMING_SNAKE_CASE (     self     )      ->	Any:
										a_									=   self.image_processing_class(**self.image_processor_dict     )
										self.assertTrue(hasattr(UpperCAmelCase__   , 'do_resize'     )     )
										self.assertTrue(hasattr(UpperCAmelCase__   , 'size'     )     )
										self.assertTrue(hasattr(UpperCAmelCase__   , 'crop_pct'     )     )
										self.assertTrue(hasattr(UpperCAmelCase__   , 'do_normalize'     )     )
										self.assertTrue(hasattr(UpperCAmelCase__   , 'image_mean'     )     )
										self.assertTrue(hasattr(UpperCAmelCase__   , 'image_std'     )     )




					def  __SCREAMING_SNAKE_CASE (     self     )      ->	Tuple:
										a_									=   self.image_processing_class.from_dict(self.image_processor_dict     )
										self.assertEqual(image_processor.size   , {'shortest_edge': 20}     )

										a_									=   self.image_processing_class.from_dict(self.image_processor_dict   , size=42     )
										self.assertEqual(image_processor.size   , {'shortest_edge': 42}     )




					def  __SCREAMING_SNAKE_CASE (     self     )      ->	Optional[Any]:
										pass




					def  __SCREAMING_SNAKE_CASE (     self     )      ->	Tuple:
										# Initialize image_processing
										a_									=   self.image_processing_class(**self.image_processor_dict     )
										# create random PIL images
										a_									=   prepare_image_inputs(self.image_processor_tester   , equal_resolution=UpperCAmelCase__     )
										for image in image_inputs:
															self.assertIsInstance(UpperCAmelCase__   , Image.Image     )

										# Test not batched input
										a_									=   image_processing(image_inputs[0]   , return_tensors='pt'     ).pixel_values
										self.assertEqual(
										    encoded_images.shape   , (
										        1,
										        self.image_processor_tester.num_channels,
										        self.image_processor_tester.size['shortest_edge'],
										        self.image_processor_tester.size['shortest_edge'],
										    )   , )

										# Test batched
										a_									=   image_processing(UpperCAmelCase__   , return_tensors='pt'     ).pixel_values
										self.assertEqual(
										    encoded_images.shape   , (
										        self.image_processor_tester.batch_size,
										        self.image_processor_tester.num_channels,
										        self.image_processor_tester.size['shortest_edge'],
										        self.image_processor_tester.size['shortest_edge'],
										    )   , )




					def  __SCREAMING_SNAKE_CASE (     self     )      ->	int:
										# Initialize image_processing
										a_									=   self.image_processing_class(**self.image_processor_dict     )
										# create random numpy tensors
										a_									=   prepare_image_inputs(self.image_processor_tester   , equal_resolution=UpperCAmelCase__   , numpify=UpperCAmelCase__     )
										for image in image_inputs:
															self.assertIsInstance(UpperCAmelCase__   , np.ndarray     )

										# Test not batched input
										a_									=   image_processing(image_inputs[0]   , return_tensors='pt'     ).pixel_values
										self.assertEqual(
										    encoded_images.shape   , (
										        1,
										        self.image_processor_tester.num_channels,
										        self.image_processor_tester.size['shortest_edge'],
										        self.image_processor_tester.size['shortest_edge'],
										    )   , )

										# Test batched
										a_									=   image_processing(UpperCAmelCase__   , return_tensors='pt'     ).pixel_values
										self.assertEqual(
										    encoded_images.shape   , (
										        self.image_processor_tester.batch_size,
										        self.image_processor_tester.num_channels,
										        self.image_processor_tester.size['shortest_edge'],
										        self.image_processor_tester.size['shortest_edge'],
										    )   , )




					def  __SCREAMING_SNAKE_CASE (     self     )      ->	int:
										# Initialize image_processing
										a_									=   self.image_processing_class(**self.image_processor_dict     )
										# create random PyTorch tensors
										a_									=   prepare_image_inputs(self.image_processor_tester   , equal_resolution=UpperCAmelCase__   , torchify=UpperCAmelCase__     )
										for image in image_inputs:
															self.assertIsInstance(UpperCAmelCase__   , torch.Tensor     )

										# Test not batched input
										a_									=   image_processing(image_inputs[0]   , return_tensors='pt'     ).pixel_values
										self.assertEqual(
										    encoded_images.shape   , (
										        1,
										        self.image_processor_tester.num_channels,
										        self.image_processor_tester.size['shortest_edge'],
										        self.image_processor_tester.size['shortest_edge'],
										    )   , )

										# Test batched
										a_									=   image_processing(UpperCAmelCase__   , return_tensors='pt'     ).pixel_values
										self.assertEqual(
										    encoded_images.shape   , (
										        self.image_processor_tester.batch_size,
										        self.image_processor_tester.num_channels,
										        self.image_processor_tester.size['shortest_edge'],
										        self.image_processor_tester.size['shortest_edge'],
										    )   , )

