


from typing import Any, Dict, Optional

import torch
import torch.nn.functional as F
from torch import nn

from ..utils import maybe_allow_in_graph
from .activations import get_activation
from .attention_processor import Attention
from .embeddings import CombinedTimestepLabelEmbeddings






@maybe_allow_in_graph
class _lowerCamelCase(		nn.Module  ):
       def __init__(   self, lowerCamelCase, lowerCamelCase, lowerCamelCase, lowerCamelCase=0.0, lowerCamelCase = None, lowerCamelCase = "geglu", lowerCamelCase = None, lowerCamelCase = False, lowerCamelCase = False, lowerCamelCase = False, lowerCamelCase = False, lowerCamelCase = True, lowerCamelCase = "layer_norm", lowerCamelCase = False, )			->  Dict:




           """simple docstring"""

           super().__init__()
           _lowercase			:				Union[str, Any] 						=						only_cross_attention

           _lowercase			:				Union[str, Any] 						=						(num_embeds_ada_norm is not None) and norm_type == 'ada_norm_zero'
           _lowercase			:				List[str] 						=						(num_embeds_ada_norm is not None) and norm_type == 'ada_norm'

           if norm_type in ("ada_norm", "ada_norm_zero") and num_embeds_ada_norm is None:
               raise ValueError(
                   F'''`norm_type` is set to {norm_type}, but `num_embeds_ada_norm` is not defined. Please make sure to'''
                   F''' define `num_embeds_ada_norm` if setting `norm_type` to {norm_type}.''')

           # Define 3 blocks. Each block has its own normalization layer.
           # 1. Self-Attn
           if self.use_ada_layer_norm:
               _lowercase			:				Union[str, Any] 						=						AdaLayerNorm(lowerCamelCase, lowerCamelCase)
           elif self.use_ada_layer_norm_zero:
               _lowercase			:				List[str] 						=						AdaLayerNormZero(lowerCamelCase, lowerCamelCase)
           else:
               _lowercase			:				Tuple 						=						nn.LayerNorm(lowerCamelCase, elementwise_affine=lowerCamelCase)
           _lowercase			:				Optional[int] 						=						Attention(
               query_dim=lowerCamelCase, heads=lowerCamelCase, dim_head=lowerCamelCase, dropout=lowerCamelCase, bias=lowerCamelCase, cross_attention_dim=cross_attention_dim if only_cross_attention else None, upcast_attention=lowerCamelCase, )

           # 2. Cross-Attn
           if cross_attention_dim is not None or double_self_attention:
               # We currently only use AdaLayerNormZero for self attention where there will only be one attention block.
               # I.e. the number of returned modulation chunks from AdaLayerZero would not make sense if returned during
               # the second cross attention block.
               _lowercase			:				Dict 						=						(
                   AdaLayerNorm(lowerCamelCase, lowerCamelCase)
                   if self.use_ada_layer_norm
                   else nn.LayerNorm(lowerCamelCase, elementwise_affine=lowerCamelCase)
               )
               _lowercase			:				str 						=						Attention(
                   query_dim=lowerCamelCase, cross_attention_dim=cross_attention_dim if not double_self_attention else None, heads=lowerCamelCase, dim_head=lowerCamelCase, dropout=lowerCamelCase, bias=lowerCamelCase, upcast_attention=lowerCamelCase, )  # is self-attn if encoder_hidden_states is none
           else:
               _lowercase			:				Optional[int] 						=						None
               _lowercase			:				List[Any] 						=						None

           # 3. Feed-forward
           _lowercase			:				Dict 						=						nn.LayerNorm(lowerCamelCase, elementwise_affine=lowerCamelCase)
           _lowercase			:				Dict 						=						FeedForward(lowerCamelCase, dropout=lowerCamelCase, activation_fn=lowerCamelCase, final_dropout=lowerCamelCase)

           # let chunk size default to None
           _lowercase			:				Union[str, Any] 						=						None
           _lowercase			:				str 						=						0
       def      UpperCamelCase	(   self, lowerCamelCase, lowerCamelCase)			->  Tuple:




           """simple docstring"""

           _lowercase			:				Any 						=						chunk_size
           _lowercase			:				Union[str, Any] 						=						dim




       def      UpperCamelCase	(   self, lowerCamelCase, lowerCamelCase = None, lowerCamelCase = None, lowerCamelCase = None, lowerCamelCase = None, lowerCamelCase = None, lowerCamelCase = None, )			->  List[Any]:




           """simple docstring"""

           if self.use_ada_layer_norm:
               _lowercase			:				List[str] 						=						self.norma(lowerCamelCase, lowerCamelCase)
           elif self.use_ada_layer_norm_zero:
               _lowercase						,							_lowercase						,							_lowercase						,							_lowercase						,							_lowercase			:				Any 						=						self.norma(
                   lowerCamelCase, lowerCamelCase, lowerCamelCase, hidden_dtype=hidden_states.dtype)
           else:
               _lowercase			:				List[Any] 						=						self.norma(lowerCamelCase)

           _lowercase			:				List[Any] 						=						cross_attention_kwargs if cross_attention_kwargs is not None else {}

           _lowercase			:				Any 						=						self.attna(
               lowerCamelCase, encoder_hidden_states=encoder_hidden_states if self.only_cross_attention else None, attention_mask=lowerCamelCase, **lowerCamelCase, )
           if self.use_ada_layer_norm_zero:
               _lowercase			:				List[Any] 						=						gate_msa.unsqueeze(1) * attn_output
           _lowercase			:				List[Any] 						=						attn_output + hidden_states

           # 2. Cross-Attention
           if self.attna is not None:
               _lowercase			:				List[Any] 						=						(
                   self.norma(lowerCamelCase, lowerCamelCase) if self.use_ada_layer_norm else self.norma(lowerCamelCase)
               )

               _lowercase			:				Any 						=						self.attna(
                   lowerCamelCase, encoder_hidden_states=lowerCamelCase, attention_mask=lowerCamelCase, **lowerCamelCase, )
               _lowercase			:				Optional[int] 						=						attn_output + hidden_states

           # 3. Feed-forward
           _lowercase			:				List[str] 						=						self.norma(lowerCamelCase)

           if self.use_ada_layer_norm_zero:
               _lowercase			:				str 						=						norm_hidden_states * (1 + scale_mlp[:, None]) + shift_mlp[:, None]

           if self._chunk_size is not None:
               # "feed_forward_chunk_size" can be used to save memory
               if norm_hidden_states.shape[self._chunk_dim] % self._chunk_size != 0:
                   raise ValueError(
                       F'''`hidden_states` dimension to be chunked: {norm_hidden_states.shape[self._chunk_dim]} has to be divisible by chunk size: {self._chunk_size}. Make sure to set an appropriate `chunk_size` when calling `unet.enable_forward_chunking`.''')

               _lowercase			:				Optional[Any] 						=						norm_hidden_states.shape[self._chunk_dim] // self._chunk_size
               _lowercase			:				Optional[Any] 						=						torch.cat(
                   [self.ff(lowerCamelCase) for hid_slice in norm_hidden_states.chunk(lowerCamelCase, dim=self._chunk_dim)], dim=self._chunk_dim, )
           else:
               _lowercase			:				Union[str, Any] 						=						self.ff(lowerCamelCase)

           if self.use_ada_layer_norm_zero:
               _lowercase			:				List[str] 						=						gate_mlp.unsqueeze(1) * ff_output

           _lowercase			:				Tuple 						=						ff_output + hidden_states

           return hidden_states






