


import logging
import os
import random
import sys
from dataclasses import dataclass, field
from typing import Optional

import datasets
import evaluate
import numpy as np
from datasets import load_dataset

import transformers
from transformers import (
    AutoConfig,
    AutoModelForSequenceClassification,
    AutoTokenizer,
    DataCollatorWithPadding,
    EvalPrediction,
    HfArgumentParser,
    Trainer,
    TrainingArguments,
    default_data_collator,
    set_seed,
)
from transformers.trainer_utils import get_last_checkpoint
from transformers.utils import check_min_version, send_example_telemetry
from transformers.utils.versions import require_version


# Will error if the minimal version of Transformers is not installed. Remove at your own risks.
check_min_version('4.31.0')

require_version('datasets>=1.8.0', 'To fix: pip install -r examples/pytorch/text-classification/requirements.txt')

__lowerCAmelCase	: List[Any]  				=logging.getLogger(__name__)
@dataclass
class        _lowercase      :






				'''simple docstring'''



				SCREAMING_SNAKE_CASE__				:			Optional[int]						=		field(
				    default=128						,			metadata={
				        '''help''': (
				            '''The maximum total input sequence length after tokenization. Sequences longer '''
				            '''than this will be truncated, sequences shorter will be padded.'''
				        )
				    }						,			)
				SCREAMING_SNAKE_CASE__				:			bool						=		field(
				    default=A__						,			metadata={'''help''': '''Overwrite the cached preprocessed datasets or not.'''}       )
				SCREAMING_SNAKE_CASE__				:			bool						=		field(
				    default=A__						,			metadata={
				        '''help''': (
				            '''Whether to pad all samples to `max_seq_length`. '''
				            '''If False, will pad the samples dynamically when batching to the maximum length in the batch.'''
				        )
				    }						,			)
				SCREAMING_SNAKE_CASE__				:			Optional[int]						=		field(
				    default=A__						,			metadata={
				        '''help''': (
				            '''For debugging purposes or quicker training, truncate the number of training examples to this '''
				            '''value if set.'''
				        )
				    }						,			)
				SCREAMING_SNAKE_CASE__				:			Optional[int]						=		field(
				    default=A__						,			metadata={
				        '''help''': (
				            '''For debugging purposes or quicker training, truncate the number of evaluation examples to this '''
				            '''value if set.'''
				        )
				    }						,			)
				SCREAMING_SNAKE_CASE__				:			Optional[int]						=		field(
				    default=A__						,			metadata={
				        '''help''': (
				            '''For debugging purposes or quicker training, truncate the number of prediction examples to this '''
				            '''value if set.'''
				        )
				    }						,			)




@dataclass
class        _lowercase      :






				'''simple docstring'''



				SCREAMING_SNAKE_CASE__				:			str						=		field(
				    default=A__						,			metadata={'''help''': '''Path to pretrained model or model identifier from huggingface.co/models'''}       )
				SCREAMING_SNAKE_CASE__				:			str						=		field(
				    default=A__						,			metadata={'''help''': '''Evaluation language. Also train language if `train_language` is set to None.'''}       )
				SCREAMING_SNAKE_CASE__				:			Optional[str]						=		field(
				    default=A__						,			metadata={'''help''': '''Train language if it is different from the evaluation language.'''}       )
				SCREAMING_SNAKE_CASE__				:			Optional[str]						=		field(
				    default=A__						,			metadata={'''help''': '''Pretrained config name or path if not the same as model_name'''}       )
				SCREAMING_SNAKE_CASE__				:			Optional[str]						=		field(
				    default=A__						,			metadata={'''help''': '''Pretrained tokenizer name or path if not the same as model_name'''}       )
				SCREAMING_SNAKE_CASE__				:			Optional[str]						=		field(
				    default=A__						,			metadata={'''help''': '''Where do you want to store the pretrained models downloaded from huggingface.co'''}						,			)
				SCREAMING_SNAKE_CASE__				:			Optional[bool]						=		field(
				    default=A__						,			metadata={'''help''': '''arg to indicate if tokenizer should do lower case in AutoTokenizer.from_pretrained()'''}						,			)
				SCREAMING_SNAKE_CASE__				:			bool						=		field(
				    default=A__						,			metadata={'''help''': '''Whether to use one of the fast tokenizer (backed by the tokenizers library) or not.'''}						,			)
				SCREAMING_SNAKE_CASE__				:			str						=		field(
				    default='''main'''						,			metadata={'''help''': '''The specific model version to use (can be a branch name, tag name or commit id).'''}						,			)
				SCREAMING_SNAKE_CASE__				:			bool						=		field(
				    default=A__						,			metadata={
				        '''help''': (
				            '''Will use the token generated when running `huggingface-cli login` (necessary to use this script '''
				            '''with private models).'''
				        )
				    }						,			)
				SCREAMING_SNAKE_CASE__				:			bool						=		field(
				    default=A__						,			metadata={'''help''': '''Will enable to load a pretrained model whose head dimensions are different.'''}						,			)




