

from __future__ import annotations

import collections
import tempfile
import unittest

import numpy as np

from transformers.testing_utils import require_tf, require_vision, slow
from transformers.utils import is_tf_available, is_vision_available

from ...test_modeling_tf_common import floats_tensor, ids_tensor, random_attention_mask
from ..bert.test_modeling_tf_bert import TFBertModelTester
from ..clip.test_modeling_tf_clip import TFCLIPVisionModelTester
from ..deit.test_modeling_tf_deit import TFDeiTModelTester
from ..roberta.test_modeling_tf_roberta import TFRobertaModelTester
from ..vit.test_modeling_tf_vit import TFViTModelTester


if is_tf_available():
				from transformers import (
				    TFBertModel,
				    TFCLIPVisionModel,
				    TFDeiTModel,
				    TFRobertaModel,
				    TFVisionTextDualEncoderModel,
				    TFViTModel,
				    VisionTextDualEncoderConfig,
				)

if is_vision_available():
				from PIL import Image

				from transformers import VisionTextDualEncoderProcessor



def    __lowerCAmelCase   (    A     ):
	if isinstance(__lowercase							,  collections.abc.Iterable     ):
		return x
	return (x, x)

@require_tf
class     __UpperCamelCase     :




	def    __A       (		self						:    Union[str, Any]		,						lowerCAmelCase						:    Optional[int]		,						lowerCAmelCase						:    Any				):


		'''simple docstring'''







		pass




	def    __A       (		self						:    Union[str, Any]				):


		'''simple docstring'''







		pass




	def    __A       (		self						:    Union[str, Any]				):


		'''simple docstring'''







		pass




	def    __A       (		self						:    List[Any]		,						lowerCAmelCase						:    str		,						lowerCAmelCase						:    Optional[Any]		,						lowerCAmelCase						:    Dict		,						lowerCAmelCase						:    List[str]		,						lowerCAmelCase						:    Tuple=None		,						**lowerCAmelCase						:    List[str]				):


		'''simple docstring'''







		UpperCAmelCase_												=      VisionTextDualEncoderConfig.from_vision_text_configs(A__		,						A__				)

		UpperCAmelCase_												=      TFVisionTextDualEncoderModel(A__				)

		UpperCAmelCase_												=      model(input_ids=A__		,						pixel_values=A__		,						attention_mask=A__				)

		self.assertEqual(output["text_embeds"].shape		,						(input_ids.shape[0], config.projection_dim)				)
		self.assertEqual(output["image_embeds"].shape		,						(pixel_values.shape[0], config.projection_dim)				)




	def    __A       (		self						:    str		,						lowerCAmelCase						:    Optional[Any]		,						lowerCAmelCase						:    int		,						lowerCAmelCase						:    Tuple		,						lowerCAmelCase						:    Union[str, Any]		,						lowerCAmelCase						:    Tuple=None		,						**lowerCAmelCase						:    int				):


		'''simple docstring'''







		UpperCAmelCase_												=      self.get_vision_text_model(A__		,						A__				)
		UpperCAmelCase_												=      TFVisionTextDualEncoderModel(vision_model=A__		,						text_model=A__				)

		UpperCAmelCase_												=      model(input_ids=A__		,						pixel_values=A__		,						attention_mask=A__				)

		self.assertEqual(output["text_embeds"].shape		,						(input_ids.shape[0], model.config.projection_dim)				)
		self.assertEqual(output["image_embeds"].shape		,						(pixel_values.shape[0], model.config.projection_dim)				)




	def    __A       (		self						:    Optional[int]		,						lowerCAmelCase						:    Any		,						lowerCAmelCase						:    Dict		,						lowerCAmelCase						:    Tuple		,						lowerCAmelCase						:    Optional[Any]		,						lowerCAmelCase						:    Any=None		,						**lowerCAmelCase						:    Optional[int]				):


		'''simple docstring'''







		UpperCAmelCase_												=      self.get_vision_text_model(A__		,						A__				)
		UpperCAmelCase_												=      {'''vision_model''': vision_model, '''text_model''': text_model}
		UpperCAmelCase_												=      TFVisionTextDualEncoderModel.from_vision_text_pretrained(**A__				)

		UpperCAmelCase_												=      model(input_ids=A__		,						pixel_values=A__		,						attention_mask=A__				)

		self.assertEqual(output["text_embeds"].shape		,						(input_ids.shape[0], model.config.projection_dim)				)
		self.assertEqual(output["image_embeds"].shape		,						(pixel_values.shape[0], model.config.projection_dim)				)




	def    __A       (		self						:    Optional[int]		,						lowerCAmelCase						:    List[Any]		,						lowerCAmelCase						:    Dict		,						lowerCAmelCase						:    List[str]		,						lowerCAmelCase						:    Optional[int]		,						lowerCAmelCase						:    Optional[Any]=None		,						**lowerCAmelCase						:    Dict				):


		'''simple docstring'''







		UpperCAmelCase_												=      self.get_vision_text_model(A__		,						A__				)
		UpperCAmelCase_												=      TFVisionTextDualEncoderModel(vision_model=A__		,						text_model=A__				)

		UpperCAmelCase_												=      model(input_ids=A__		,						pixel_values=A__		,						attention_mask=A__				)
		UpperCAmelCase_												=      output[0].numpy()

		with tempfile.TemporaryDirectory() as tmpdirname:
			model.save_pretrained(A__				)
			UpperCAmelCase_												=      TFVisionTextDualEncoderModel.from_pretrained(A__				)

			UpperCAmelCase_												=      model(input_ids=A__		,						pixel_values=A__		,						attention_mask=A__				)
			UpperCAmelCase_												=      after_output[0].numpy()
			UpperCAmelCase_												=      np.amax(np.abs(out_a - out_a				)				)
			self.assertLessEqual(A__		,						1e-5				)




	def    __A       (		self						:    Dict		,						lowerCAmelCase						:    List[str]		,						lowerCAmelCase						:    Union[str, Any]		,						lowerCAmelCase						:    List[Any]		,						lowerCAmelCase						:    Any		,						lowerCAmelCase						:    Optional[int]=None		,						**lowerCAmelCase						:    int				):


		'''simple docstring'''







		UpperCAmelCase_												=      self.get_vision_text_model(A__		,						A__				)
		UpperCAmelCase_												=      TFVisionTextDualEncoderModel(vision_model=A__		,						text_model=A__				)

		UpperCAmelCase_												=      model(
		    input_ids=A__		,						pixel_values=A__		,						attention_mask=A__		,						output_attentions=A__				)

		UpperCAmelCase_												=      output.vision_model_output.attentions
		self.assertEqual(len(A__				)		,						vision_config.num_hidden_layers				)

		# in ViT, the seq_len equals the number of patches + 1 (we add 1 for the [CLS] token)
		UpperCAmelCase_												=      to_atuple(vision_model.config.image_size				)
		UpperCAmelCase_												=      to_atuple(vision_model.config.patch_size				)
		UpperCAmelCase_												=      (image_size[1] // patch_size[1]) * (image_size[0] // patch_size[0])
		UpperCAmelCase_												=      num_patches + 1
		self.assertEqual(vision_attentions[0].shape[-3:]		,						(vision_config.num_attention_heads, seq_len, seq_len)				)

		UpperCAmelCase_												=      output.text_model_output.attentions
		self.assertEqual(len(A__				)		,						text_config.num_hidden_layers				)

		self.assertEqual(
		    text_attentions[0].shape[-3:]		,						(text_config.num_attention_heads, input_ids.shape[-1], input_ids.shape[-1])		,						)




	def    __A       (		self						:    Union[str, Any]		,						lowerCAmelCase						:    Tuple		,						lowerCAmelCase						:    Union[str, Any]		,						lowerCAmelCase						:    Tuple				):


		'''simple docstring'''







		UpperCAmelCase_												=      np.abs((a - b)				).max()
		self.assertLessEqual(A__		,						A__		,						F"Difference between torch and flax is {diff} (>= {tol})."				)




	def    __A       (		self						:    Optional[int]				):


		'''simple docstring'''







		UpperCAmelCase_												=      self.prepare_config_and_inputs()
		self.check_vision_text_dual_encoder_model(**A__				)




	def    __A       (		self						:    int				):


		'''simple docstring'''







		UpperCAmelCase_												=      self.prepare_config_and_inputs()
		self.check_model_from_pretrained_configs(**A__				)




	def    __A       (		self						:    Optional[int]				):


		'''simple docstring'''







		UpperCAmelCase_												=      self.prepare_config_and_inputs()
		self.check_vision_text_dual_encoder_from_pretrained(**A__				)




	def    __A       (		self						:    Union[str, Any]				):


		'''simple docstring'''







		UpperCAmelCase_												=      self.prepare_config_and_inputs()
		self.check_save_load(**A__				)




	def    __A       (		self						:    Union[str, Any]				):


		'''simple docstring'''







		UpperCAmelCase_												=      self.prepare_config_and_inputs()
		self.check_vision_text_output_attention(**A__				)




	@slow
	def    __A       (		self						:    Tuple				):


		'''simple docstring'''







		UpperCAmelCase_												=      self.get_pretrained_model_and_inputs()

		UpperCAmelCase_												=      model_a(**A__				)
		UpperCAmelCase_												=      outputs[0].numpy()

		with tempfile.TemporaryDirectory() as tmp_dirname:
			model_a.save_pretrained(A__				)
			UpperCAmelCase_												=      TFVisionTextDualEncoderModel.from_pretrained(A__				)

			UpperCAmelCase_												=      model_a(**A__				)
			UpperCAmelCase_												=      after_outputs[0].numpy()
			UpperCAmelCase_												=      np.amax(np.abs(out_a - out_a				)				)
			self.assertLessEqual(A__		,						1e-5				)


@require_tf
class     __UpperCamelCase     (       __snake_case  ,  unittest.TestCase    ):




	def    __A       (		self						:    Tuple				):


		'''simple docstring'''







		UpperCAmelCase_												=      TFVisionTextDualEncoderModel.from_vision_text_pretrained(
		    "hf-internal-testing/tiny-random-vit"		,						"hf-internal-testing/tiny-random-bert"				)
		UpperCAmelCase_												=      13
		UpperCAmelCase_												=      floats_tensor(
		    [
		        batch_size,
		        model.vision_model.config.num_channels,
		        model.vision_model.config.image_size,
		        model.vision_model.config.image_size,
		    ]				)
		UpperCAmelCase_												=      ids_tensor([batch_size, 4]		,						model.text_model.config.vocab_size				)
		UpperCAmelCase_												=      random_attention_mask([batch_size, 4]				)
		UpperCAmelCase_												=      {'''pixel_values''': pixel_values, '''input_ids''': input_ids, '''attention_mask''': attention_mask}

		return model, inputs




	def    __A       (		self						:    Tuple		,						lowerCAmelCase						:    Tuple		,						lowerCAmelCase						:    Dict				):


		'''simple docstring'''







		UpperCAmelCase_												=      TFViTModel(A__		,						name="vision_model"				)
		UpperCAmelCase_												=      TFBertModel(A__		,						name="text_model"				)
		return vision_model, text_model




	def    __A       (		self						:    str				):


		'''simple docstring'''







		UpperCAmelCase_												=      TFViTModelTester(self				)
		UpperCAmelCase_												=      TFBertModelTester(self				)
		UpperCAmelCase_												=      vit_model_tester.prepare_config_and_inputs()
		UpperCAmelCase_												=      bert_model_tester.prepare_config_and_inputs()

		UpperCAmelCase_												=      vision_config_and_inputs

		(
		    UpperCAmelCase_				
		)												=      text_config_and_inputs

		return {
		    "text_config": text_config,
		    "vision_config": vision_config,
		    "pixel_values": pixel_values,
		    "attention_mask": input_mask,
		    "input_ids": input_ids,
		    "text_token_type_ids": token_type_ids,
		    "text_sequence_labels": sequence_labels,
		    "text_token_labels": token_labels,
		    "text_choice_labels": choice_labels,
		}


@require_tf
class     __UpperCamelCase     (       __snake_case  ,  unittest.TestCase    ):




	def    __A       (		self						:    int				):


		'''simple docstring'''







		UpperCAmelCase_												=      TFVisionTextDualEncoderModel.from_vision_text_pretrained(
		    "Rocketknight1/tiny-random-deit-tf"		,						"hf-internal-testing/tiny-random-roberta"				)
		UpperCAmelCase_												=      13
		UpperCAmelCase_												=      floats_tensor(
		    [
		        batch_size,
		        model.vision_model.config.num_channels,
		        model.vision_model.config.image_size,
		        model.vision_model.config.image_size,
		    ]				)
		UpperCAmelCase_												=      ids_tensor([batch_size, 4]		,						model.text_model.config.vocab_size				)
		UpperCAmelCase_												=      random_attention_mask([batch_size, 4]				)
		UpperCAmelCase_												=      {'''pixel_values''': pixel_values, '''input_ids''': input_ids, '''attention_mask''': attention_mask}

		return model, inputs




	def    __A       (		self						:    Optional[int]		,						lowerCAmelCase						:    Any		,						lowerCAmelCase						:    str		,						lowerCAmelCase						:    Any		,						lowerCAmelCase						:    Optional[int]		,						lowerCAmelCase						:    Tuple=None		,						**lowerCAmelCase						:    List[Any]				):


		'''simple docstring'''







		UpperCAmelCase_												=      self.get_vision_text_model(A__		,						A__				)
		UpperCAmelCase_												=      TFVisionTextDualEncoderModel(vision_model=A__		,						text_model=A__				)

		UpperCAmelCase_												=      model(
		    input_ids=A__		,						pixel_values=A__		,						attention_mask=A__		,						output_attentions=A__				)

		UpperCAmelCase_												=      output.vision_model_output.attentions
		self.assertEqual(len(A__				)		,						vision_config.num_hidden_layers				)

		# in DEiT, the seq_len equals the number of patches + 2 (we add 2 for the [CLS] and distillation tokens)
		UpperCAmelCase_												=      to_atuple(vision_model.config.image_size				)
		UpperCAmelCase_												=      to_atuple(vision_model.config.patch_size				)
		UpperCAmelCase_												=      (image_size[1] // patch_size[1]) * (image_size[0] // patch_size[0])
		UpperCAmelCase_												=      num_patches + 2
		self.assertEqual(vision_attentions[0].shape[-3:]		,						(vision_config.num_attention_heads, seq_len, seq_len)				)

		UpperCAmelCase_												=      output.text_model_output.attentions
		self.assertEqual(len(A__				)		,						text_config.num_hidden_layers				)

		self.assertEqual(
		    text_attentions[0].shape[-3:]		,						(text_config.num_attention_heads, input_ids.shape[-1], input_ids.shape[-1])		,						)




	def    __A       (		self						:    List[str]		,						lowerCAmelCase						:    int		,						lowerCAmelCase						:    List[str]				):


		'''simple docstring'''







		UpperCAmelCase_												=      TFDeiTModel(A__		,						name="vision_model"				)
		UpperCAmelCase_												=      TFRobertaModel(A__		,						name="text_model"				)
		return vision_model, text_model




	def    __A       (		self						:    List[Any]				):


		'''simple docstring'''







		UpperCAmelCase_												=      TFDeiTModelTester(self				)
		UpperCAmelCase_												=      TFRobertaModelTester(self				)
		UpperCAmelCase_												=      vit_model_tester.prepare_config_and_inputs()
		UpperCAmelCase_												=      bert_model_tester.prepare_config_and_inputs()

		UpperCAmelCase_												=      vision_config_and_inputs

		(
		    UpperCAmelCase_				
		)												=      text_config_and_inputs

		return {
		    "text_config": text_config,
		    "vision_config": vision_config,
		    "pixel_values": pixel_values,
		    "attention_mask": input_mask,
		    "input_ids": input_ids,
		    "text_token_type_ids": token_type_ids,
		    "text_sequence_labels": sequence_labels,
		    "text_token_labels": token_labels,
		    "text_choice_labels": choice_labels,
		}


@require_tf
class     __UpperCamelCase     (       __snake_case  ,  unittest.TestCase    ):




	def    __A       (		self						:    Union[str, Any]				):


		'''simple docstring'''







		UpperCAmelCase_												=      TFVisionTextDualEncoderModel.from_vision_text_pretrained(
		    "Rocketknight1/tiny-random-clip-tf"		,						"hf-internal-testing/tiny-random-bert"				)
		UpperCAmelCase_												=      13
		UpperCAmelCase_												=      floats_tensor(
		    [
		        batch_size,
		        model.vision_model.config.num_channels,
		        model.vision_model.config.image_size,
		        model.vision_model.config.image_size,
		    ]				)
		UpperCAmelCase_												=      ids_tensor([batch_size, 4]		,						model.text_model.config.vocab_size				)
		UpperCAmelCase_												=      random_attention_mask([batch_size, 4]				)
		UpperCAmelCase_												=      {'''pixel_values''': pixel_values, '''input_ids''': input_ids, '''attention_mask''': attention_mask}

		return model, inputs




	def    __A       (		self						:    List[Any]		,						lowerCAmelCase						:    Tuple		,						lowerCAmelCase						:    Any				):


		'''simple docstring'''







		UpperCAmelCase_												=      TFCLIPVisionModel(A__		,						name="vision_model"				)
		UpperCAmelCase_												=      TFBertModel(A__		,						name="text_model"				)
		return vision_model, text_model




	def    __A       (		self						:    List[str]				):


		'''simple docstring'''







		UpperCAmelCase_												=      TFCLIPVisionModelTester(self				)
		UpperCAmelCase_												=      TFBertModelTester(self				)
		UpperCAmelCase_												=      clip_model_tester.prepare_config_and_inputs()
		UpperCAmelCase_												=      bert_model_tester.prepare_config_and_inputs()

		UpperCAmelCase_												=      vision_config_and_inputs

		(
		    UpperCAmelCase_				
		)												=      text_config_and_inputs

		return {
		    "text_config": text_config,
		    "vision_config": vision_config,
		    "pixel_values": pixel_values,
		    "attention_mask": input_mask,
		    "input_ids": input_ids,
		    "text_token_type_ids": token_type_ids,
		    "text_sequence_labels": sequence_labels,
		    "text_token_labels": token_labels,
		    "text_choice_labels": choice_labels,
		}


@require_vision
@require_tf
class     __UpperCamelCase     (       unittest.TestCase    ):




	@slow
	def    __A       (		self						:    str				):


		'''simple docstring'''







		UpperCAmelCase_												=      TFVisionTextDualEncoderModel.from_pretrained(
		    "clip-italian/clip-italian"		,						logit_scale_init_value=1.0		,						from_pt=A__				)
		UpperCAmelCase_												=      VisionTextDualEncoderProcessor.from_pretrained("clip-italian/clip-italian"				)

		UpperCAmelCase_												=      Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png"				)
		UpperCAmelCase_												=      processor(
		    text=["una foto di un gatto", "una foto di un cane"]		,						images=A__		,						padding=A__		,						return_tensors="np"				)

		UpperCAmelCase_												=      model(**A__				)

		# verify the logits
		self.assertEqual(outputs.logits_per_image.shape		,						(inputs.pixel_values.shape[0], inputs.input_ids.shape[0])				)
		self.assertEqual(
		    outputs.logits_per_text.shape		,						(inputs.input_ids.shape[0], inputs.pixel_values.shape[0])		,						)

		UpperCAmelCase_												=      np.array([[1.2_284_727, 0.3_104_122]]				)

		self.assertTrue(np.allclose(outputs.logits_per_image.numpy()		,						A__		,						atol=1e-3				)				)







import os
import tempfile
from functools import partial
from unittest import TestCase
from unittest.mock import patch

import numpy as np
import pytest

from datasets.arrow_dataset import Dataset
from datasets.search import ElasticSearchIndex, FaissIndex, MissingIndex

from .utils import require_elasticsearch, require_faiss


_UpperCamelCase			:	Optional[Any]								=pytest.mark.integration






@require_faiss
class 						UpperCAmelCase__ (							__snake_case							):







		def  A__      (					self		):
						_A	:				Any   =      Dataset.from_dict({'''filename''': ['''my_name-train''' + '''_''' + str(A__		) for x in np.arange(30		).tolist()]}		)
						return dset







		def  A__      (					self		):
						import faiss

						_A	:				Dataset   =      self._create_dummy_dataset()
						_A	:				List[str]   =      dset.map(
						    lambda A__       ,A__		: {"vecs": i * np.ones(5       ,dtype=np.floataa		)}       ,with_indices=A__       ,keep_in_memory=A__		)
						_A	:				Tuple   =      dset.add_faiss_index('''vecs'''       ,batch_size=100       ,metric_type=faiss.METRIC_INNER_PRODUCT		)
						_A  ,				_A	:				List[Any]   =      dset.get_nearest_examples('''vecs'''       ,np.ones(5       ,dtype=np.floataa		)		)
						self.assertEqual(examples['''filename'''][0]       ,'''my_name-train_29'''		)
						dset.drop_index('''vecs'''		)







		def  A__      (					self		):
						import faiss

						_A	:				Dataset   =      self._create_dummy_dataset()
						dset.add_faiss_index_from_external_arrays(
						    external_arrays=np.ones((30, 5)		) * np.arange(30		).reshape(-1       ,1		)       ,index_name='''vecs'''       ,batch_size=100       ,metric_type=faiss.METRIC_INNER_PRODUCT       ,)
						_A  ,				_A	:				Tuple   =      dset.get_nearest_examples('''vecs'''       ,np.ones(5       ,dtype=np.floataa		)		)
						self.assertEqual(examples['''filename'''][0]       ,'''my_name-train_29'''		)







		def  A__      (					self		):
						import faiss

						_A	:				Dataset   =      self._create_dummy_dataset()
						dset.add_faiss_index_from_external_arrays(
						    external_arrays=np.ones((30, 5)		) * np.arange(30		).reshape(-1       ,1		)       ,index_name='''vecs'''       ,metric_type=faiss.METRIC_INNER_PRODUCT       ,)

						# Setting delete=False and unlinking manually is not pretty... but it is required on Windows to
						# ensure somewhat stable behaviour. If we don't, we get PermissionErrors. This is an age-old issue.
						# see https://bugs.python.org/issue14243 and
						# https://stackoverflow.com/questions/23212435/permission-denied-to-write-to-my-temporary-file/23212515
						with tempfile.NamedTemporaryFile(delete=A__		) as tmp_file:
										dset.save_faiss_index('''vecs'''       ,tmp_file.name		)
										dset.load_faiss_index('''vecs2'''       ,tmp_file.name		)
						os.unlink(tmp_file.name		)

						_A  ,				_A	:				Optional[int]   =      dset.get_nearest_examples('''vecs2'''       ,np.ones(5       ,dtype=np.floataa		)		)
						self.assertEqual(examples['''filename'''][0]       ,'''my_name-train_29'''		)







		def  A__      (					self		):
						_A	:				Dataset   =      self._create_dummy_dataset()
						dset.add_faiss_index_from_external_arrays(
						    external_arrays=np.ones((30, 5)		) * np.arange(30		).reshape(-1       ,1		)       ,index_name='''vecs'''		)
						dset.drop_index('''vecs'''		)
						self.assertRaises(A__       ,partial(dset.get_nearest_examples       ,'''vecs2'''       ,np.ones(5       ,dtype=np.floataa		)		)		)







		def  A__      (					self		):
						from elasticsearch import Elasticsearch

						_A	:				Dataset   =      self._create_dummy_dataset()
						with patch('''elasticsearch.Elasticsearch.search'''		) as mocked_search, patch(
						    '''elasticsearch.client.IndicesClient.create'''		) as mocked_index_create, patch('''elasticsearch.helpers.streaming_bulk'''		) as mocked_bulk:
										_A	:				str   =      {'''acknowledged''': True}
										mocked_bulk.return_value([(True, None)] * 30		)
										_A	:				Optional[Any]   =      {'''hits''': {'''hits''': [{'''_score''': 1, '''_id''': 29}]}}
										_A	:				Optional[Any]   =      Elasticsearch()

										dset.add_elasticsearch_index('''filename'''       ,es_client=A__		)
										_A  ,				_A	:				Tuple   =      dset.get_nearest_examples('''filename'''       ,'''my_name-train_29'''		)
										self.assertEqual(examples['''filename'''][0]       ,'''my_name-train_29'''		)






@require_faiss
class 						UpperCAmelCase__ (							__snake_case							):







		def  A__      (					self		):
						import faiss

						_A	:				Union[str, Any]   =      FaissIndex(metric_type=faiss.METRIC_INNER_PRODUCT		)

						# add vectors
						index.add_vectors(np.eye(5       ,dtype=np.floataa		)		)
						self.assertIsNotNone(index.faiss_index		)
						self.assertEqual(index.faiss_index.ntotal       ,5		)
						index.add_vectors(np.zeros((5, 5)       ,dtype=np.floataa		)		)
						self.assertEqual(index.faiss_index.ntotal       ,10		)

						# single query
						_A	:				str   =      np.zeros(5       ,dtype=np.floataa		)
						_A	:				Dict   =      1
						_A  ,				_A	:				Optional[int]   =      index.search(A__		)
						self.assertRaises(A__       ,index.search       ,query.reshape(-1       ,1		)		)
						self.assertGreater(scores[0]       ,0		)
						self.assertEqual(indices[0]       ,1		)

						# batched queries
						_A	:				Optional[Any]   =      np.eye(5       ,dtype=np.floataa		)[::-1]
						_A  ,				_A	:				Dict   =      index.search_batch(A__		)
						self.assertRaises(A__       ,index.search_batch       ,queries[0]		)
						_A	:				Tuple   =      [scores[0] for scores in total_scores]
						_A	:				List[str]   =      [indices[0] for indices in total_indices]
						self.assertGreater(np.min(A__		)       ,0		)
						self.assertListEqual([4, 3, 2, 1, 0]       ,A__		)







		def  A__      (					self		):
						import faiss

						_A	:				List[Any]   =      FaissIndex(string_factory='''Flat'''		)
						index.add_vectors(np.eye(5       ,dtype=np.floataa		)		)
						self.assertIsInstance(index.faiss_index       ,faiss.IndexFlat		)
						_A	:				str   =      FaissIndex(string_factory='''LSH'''		)
						index.add_vectors(np.eye(5       ,dtype=np.floataa		)		)
						self.assertIsInstance(index.faiss_index       ,faiss.IndexLSH		)
						with self.assertRaises(A__		):
										_A	:				List[str]   =      FaissIndex(string_factory='''Flat'''       ,custom_index=faiss.IndexFlat(5		)		)







		def  A__      (					self		):
						import faiss

						_A	:				Any   =      faiss.IndexFlat(5		)
						_A	:				Any   =      FaissIndex(custom_index=A__		)
						index.add_vectors(np.eye(5       ,dtype=np.floataa		)		)
						self.assertIsInstance(index.faiss_index       ,faiss.IndexFlat		)







		def  A__      (					self		):
						import faiss

						_A	:				Optional[int]   =      FaissIndex(metric_type=faiss.METRIC_INNER_PRODUCT		)
						index.add_vectors(np.eye(5       ,dtype=np.floataa		)		)

						# Setting delete=False and unlinking manually is not pretty... but it is required on Windows to
						# ensure somewhat stable behaviour. If we don't, we get PermissionErrors. This is an age-old issue.
						# see https://bugs.python.org/issue14243 and
						# https://stackoverflow.com/questions/23212435/permission-denied-to-write-to-my-temporary-file/23212515
						with tempfile.NamedTemporaryFile(delete=A__		) as tmp_file:
										index.save(tmp_file.name		)
										_A	:				str   =      FaissIndex.load(tmp_file.name		)
						os.unlink(tmp_file.name		)

						_A	:				Tuple   =      np.zeros(5       ,dtype=np.floataa		)
						_A	:				Tuple   =      1
						_A  ,				_A	:				Union[str, Any]   =      index.search(A__		)
						self.assertGreater(scores[0]       ,0		)
						self.assertEqual(indices[0]       ,1		)



@require_faiss
def  a__ (__lowercase							:List[Any]   )		-> int:
				import faiss

				_A	:				Any   =      FaissIndex(metric_type=faiss.METRIC_INNER_PRODUCT   )
				index.add_vectors(np.eye(5					,  dtype=np.floataa   )   )

				_A	:				List[Any]   =      '''index.faiss'''
				_A	:				Any   =      f"""mock://{index_name}"""
				index.save(__lowercase					,  storage_options=mockfs.storage_options   )
				_A	:				Dict   =      FaissIndex.load(__lowercase					,  storage_options=mockfs.storage_options   )

				_A	:				Tuple   =      np.zeros(5					,  dtype=np.floataa   )
				_A	:				Union[str, Any]   =      1
				_A  ,				_A	:				Optional[int]   =      index.search(__lowercase   )
				assert scores[0] > 0
				assert indices[0] == 1






@require_elasticsearch
class 						UpperCAmelCase__ (							__snake_case							):







		def  A__      (					self		):
						from elasticsearch import Elasticsearch

						with patch('''elasticsearch.Elasticsearch.search'''		) as mocked_search, patch(
						    '''elasticsearch.client.IndicesClient.create'''		) as mocked_index_create, patch('''elasticsearch.helpers.streaming_bulk'''		) as mocked_bulk:
										_A	:				Tuple   =      Elasticsearch()
										_A	:				Tuple   =      {'''acknowledged''': True}
										_A	:				int   =      ElasticSearchIndex(es_client=A__		)
										mocked_bulk.return_value([(True, None)] * 3		)
										index.add_documents(['''foo''', '''bar''', '''foobar''']		)

										# single query
										_A	:				Dict   =      '''foo'''
										_A	:				Optional[Any]   =      {'''hits''': {'''hits''': [{'''_score''': 1, '''_id''': 0}]}}
										_A  ,				_A	:				Optional[Any]   =      index.search(A__		)
										self.assertEqual(scores[0]       ,1		)
										self.assertEqual(indices[0]       ,0		)

										# single query with timeout
										_A	:				Union[str, Any]   =      '''foo'''
										_A	:				Optional[int]   =      {'''hits''': {'''hits''': [{'''_score''': 1, '''_id''': 0}]}}
										_A  ,				_A	:				Optional[int]   =      index.search(A__       ,request_timeout=30		)
										self.assertEqual(scores[0]       ,1		)
										self.assertEqual(indices[0]       ,0		)

										# batched queries
										_A	:				Dict   =      ['''foo''', '''bar''', '''foobar''']
										_A	:				Dict   =      {'''hits''': {'''hits''': [{'''_score''': 1, '''_id''': 1}]}}
										_A  ,				_A	:				Dict   =      index.search_batch(A__		)
										_A	:				Union[str, Any]   =      [scores[0] for scores in total_scores]
										_A	:				Tuple   =      [indices[0] for indices in total_indices]
										self.assertGreater(np.min(A__		)       ,0		)
										self.assertListEqual([1, 1, 1]       ,A__		)

										# batched queries with timeout
										_A	:				Union[str, Any]   =      ['''foo''', '''bar''', '''foobar''']
										_A	:				int   =      {'''hits''': {'''hits''': [{'''_score''': 1, '''_id''': 1}]}}
										_A  ,				_A	:				str   =      index.search_batch(A__       ,request_timeout=30		)
										_A	:				Dict   =      [scores[0] for scores in total_scores]
										_A	:				List[str]   =      [indices[0] for indices in total_indices]
										self.assertGreater(np.min(A__		)       ,0		)
										self.assertListEqual([1, 1, 1]       ,A__		)