class _lowerCamelCase(		nn.Module  ):
       def __init__(   self, lowerCamelCase, lowerCamelCase = None, lowerCamelCase = 4, lowerCamelCase = 0.0, lowerCamelCase = "geglu", lowerCamelCase = False, )			->  Tuple:




           """simple docstring"""

           super().__init__()
           _lowercase			:				Optional[Any] 						=						int(dim * mult)
           _lowercase			:				Optional[Any] 						=						dim_out if dim_out is not None else dim

           if activation_fn == "gelu":
               _lowercase			:				Union[str, Any] 						=						GELU(lowerCamelCase, lowerCamelCase)
           if activation_fn == "gelu-approximate":
               _lowercase			:				Optional[Any] 						=						GELU(lowerCamelCase, lowerCamelCase, approximate='tanh')
           elif activation_fn == "geglu":
               _lowercase			:				str 						=						GEGLU(lowerCamelCase, lowerCamelCase)
           elif activation_fn == "geglu-approximate":
               _lowercase			:				Union[str, Any] 						=						ApproximateGELU(lowerCamelCase, lowerCamelCase)

           _lowercase			:				List[Any] 						=						nn.ModuleList([])
           # project in
           self.net.append(lowerCamelCase)
           # project dropout
           self.net.append(nn.Dropout(lowerCamelCase))
           # project out
           self.net.append(nn.Linear(lowerCamelCase, lowerCamelCase))
           # FF as used in Vision Transformer, MLP-Mixer, etc. have a final dropout
           if final_dropout:
               self.net.append(nn.Dropout(lowerCamelCase))




       def      UpperCamelCase	(   self, lowerCamelCase)			->  Optional[int]:




           """simple docstring"""

           for module in self.net:
               _lowercase			:				Union[str, Any] 						=						module(lowerCamelCase)
           return hidden_states