def        _UpperCamelCase							(      ):
						__SCREAMING_SNAKE_CASE						:     List[Any]					=		HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)      )
						__SCREAMING_SNAKE_CASE						:     Dict					=		parser.parse_args_into_dataclasses()

						# Sending telemetry. Tracking the example usage helps us better allocate resources to maintain them. The
						# information sent is the one passed as arguments along with your Python/PyTorch versions.
						send_example_telemetry('''run_xnli'''    ,       lowercase__      )

						# Setup logging
						logging.basicConfig(
						    format='''%(asctime)s - %(levelname)s - %(name)s - %(message)s'''    ,       datefmt='''%m/%d/%Y %H:%M:%S'''    ,       handlers=[logging.StreamHandler(sys.stdout      )]    ,       )

						if training_args.should_log:
												# The default of training_args.log_level is passive, so we set log level at info here to have that default.
												transformers.utils.logging.set_verbosity_info()

						__SCREAMING_SNAKE_CASE						:     Union[str, Any]					=		training_args.get_process_log_level()
						logger.setLevel(lowercase__      )
						datasets.utils.logging.set_verbosity(lowercase__      )
						transformers.utils.logging.set_verbosity(lowercase__      )
						transformers.utils.logging.enable_default_handler()
						transformers.utils.logging.enable_explicit_format()

						# Log on each process the small summary:
						logger.warning(
						    F'''Process rank: {training_args.local_rank}, device: {training_args.device}, n_gpu: {training_args.n_gpu}'''
						    + F'''distributed training: {bool(training_args.local_rank != -1      )}, 16-bits training: {training_args.fpaa}'''      )
						logger.info(F'''Training/evaluation parameters {training_args}'''      )

						# Detecting last checkpoint.
						__SCREAMING_SNAKE_CASE						:     Tuple					=		None
						if os.path.isdir(training_args.output_dir      ) and training_args.do_train and not training_args.overwrite_output_dir:
												__SCREAMING_SNAKE_CASE						:     Dict					=		get_last_checkpoint(training_args.output_dir      )
												if last_checkpoint is None and len(os.listdir(training_args.output_dir      )      ) > 0:
																		raise ValueError(
																		    F'''Output directory ({training_args.output_dir}) already exists and is not empty. '''
																		    '''Use --overwrite_output_dir to overcome.'''      )
												elif last_checkpoint is not None:
																		logger.info(
																		    F'''Checkpoint detected, resuming training at {last_checkpoint}. To avoid this behavior, change '''
																		    '''the `--output_dir` or add `--overwrite_output_dir` to train from scratch.'''      )

												# Set seed before initializing model.
						set_seed(training_args.seed      )

						# In distributed training, the load_dataset function guarantees that only one local process can concurrently
						# download the dataset.
						# Downloading and loading xnli dataset from the hub.
						if training_args.do_train:
												if model_args.train_language is None:
																		__SCREAMING_SNAKE_CASE						:     Union[str, Any]					=		load_dataset(
																		    '''xnli'''    ,       model_args.language    ,       split='''train'''    ,       cache_dir=model_args.cache_dir    ,       use_auth_token=True if model_args.use_auth_token else None    ,       )
												else:
																		__SCREAMING_SNAKE_CASE						:     List[Any]					=		load_dataset(
																		    '''xnli'''    ,       model_args.train_language    ,       split='''train'''    ,       cache_dir=model_args.cache_dir    ,       use_auth_token=True if model_args.use_auth_token else None    ,       )
												__SCREAMING_SNAKE_CASE						:     int					=		train_dataset.features["label"].names

						if training_args.do_eval:
												__SCREAMING_SNAKE_CASE						:     List[Any]					=		load_dataset(
												    '''xnli'''    ,       model_args.language    ,       split='''validation'''    ,       cache_dir=model_args.cache_dir    ,       use_auth_token=True if model_args.use_auth_token else None    ,       )
												__SCREAMING_SNAKE_CASE						:     Dict					=		eval_dataset.features["label"].names

						if training_args.do_predict:
												__SCREAMING_SNAKE_CASE						:     List[str]					=		load_dataset(
												    '''xnli'''    ,       model_args.language    ,       split='''test'''    ,       cache_dir=model_args.cache_dir    ,       use_auth_token=True if model_args.use_auth_token else None    ,       )
												__SCREAMING_SNAKE_CASE						:     List[Any]					=		predict_dataset.features["label"].names

						# Labels
						__SCREAMING_SNAKE_CASE						:     Optional[int]					=		len(lowercase__      )

						# Load pretrained model and tokenizer
						# In distributed training, the .from_pretrained methods guarantee that only one local process can concurrently
						# download model & vocab.
						__SCREAMING_SNAKE_CASE						:     Any					=		AutoConfig.from_pretrained(
						    model_args.config_name if model_args.config_name else model_args.model_name_or_path    ,       num_labels=lowercase__    ,       idalabel={str(lowercase__      ): label for i, label in enumerate(lowercase__      )}    ,       labelaid={label: i for i, label in enumerate(lowercase__      )}    ,       finetuning_task='''xnli'''    ,       cache_dir=model_args.cache_dir    ,       revision=model_args.model_revision    ,       use_auth_token=True if model_args.use_auth_token else None    ,       )
						__SCREAMING_SNAKE_CASE						:     Optional[Any]					=		AutoTokenizer.from_pretrained(
						    model_args.tokenizer_name if model_args.tokenizer_name else model_args.model_name_or_path    ,       do_lower_case=model_args.do_lower_case    ,       cache_dir=model_args.cache_dir    ,       use_fast=model_args.use_fast_tokenizer    ,       revision=model_args.model_revision    ,       use_auth_token=True if model_args.use_auth_token else None    ,       )
						__SCREAMING_SNAKE_CASE						:     Optional[Any]					=		AutoModelForSequenceClassification.from_pretrained(
						    model_args.model_name_or_path    ,       from_tf=bool('''.ckpt''' in model_args.model_name_or_path      )    ,       config=lowercase__    ,       cache_dir=model_args.cache_dir    ,       revision=model_args.model_revision    ,       use_auth_token=True if model_args.use_auth_token else None    ,       ignore_mismatched_sizes=model_args.ignore_mismatched_sizes    ,       )

						# Preprocessing the datasets
						# Padding strategy
						if data_args.pad_to_max_length:
												__SCREAMING_SNAKE_CASE						:     Optional[int]					=		"max_length"
						else:
												# We will pad later, dynamically at batch creation, to the max sequence length in each batch
												__SCREAMING_SNAKE_CASE						:     Union[str, Any]					=		False

						def preprocess_function(lowercase__      ):
												# Tokenize the texts
												return tokenizer(
												    examples['''premise''']    ,       examples['''hypothesis''']    ,       padding=lowercase__    ,       max_length=data_args.max_seq_length    ,       truncation=lowercase__    ,       )

						if training_args.do_train:
												if data_args.max_train_samples is not None:
																		__SCREAMING_SNAKE_CASE						:     str					=		min(len(lowercase__      )    ,       data_args.max_train_samples      )
																		__SCREAMING_SNAKE_CASE						:     Union[str, Any]					=		train_dataset.select(range(lowercase__      )      )
												with training_args.main_process_first(desc='''train dataset map pre-processing'''      ):
																		__SCREAMING_SNAKE_CASE						:     Dict					=		train_dataset.map(
																		    lowercase__    ,       batched=lowercase__    ,       load_from_cache_file=not data_args.overwrite_cache    ,       desc='''Running tokenizer on train dataset'''    ,       )
												# Log a few random samples from the training set:
												for index in random.sample(range(len(lowercase__      )      )    ,       3      ):
																		logger.info(F'''Sample {index} of the training set: {train_dataset[index]}.'''      )

						if training_args.do_eval:
												if data_args.max_eval_samples is not None:
																		__SCREAMING_SNAKE_CASE						:     Dict					=		min(len(lowercase__      )    ,       data_args.max_eval_samples      )
																		__SCREAMING_SNAKE_CASE						:     int					=		eval_dataset.select(range(lowercase__      )      )
												with training_args.main_process_first(desc='''validation dataset map pre-processing'''      ):
																		__SCREAMING_SNAKE_CASE						:     List[Any]					=		eval_dataset.map(
																		    lowercase__    ,       batched=lowercase__    ,       load_from_cache_file=not data_args.overwrite_cache    ,       desc='''Running tokenizer on validation dataset'''    ,       )

						if training_args.do_predict:
												if data_args.max_predict_samples is not None:
																		__SCREAMING_SNAKE_CASE						:     Any					=		min(len(lowercase__      )    ,       data_args.max_predict_samples      )
																		__SCREAMING_SNAKE_CASE						:     List[Any]					=		predict_dataset.select(range(lowercase__      )      )
												with training_args.main_process_first(desc='''prediction dataset map pre-processing'''      ):
																		__SCREAMING_SNAKE_CASE						:     Optional[int]					=		predict_dataset.map(
																		    lowercase__    ,       batched=lowercase__    ,       load_from_cache_file=not data_args.overwrite_cache    ,       desc='''Running tokenizer on prediction dataset'''    ,       )

												# Get the metric function
						__SCREAMING_SNAKE_CASE						:     List[str]					=		evaluate.load('''xnli'''      )

						# You can define your custom compute_metrics function. It takes an `EvalPrediction` object (a namedtuple with a
						# predictions and label_ids field) and has to return a dictionary string to float.
						def compute_metrics(lowercase__      ):
												__SCREAMING_SNAKE_CASE						:     List[str]					=		p.predictions[0] if isinstance(p.predictions    ,       lowercase__      ) else p.predictions
												__SCREAMING_SNAKE_CASE						:     Dict					=		np.argmax(lowercase__    ,       axis=1      )
												return metric.compute(predictions=lowercase__    ,       references=p.label_ids      )

						# Data collator will default to DataCollatorWithPadding, so we change it if we already did the padding.
						if data_args.pad_to_max_length:
												__SCREAMING_SNAKE_CASE						:     Optional[Any]					=		default_data_collator
						elif training_args.fpaa:
												__SCREAMING_SNAKE_CASE						:     Tuple					=		DataCollatorWithPadding(lowercase__    ,       pad_to_multiple_of=8      )
						else:
												__SCREAMING_SNAKE_CASE						:     List[str]					=		None

						# Initialize our Trainer
						__SCREAMING_SNAKE_CASE						:     Optional[int]					=		Trainer(
						    model=lowercase__    ,       args=lowercase__    ,       train_dataset=train_dataset if training_args.do_train else None    ,       eval_dataset=eval_dataset if training_args.do_eval else None    ,       compute_metrics=lowercase__    ,       tokenizer=lowercase__    ,       data_collator=lowercase__    ,       )

						# Training
						if training_args.do_train:
												__SCREAMING_SNAKE_CASE						:     Optional[int]					=		None
												if training_args.resume_from_checkpoint is not None:
																		__SCREAMING_SNAKE_CASE						:     Any					=		training_args.resume_from_checkpoint
												elif last_checkpoint is not None:
																		__SCREAMING_SNAKE_CASE						:     Any					=		last_checkpoint
												__SCREAMING_SNAKE_CASE						:     Optional[Any]					=		trainer.train(resume_from_checkpoint=lowercase__      )
												__SCREAMING_SNAKE_CASE						:     Optional[int]					=		train_result.metrics
												__SCREAMING_SNAKE_CASE						:     List[str]					=		(
												    data_args.max_train_samples if data_args.max_train_samples is not None else len(lowercase__      )
												)
												__SCREAMING_SNAKE_CASE						:     Tuple					=		min(lowercase__    ,       len(lowercase__      )      )

												trainer.save_model()  # Saves the tokenizer too for easy upload

												trainer.log_metrics('''train'''    ,       lowercase__      )
												trainer.save_metrics('''train'''    ,       lowercase__      )
												trainer.save_state()

						# Evaluation
						if training_args.do_eval:
												logger.info('''*** Evaluate ***'''      )
												__SCREAMING_SNAKE_CASE						:     Optional[Any]					=		trainer.evaluate(eval_dataset=lowercase__      )

												__SCREAMING_SNAKE_CASE						:     Any					=		data_args.max_eval_samples if data_args.max_eval_samples is not None else len(lowercase__      )
												__SCREAMING_SNAKE_CASE						:     List[Any]					=		min(lowercase__    ,       len(lowercase__      )      )

												trainer.log_metrics('''eval'''    ,       lowercase__      )
												trainer.save_metrics('''eval'''    ,       lowercase__      )

						# Prediction
						if training_args.do_predict:
												logger.info('''*** Predict ***'''      )
												__SCREAMING_SNAKE_CASE						:     str					=		trainer.predict(lowercase__    ,       metric_key_prefix='''predict'''      )

												__SCREAMING_SNAKE_CASE						:     Optional[int]					=		(
												    data_args.max_predict_samples if data_args.max_predict_samples is not None else len(lowercase__      )
												)
												__SCREAMING_SNAKE_CASE						:     List[Any]					=		min(lowercase__    ,       len(lowercase__      )      )

												trainer.log_metrics('''predict'''    ,       lowercase__      )
												trainer.save_metrics('''predict'''    ,       lowercase__      )

												__SCREAMING_SNAKE_CASE						:     Optional[Any]					=		np.argmax(lowercase__    ,       axis=1      )
												__SCREAMING_SNAKE_CASE						:     Any					=		os.path.join(training_args.output_dir    ,       '''predictions.txt'''      )
												if trainer.is_world_process_zero():
																		with open(lowercase__    ,       '''w'''      ) as writer:
																								writer.write('''index\tprediction\n'''      )
																								for index, item in enumerate(lowercase__      ):
																														__SCREAMING_SNAKE_CASE						:     Optional[Any]					=		label_list[item]
																														writer.write(F'''{index}\t{item}\n'''      )