class _lowerCamelCase(		nn.Module  ):
       def __init__(   self, lowerCamelCase, lowerCamelCase, lowerCamelCase = "none")			->  Optional[int]:




           """simple docstring"""

           super().__init__()
           _lowercase			:				Union[str, Any] 						=						nn.Linear(lowerCamelCase, lowerCamelCase)
           _lowercase			:				List[Any] 						=						approximate
       def      UpperCamelCase	(   self, lowerCamelCase)			->  Tuple:




           """simple docstring"""

           if gate.device.type != "mps":
               return F.gelu(lowerCamelCase, approximate=self.approximate)
           # mps: gelu is not implemented for float16
           return F.gelu(gate.to(dtype=torch.floataa), approximate=self.approximate).to(dtype=gate.dtype)




       def      UpperCamelCase	(   self, lowerCamelCase)			->  Tuple:




           """simple docstring"""

           _lowercase			:				Optional[int] 						=						self.proj(lowerCamelCase)
           _lowercase			:				Union[str, Any] 						=						self.gelu(lowerCamelCase)
           return hidden_states






class _lowerCamelCase(		nn.Module  ):
       def __init__(   self, lowerCamelCase, lowerCamelCase)			->  int:




           """simple docstring"""

           super().__init__()
           _lowercase			:				Optional[Any] 						=						nn.Linear(lowerCamelCase, dim_out * 2)
       def      UpperCamelCase	(   self, lowerCamelCase)			->  Any:




           """simple docstring"""

           if gate.device.type != "mps":
               return F.gelu(lowerCamelCase)
           # mps: gelu is not implemented for float16
           return F.gelu(gate.to(dtype=torch.floataa)).to(dtype=gate.dtype)




       def      UpperCamelCase	(   self, lowerCamelCase)			->  Optional[int]:




           """simple docstring"""

           _lowercase						,							_lowercase			:				Tuple 						=						self.proj(lowerCamelCase).chunk(2, dim=-1)
           return hidden_states * self.gelu(lowerCamelCase)






class _lowerCamelCase(		nn.Module  ):
       def __init__(   self, lowerCamelCase, lowerCamelCase)			->  Any:




           """simple docstring"""

           super().__init__()
           _lowercase			:				str 						=						nn.Linear(lowerCamelCase, lowerCamelCase)




       def      UpperCamelCase	(   self, lowerCamelCase)			->  Union[str, Any]:




           """simple docstring"""

           _lowercase			:				Optional[int] 						=						self.proj(lowerCamelCase)
           return x * torch.sigmoid(1.7_0_2 * x)






class _lowerCamelCase(		nn.Module  ):
       def __init__(   self, lowerCamelCase, lowerCamelCase)			->  Tuple:




           """simple docstring"""

           super().__init__()
           _lowercase			:				int 						=						nn.Embedding(lowerCamelCase, lowerCamelCase)
           _lowercase			:				List[Any] 						=						nn.SiLU()
           _lowercase			:				Optional[Any] 						=						nn.Linear(lowerCamelCase, embedding_dim * 2)
           _lowercase			:				List[Any] 						=						nn.LayerNorm(lowerCamelCase, elementwise_affine=lowerCamelCase)




       def      UpperCamelCase	(   self, lowerCamelCase, lowerCamelCase)			->  Optional[Any]:




           """simple docstring"""

           _lowercase			:				List[str] 						=						self.linear(self.silu(self.emb(lowerCamelCase)))
           _lowercase						,							_lowercase			:				int 						=						torch.chunk(lowerCamelCase, 2)
           _lowercase			:				Optional[Any] 						=						self.norm(lowerCamelCase) * (1 + scale) + shift
           return x






class _lowerCamelCase(		nn.Module  ):
       def __init__(   self, lowerCamelCase, lowerCamelCase)			->  Union[str, Any]:




           """simple docstring"""

           super().__init__()

           _lowercase			:				List[Any] 						=						CombinedTimestepLabelEmbeddings(lowerCamelCase, lowerCamelCase)

           _lowercase			:				Tuple 						=						nn.SiLU()
           _lowercase			:				Any 						=						nn.Linear(lowerCamelCase, 6 * embedding_dim, bias=lowerCamelCase)
           _lowercase			:				List[str] 						=						nn.LayerNorm(lowerCamelCase, elementwise_affine=lowerCamelCase, eps=1E-6)




       def      UpperCamelCase	(   self, lowerCamelCase, lowerCamelCase, lowerCamelCase, lowerCamelCase=None)			->  Dict:




           """simple docstring"""

           _lowercase			:				Optional[int] 						=						self.linear(self.silu(self.emb(lowerCamelCase, lowerCamelCase, hidden_dtype=lowerCamelCase)))
           _lowercase						,							_lowercase						,							_lowercase						,							_lowercase						,							_lowercase						,							_lowercase			:				List[str] 						=						emb.chunk(6, dim=1)
           _lowercase			:				Dict 						=						self.norm(lowerCamelCase) * (1 + scale_msa[:, None]) + shift_msa[:, None]
           return x, gate_msa, shift_mlp, scale_mlp, gate_mlp






class _lowerCamelCase(		nn.Module  ):
       def __init__(   self, lowerCamelCase, lowerCamelCase, lowerCamelCase, lowerCamelCase = None, lowerCamelCase = 1E-5)			->  Union[str, Any]:




           """simple docstring"""

           super().__init__()
           _lowercase			:				Optional[Any] 						=						num_groups
           _lowercase			:				Any 						=						eps

           if act_fn is None:
               _lowercase			:				Optional[Any] 						=						None
           else:
               _lowercase			:				Any 						=						get_activation(lowerCamelCase)

           _lowercase			:				int 						=						nn.Linear(lowerCamelCase, out_dim * 2)




       def      UpperCamelCase	(   self, lowerCamelCase, lowerCamelCase)			->  Optional[int]:




           """simple docstring"""

           if self.act:
               _lowercase			:				Optional[int] 						=						self.act(lowerCamelCase)
           _lowercase			:				Union[str, Any] 						=						self.linear(lowerCamelCase)
           _lowercase			:				Optional[Any] 						=						emb[:, :, None, None]
           _lowercase						,							_lowercase			:				Optional[Any] 						=						emb.chunk(2, dim=1)

           _lowercase			:				Any 						=						F.group_norm(lowerCamelCase, self.num_groups, eps=self.eps)
           _lowercase			:				Optional[Any] 						=						x * (1 + scale) + shift
           return x





"""simple docstring"""




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


A_				:      List[str]           =			pytest.mark.integration