if __name__ == "__main__":
						main()




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


lowerCAmelCase__      				=   pytest.mark.integration


@require_faiss
class       SCREAMING_SNAKE_CASE__				(		lowercase				):
       """simple docstring"""


       def        lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase     :     List[str]								=							Dataset.from_dict({"filename": ["my_name-train" + "_" + str(snake_case__						) for x in np.arange(30						).tolist()]}						)
         return dset

       def        lowercase__							(						self						):






         """simple docstring"""





         import faiss

         lowerCAmelCase     :     Dataset								=							self._create_dummy_dataset()
         lowerCAmelCase     :     Union[str, Any]								=							dset.map(
             lambda snake_case__		,     snake_case__						: {"vecs": i * np.ones(5		,     dtype=np.floataa						)}		,     with_indices=snake_case__		,     keep_in_memory=snake_case__						)
         lowerCAmelCase     :     Union[str, Any]								=							dset.add_faiss_index("vecs"		,     batch_size=100		,     metric_type=faiss.METRIC_INNER_PRODUCT						)
         lowerCAmelCase						,  lowerCAmelCase     :     Tuple								=							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        lowercase__							(						self						):






         """simple docstring"""





         import faiss

         lowerCAmelCase     :     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		,     )
         lowerCAmelCase						,  lowerCAmelCase     :     Optional[Any]								=							dset.get_nearest_examples("vecs"		,     np.ones(5		,     dtype=np.floataa						)						)
         self.assertEqual(examples["filename"][0]		,     "my_name-train_29"						)

       def        lowercase__							(						self						):






         """simple docstring"""





         import faiss

         lowerCAmelCase     :     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=snake_case__						) as tmp_file:
           dset.save_faiss_index("vecs"		,     tmp_file.name						)
           dset.load_faiss_index("vecs2"		,     tmp_file.name						)
         os.unlink(tmp_file.name						)

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

       def        lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase     :     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(snake_case__		,     partial(dset.get_nearest_examples		,     "vecs2"		,     np.ones(5		,     dtype=np.floataa						)						)						)







       def        lowercase__							(						self						):






         """simple docstring"""





         from elasticsearch import Elasticsearch

         lowerCAmelCase     :     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:
           lowerCAmelCase     :     List[str]								=							{"acknowledged": True}
           mocked_bulk.return_value([(True, None)] * 30						)
           lowerCAmelCase     :     List[str]								=							{"hits": {"hits": [{"_score": 1, "_id": 29}]}}
           lowerCAmelCase     :     str								=							Elasticsearch()

           dset.add_elasticsearch_index("filename"		,     es_client=snake_case__						)
           lowerCAmelCase						,  lowerCAmelCase     :     int								=							dset.get_nearest_examples("filename"		,     "my_name-train_29"						)
           self.assertEqual(examples["filename"][0]		,     "my_name-train_29"						)