@require_faiss
class  lowerCamelCase    (A__      ):

			def        SCREAMING_SNAKE_CASE					(       self					:    Dict							) ->     Tuple:
						SCREAMING_SNAKE_CASE__										=   Dataset.from_dict({"""filename""": ["""my_name-train""" + """_""" + str(__UpperCAmelCase							) for x in np.arange(3_0							).tolist()]}							)
						return dset

			def        SCREAMING_SNAKE_CASE					(       self					:    Dict							) ->     Tuple:
						import faiss

						SCREAMING_SNAKE_CASE__										=   self._create_dummy_dataset()
						SCREAMING_SNAKE_CASE__										=   dset.map(
						    lambda __UpperCAmelCase     ,					__UpperCAmelCase							: {"vecs": i * np.ones(5     ,					dtype=np.floataa							)}     ,					with_indices=__UpperCAmelCase     ,					keep_in_memory=__UpperCAmelCase							)
						SCREAMING_SNAKE_CASE__										=   dset.add_faiss_index("""vecs"""     ,					batch_size=1_0_0     ,					metric_type=faiss.METRIC_INNER_PRODUCT							)
						SCREAMING_SNAKE_CASE__				,		SCREAMING_SNAKE_CASE__										=   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        SCREAMING_SNAKE_CASE					(       self					:    List[str]							) ->     str:
						import faiss

						SCREAMING_SNAKE_CASE__										=   self._create_dummy_dataset()
						dset.add_faiss_index_from_external_arrays(
						    external_arrays=np.ones((3_0, 5)							) * np.arange(3_0							).reshape(-1     ,					1							)     ,					index_name="""vecs"""     ,					batch_size=1_0_0     ,					metric_type=faiss.METRIC_INNER_PRODUCT     ,					)
						SCREAMING_SNAKE_CASE__				,		SCREAMING_SNAKE_CASE__										=   dset.get_nearest_examples("""vecs"""     ,					np.ones(5     ,					dtype=np.floataa							)							)
						self.assertEqual(examples["""filename"""][0]     ,					"""my_name-train_29"""							)

			def        SCREAMING_SNAKE_CASE					(       self					:    List[Any]							) ->     Any:
						import faiss

						SCREAMING_SNAKE_CASE__										=   self._create_dummy_dataset()
						dset.add_faiss_index_from_external_arrays(
						    external_arrays=np.ones((3_0, 5)							) * np.arange(3_0							).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=__UpperCAmelCase							) as tmp_file:
									dset.save_faiss_index("""vecs"""     ,					tmp_file.name							)
									dset.load_faiss_index("""vecs2"""     ,					tmp_file.name							)
						os.unlink(tmp_file.name							)

						SCREAMING_SNAKE_CASE__				,		SCREAMING_SNAKE_CASE__										=   dset.get_nearest_examples("""vecs2"""     ,					np.ones(5     ,					dtype=np.floataa							)							)
						self.assertEqual(examples["""filename"""][0]     ,					"""my_name-train_29"""							)

			def        SCREAMING_SNAKE_CASE					(       self					:    Optional[int]							) ->     int:
						SCREAMING_SNAKE_CASE__										=   self._create_dummy_dataset()
						dset.add_faiss_index_from_external_arrays(
						    external_arrays=np.ones((3_0, 5)							) * np.arange(3_0							).reshape(-1     ,					1							)     ,					index_name="""vecs"""							)
						dset.drop_index("""vecs"""							)
						self.assertRaises(__UpperCAmelCase     ,					partial(dset.get_nearest_examples     ,					"""vecs2"""     ,					np.ones(5     ,					dtype=np.floataa							)							)							)





			def        SCREAMING_SNAKE_CASE					(       self					:    List[str]							) ->     str:
						from elasticsearch import Elasticsearch

						SCREAMING_SNAKE_CASE__										=   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:
									SCREAMING_SNAKE_CASE__										=   {"""acknowledged""": True}
									mocked_bulk.return_value([(True, None)] * 3_0							)
									SCREAMING_SNAKE_CASE__										=   {"""hits""": {"""hits""": [{"""_score""": 1, """_id""": 2_9}]}}
									SCREAMING_SNAKE_CASE__										=   Elasticsearch()

									dset.add_elasticsearch_index("""filename"""     ,					es_client=__UpperCAmelCase							)
									SCREAMING_SNAKE_CASE__				,		SCREAMING_SNAKE_CASE__										=   dset.get_nearest_examples("""filename"""     ,					"""my_name-train_29"""							)
									self.assertEqual(examples["""filename"""][0]     ,					"""my_name-train_29"""							)





@require_faiss
class  lowerCamelCase    (A__      ):

			def        SCREAMING_SNAKE_CASE					(       self					:    int							) ->     List[str]:
						import faiss

						SCREAMING_SNAKE_CASE__										=   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     ,					1_0							)

						# single query
						SCREAMING_SNAKE_CASE__										=   np.zeros(5     ,					dtype=np.floataa							)
						SCREAMING_SNAKE_CASE__										=   1
						SCREAMING_SNAKE_CASE__				,		SCREAMING_SNAKE_CASE__										=   index.search(__UpperCAmelCase							)
						self.assertRaises(__UpperCAmelCase     ,					index.search     ,					query.reshape(-1     ,					1							)							)
						self.assertGreater(scores[0]     ,					0							)
						self.assertEqual(indices[0]     ,					1							)

						# batched queries
						SCREAMING_SNAKE_CASE__										=   np.eye(5     ,					dtype=np.floataa							)[::-1]
						SCREAMING_SNAKE_CASE__				,		SCREAMING_SNAKE_CASE__										=   index.search_batch(__UpperCAmelCase							)
						self.assertRaises(__UpperCAmelCase     ,					index.search_batch     ,					queries[0]							)
						SCREAMING_SNAKE_CASE__										=   [scores[0] for scores in total_scores]
						SCREAMING_SNAKE_CASE__										=   [indices[0] for indices in total_indices]
						self.assertGreater(np.min(__UpperCAmelCase							)     ,					0							)
						self.assertListEqual([4, 3, 2, 1, 0]     ,					__UpperCAmelCase							)

			def        SCREAMING_SNAKE_CASE					(       self					:    List[str]							) ->     str:
						import faiss

						SCREAMING_SNAKE_CASE__										=   FaissIndex(string_factory="""Flat"""							)
						index.add_vectors(np.eye(5     ,					dtype=np.floataa							)							)
						self.assertIsInstance(index.faiss_index     ,					faiss.IndexFlat							)
						SCREAMING_SNAKE_CASE__										=   FaissIndex(string_factory="""LSH"""							)
						index.add_vectors(np.eye(5     ,					dtype=np.floataa							)							)
						self.assertIsInstance(index.faiss_index     ,					faiss.IndexLSH							)
						with self.assertRaises(__UpperCAmelCase							):
									SCREAMING_SNAKE_CASE__										=   FaissIndex(string_factory="""Flat"""     ,					custom_index=faiss.IndexFlat(5							)							)

			def        SCREAMING_SNAKE_CASE					(       self					:    Dict							) ->     Any:
						import faiss

						SCREAMING_SNAKE_CASE__										=   faiss.IndexFlat(5							)
						SCREAMING_SNAKE_CASE__										=   FaissIndex(custom_index=__UpperCAmelCase							)
						index.add_vectors(np.eye(5     ,					dtype=np.floataa							)							)
						self.assertIsInstance(index.faiss_index     ,					faiss.IndexFlat							)





			def        SCREAMING_SNAKE_CASE					(       self					:    Union[str, Any]							) ->     str:
						import faiss

						SCREAMING_SNAKE_CASE__										=   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=__UpperCAmelCase							) as tmp_file:
									index.save(tmp_file.name							)
									SCREAMING_SNAKE_CASE__										=   FaissIndex.load(tmp_file.name							)
						os.unlink(tmp_file.name							)

						SCREAMING_SNAKE_CASE__										=   np.zeros(5     ,					dtype=np.floataa							)
						SCREAMING_SNAKE_CASE__										=   1
						SCREAMING_SNAKE_CASE__				,		SCREAMING_SNAKE_CASE__										=   index.search(__UpperCAmelCase							)
						self.assertGreater(scores[0]     ,					0							)
						self.assertEqual(indices[0]     ,					1							)