@require_faiss
class       SCREAMING_SNAKE_CASE__				(		lowercase				):
       """simple docstring"""


       def        lowercase__							(						self						):






         """simple docstring"""





         import faiss

         lowerCAmelCase     :     Optional[int]								=							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
         lowerCAmelCase     :     int								=							np.zeros(5		,     dtype=np.floataa						)
         lowerCAmelCase     :     Optional[int]								=							1
         lowerCAmelCase						,  lowerCAmelCase     :     Optional[Any]								=							index.search(snake_case__						)
         self.assertRaises(snake_case__		,     index.search		,     query.reshape(-1		,     1						)						)
         self.assertGreater(scores[0]		,     0						)
         self.assertEqual(indices[0]		,     1						)

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

       def        lowercase__							(						self						):






         """simple docstring"""





         import faiss

         lowerCAmelCase     :     Dict								=							FaissIndex(string_factory="Flat"						)
         index.add_vectors(np.eye(5		,     dtype=np.floataa						)						)
         self.assertIsInstance(index.faiss_index		,     faiss.IndexFlat						)
         lowerCAmelCase     :     Union[str, Any]								=							FaissIndex(string_factory="LSH"						)
         index.add_vectors(np.eye(5		,     dtype=np.floataa						)						)
         self.assertIsInstance(index.faiss_index		,     faiss.IndexLSH						)
         with self.assertRaises(snake_case__						):
           lowerCAmelCase     :     List[Any]								=							FaissIndex(string_factory="Flat"		,     custom_index=faiss.IndexFlat(5						)						)

       def        lowercase__							(						self						):






         """simple docstring"""





         import faiss

         lowerCAmelCase     :     Any								=							faiss.IndexFlat(5						)
         lowerCAmelCase     :     Union[str, Any]								=							FaissIndex(custom_index=snake_case__						)
         index.add_vectors(np.eye(5		,     dtype=np.floataa						)						)
         self.assertIsInstance(index.faiss_index		,     faiss.IndexFlat						)







       def        lowercase__							(						self						):






         """simple docstring"""





         import faiss

         lowerCAmelCase     :     str								=							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=snake_case__						) as tmp_file:
           index.save(tmp_file.name						)
           lowerCAmelCase     :     int								=							FaissIndex.load(tmp_file.name						)
         os.unlink(tmp_file.name						)

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

@require_faiss
def 					a__			(				SCREAMING_SNAKE_CASE    :    Any				):


  '''simple docstring'''







  import faiss

  lowerCAmelCase     :     List[Any]								=							FaissIndex(metric_type=faiss.METRIC_INNER_PRODUCT				)
  index.add_vectors(np.eye(5				,      dtype=np.floataa				)				)

  lowerCAmelCase     :     Union[str, Any]								=							"index.faiss"
  lowerCAmelCase     :     List[str]								=							f"""mock://{index_name}"""
  index.save(SCREAMING_SNAKE_CASE				,      storage_options=mockfs.storage_options				)
  lowerCAmelCase     :     Optional[Any]								=							FaissIndex.load(SCREAMING_SNAKE_CASE				,      storage_options=mockfs.storage_options				)

  lowerCAmelCase     :     Optional[int]								=							np.zeros(5				,      dtype=np.floataa				)
  lowerCAmelCase     :     Any								=							1
  lowerCAmelCase						,  lowerCAmelCase     :     Optional[int]								=							index.search(SCREAMING_SNAKE_CASE				)
  assert scores[0] > 0
  assert indices[0] == 1





@require_elasticsearch
class       SCREAMING_SNAKE_CASE__				(		lowercase				):
       """simple docstring"""


       def        lowercase__							(						self						):






         """simple docstring"""





         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:
           lowerCAmelCase     :     List[str]								=							Elasticsearch()
           lowerCAmelCase     :     Dict								=							{"acknowledged": True}
           lowerCAmelCase     :     Optional[int]								=							ElasticSearchIndex(es_client=snake_case__						)
           mocked_bulk.return_value([(True, None)] * 3						)
           index.add_documents(["foo", "bar", "foobar"]						)

           # single query
           lowerCAmelCase     :     List[str]								=							"foo"
           lowerCAmelCase     :     List[str]								=							{"hits": {"hits": [{"_score": 1, "_id": 0}]}}
           lowerCAmelCase						,  lowerCAmelCase     :     Optional[int]								=							index.search(snake_case__						)
           self.assertEqual(scores[0]		,     1						)
           self.assertEqual(indices[0]		,     0						)

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

           # batched queries
           lowerCAmelCase     :     Any								=							["foo", "bar", "foobar"]
           lowerCAmelCase     :     Optional[int]								=							{"hits": {"hits": [{"_score": 1, "_id": 1}]}}
           lowerCAmelCase						,  lowerCAmelCase     :     Any								=							index.search_batch(snake_case__						)
           lowerCAmelCase     :     Tuple								=							[scores[0] for scores in total_scores]
           lowerCAmelCase     :     Dict								=							[indices[0] for indices in total_indices]
           self.assertGreater(np.min(snake_case__						)		,     0						)
           self.assertListEqual([1, 1, 1]		,     snake_case__						)

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