@require_faiss
def 						A							(  snake_case__     ):




			'''simple docstring'''


			import faiss

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

			SCREAMING_SNAKE_CASE__										=   """index.faiss"""
			SCREAMING_SNAKE_CASE__										=   f"""mock://{index_name}"""
			index.save(snake_case__		,					storage_options=mockfs.storage_options     )
			SCREAMING_SNAKE_CASE__										=   FaissIndex.load(snake_case__		,					storage_options=mockfs.storage_options     )

			SCREAMING_SNAKE_CASE__										=   np.zeros(5		,					dtype=np.floataa     )
			SCREAMING_SNAKE_CASE__										=   1
			SCREAMING_SNAKE_CASE__				,		SCREAMING_SNAKE_CASE__										=   index.search(snake_case__     )
			assert scores[0] > 0
			assert indices[0] == 1





@require_elasticsearch
class  lowerCamelCase    (A__      ):

			def        SCREAMING_SNAKE_CASE					(       self					:    Tuple							) ->     Dict:
						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:
									SCREAMING_SNAKE_CASE__										=   Elasticsearch()
									SCREAMING_SNAKE_CASE__										=   {"""acknowledged""": True}
									SCREAMING_SNAKE_CASE__										=   ElasticSearchIndex(es_client=__UpperCAmelCase							)
									mocked_bulk.return_value([(True, None)] * 3							)
									index.add_documents(["""foo""", """bar""", """foobar"""]							)

									# single query
									SCREAMING_SNAKE_CASE__										=   """foo"""
									SCREAMING_SNAKE_CASE__										=   {"""hits""": {"""hits""": [{"""_score""": 1, """_id""": 0}]}}
									SCREAMING_SNAKE_CASE__				,		SCREAMING_SNAKE_CASE__										=   index.search(__UpperCAmelCase							)
									self.assertEqual(scores[0]     ,					1							)
									self.assertEqual(indices[0]     ,					0							)

									# single query with timeout
									SCREAMING_SNAKE_CASE__										=   """foo"""
									SCREAMING_SNAKE_CASE__										=   {"""hits""": {"""hits""": [{"""_score""": 1, """_id""": 0}]}}
									SCREAMING_SNAKE_CASE__				,		SCREAMING_SNAKE_CASE__										=   index.search(__UpperCAmelCase     ,					request_timeout=3_0							)
									self.assertEqual(scores[0]     ,					1							)
									self.assertEqual(indices[0]     ,					0							)

									# batched queries
									SCREAMING_SNAKE_CASE__										=   ["""foo""", """bar""", """foobar"""]
									SCREAMING_SNAKE_CASE__										=   {"""hits""": {"""hits""": [{"""_score""": 1, """_id""": 1}]}}
									SCREAMING_SNAKE_CASE__				,		SCREAMING_SNAKE_CASE__										=   index.search_batch(__UpperCAmelCase							)
									SCREAMING_SNAKE_CASE__										=   [scores[0] for scores in total_scores]
									SCREAMING_SNAKE_CASE__										=   [indices[0] for indices in total_indices]
									self.assertGreater(np.min(__UpperCAmelCase							)     ,					0							)
									self.assertListEqual([1, 1, 1]     ,					__UpperCAmelCase							)

									# batched queries with timeout
									SCREAMING_SNAKE_CASE__										=   ["""foo""", """bar""", """foobar"""]
									SCREAMING_SNAKE_CASE__										=   {"""hits""": {"""hits""": [{"""_score""": 1, """_id""": 1}]}}
									SCREAMING_SNAKE_CASE__				,		SCREAMING_SNAKE_CASE__										=   index.search_batch(__UpperCAmelCase     ,					request_timeout=3_0							)
									SCREAMING_SNAKE_CASE__										=   [scores[0] for scores in total_scores]
									SCREAMING_SNAKE_CASE__										=   [indices[0] for indices in total_indices]
									self.assertGreater(np.min(__UpperCAmelCase							)     ,					0							)
									self.assertListEqual([1, 1, 1]     ,					__UpperCAmelCase							)


